F: Engine owns hybrid fallback chain

`engine.search_with_fallback(query, source_chain, ...)` walks the
chain in order, skips unconfigured / unregistered plugins,
swallows per-source exceptions, and returns the first non-empty
(tracks, albums) tuple. Replaces orchestrator's hand-rolled
hybrid search loop.

`engine.download_with_fallback(username, filename, file_size,
source_chain)` falls through the chain when a source returns
None / raises. Username hint promotes a matching source-chain
entry to head of order. NOT yet wired into orchestrator.download
— today's username comes from a search result and represents
the user's explicit source pick, so silently falling through
would override their choice. Engine method is available for
future callers that want fallback semantics
(search_and_download_best, automation).

Orchestrator gains _resolve_source_chain helper that builds
the ordered list (hybrid_order config, falling back to legacy
primary/secondary pair). Orchestrator.search hands chain off
to engine.search_with_fallback for hybrid mode.

8 new tests pin the fallback semantics: chain ordering,
unconfigured-skip, exception-continue, empty-when-exhausted,
username-hint promotion. Suite still green (2050 passed).

core/download_engine/engine.py

@@ -302,3 +302,80 @@ class DownloadEngine:
                 logger.warning("%s clear_all_completed_downloads failed: %s", source_name, exc)
                 results.append(False)
         return all(results) if results else True
+
+    # ------------------------------------------------------------------
+    # Hybrid fallback — Phase F surface
+    # ------------------------------------------------------------------
+
+    async def search_with_fallback(self, query: str, source_chain,
+                                   timeout=None, progress_callback=None):
+        """Try each source in ``source_chain`` until one returns
+        tracks. Skips unconfigured / unregistered sources, swallows
+        per-source exceptions. Returns the first non-empty
+        (tracks, albums) tuple, or ``([], [])`` when every source
+        in the chain is exhausted.
+
+        Replaces orchestrator's hand-rolled hybrid search loop. The
+        chain is ordered (most-preferred first).
+        """
+        for i, source_name in enumerate(source_chain):
+            plugin = self._plugins.get(source_name)
+            if plugin is None:
+                logger.info(f"Skipping {source_name} (not available)")
+                continue
+            if hasattr(plugin, 'is_configured') and not plugin.is_configured():
+                logger.info(f"Skipping {source_name} (not configured)")
+                continue
+
+            try:
+                logger.info(f"Trying {source_name} (priority {i+1}): {query}")
+                tracks, albums = await plugin.search(query, timeout, progress_callback)
+                if tracks:
+                    logger.info(f"{source_name} found {len(tracks)} tracks")
+                    return (tracks, albums)
+            except Exception as e:
+                logger.warning(f"{source_name} search failed: {e}")
+
+        logger.warning(
+            "Hybrid search: all sources (%s) found nothing for: %s",
+            ', '.join(source_chain), query,
+        )
+        return ([], [])
+
+    async def download_with_fallback(self, username: str, filename: str,
+                                     file_size: int, source_chain) -> Optional[str]:
+        """Try each source in ``source_chain`` until one accepts the
+        download (returns a non-None download_id). Fixes the legacy
+        bug where hybrid mode silently routed to a single source via
+        the username hint with no retry on failure.
+
+        ``username`` is treated as a hint when it matches a source
+        name in the chain — that source is tried FIRST regardless of
+        chain order. Anything else (e.g. a real Soulseek peer name)
+        routes through the chain in declared order.
+        """
+        # Promote a matching source-name hint to the head of the chain.
+        ordered_chain = list(source_chain)
+        if username and username in ordered_chain:
+            ordered_chain.remove(username)
+            ordered_chain.insert(0, username)
+
+        for source_name in ordered_chain:
+            plugin = self._plugins.get(source_name)
+            if plugin is None:
+                continue
+            if hasattr(plugin, 'is_configured') and not plugin.is_configured():
+                continue
+            try:
+                download_id = await plugin.download(username, filename, file_size)
+                if download_id is not None:
+                    return download_id
+                logger.info(f"{source_name} declined download — trying next in chain")
+            except Exception as e:
+                logger.warning(f"{source_name} download raised — trying next in chain: {e}")
+
+        logger.warning(
+            "Hybrid download: every source in chain (%s) refused %r",
+            ', '.join(ordered_chain), filename,
+        )
+        return None

core/download_orchestrator.py

@@ -196,18 +196,26 @@ class DownloadOrchestrator:
 
         return False
 
+    def _resolve_source_chain(self) -> List[str]:
+        """Order the configured sources for hybrid mode. Prefers
+        ``hybrid_order`` config; falls back to legacy
+        primary/secondary pair when no order set."""
+        registry_names = set(self.registry.names())
+        if self.hybrid_order:
+            return [s for s in self.hybrid_order if s in registry_names]
+        primary = self.hybrid_primary if self.hybrid_primary in registry_names else 'soulseek'
+        secondary = self.hybrid_secondary if self.hybrid_secondary in registry_names else 'soulseek'
+        if secondary == primary:
+            secondary = next((name for name in registry_names if name != primary), 'soulseek')
+        chain = [primary, secondary]
+        if not chain:
+            chain = ['soulseek']
+        return chain
+
     async def search(self, query: str, timeout: int = None, progress_callback=None) -> Tuple[List[TrackResult], List[AlbumResult]]:
-        """
+        """Search for tracks using configured source(s). Single-source
-        Search for tracks using configured source(s).
+        modes route directly; hybrid mode delegates to
-
+        ``engine.search_with_fallback`` which tries the chain in order."""
-        Args:
-            query: Search query
-            timeout: Search timeout (for Soulseek)
-            progress_callback: Progress callback (for Soulseek)
-
-        Returns:
-            Tuple of (track_results, album_results)
-        """
         if self.mode != 'hybrid':
             client = self._client(self.mode)
             if not client:
@@ -216,53 +224,9 @@ class DownloadOrchestrator:
             logger.info(f"Searching {self.registry.display_name(self.mode)}: {query}")
             return await client.search(query, timeout, progress_callback)
 
-        elif self.mode == 'hybrid':
+        chain = self._resolve_source_chain()
-            clients = {name: self._client(name) for name in self.registry.names()}
+        logger.info(f"Hybrid search ({' → '.join(chain)}): {query}")
-
+        return await self.engine.search_with_fallback(query, chain, timeout, progress_callback)
-            # Build ordered source list: prefer hybrid_order, fall back to legacy primary/secondary
-            if self.hybrid_order:
-                source_order = [s for s in self.hybrid_order if s in clients]
-            else:
-                primary = self.hybrid_primary if self.hybrid_primary in clients else 'soulseek'
-                secondary = self.hybrid_secondary if self.hybrid_secondary in clients else 'soulseek'
-                if secondary == primary:
-                    secondary = next((name for name in clients if name != primary), 'soulseek')
-                source_order = [primary, secondary]
-
-            if not source_order:
-                source_order = ['soulseek']
-
-            logger.info(f"Hybrid search ({' → '.join(source_order)}): {query}")
-
-            # Try each source in priority order (skip unconfigured/unavailable ones)
-            for i, source_name in enumerate(source_order):
-                client = clients.get(source_name)
-                if not client:
-                    logger.info(f"Skipping {source_name} (not available)")
-                    continue
-                if hasattr(client, 'is_configured') and not client.is_configured():
-                    logger.info(f"Skipping {source_name} (not configured)")
-                    continue
-
-                try:
-                    if i == 0:
-                        logger.info(f"Trying {source_name} (priority {i+1}): {query}")
-                    else:
-                        logger.info(f"Trying {source_name} (priority {i+1}): {query}")
-
-                    tracks, albums = await client.search(query, timeout, progress_callback)
-                    if tracks:
-                        logger.info(f"{source_name} found {len(tracks)} tracks")
-                        return (tracks, albums)
-                except Exception as e:
-                    logger.warning(f"{source_name} search failed: {e}")
-
-            # Nothing found from any source
-            logger.warning(f"Hybrid search: all sources ({', '.join(source_order)}) found nothing for: {query}")
-            return ([], [])
-
-        # Fallback: empty results
-        return ([], [])
 
     async def search_and_download_best(self, query: str, expected_track=None) -> Optional[str]:
         """

tests/downloads/test_download_engine.py

@@ -387,3 +387,160 @@ def test_engine_clear_all_returns_false_when_any_configured_plugin_fails():
 
     result = _run_async(engine.clear_all_completed_downloads())
     assert result is False
+
+
+# ---------------------------------------------------------------------------
+# Hybrid fallback (Phase F)
+# ---------------------------------------------------------------------------
+
+
+class _FakeSearchPlugin:
+    def __init__(self, name, configured=True, search_result=None, raises=None):
+        self.name = name
+        self._configured = configured
+        self._search_result = search_result if search_result is not None else ([], [])
+        self._raises = raises
+        self.search_calls = 0
+
+    def is_configured(self):
+        return self._configured
+
+    async def search(self, query, timeout=None, progress_callback=None):
+        self.search_calls += 1
+        if self._raises:
+            raise self._raises
+        return self._search_result
+
+
+class _FakeDownloadPlugin:
+    def __init__(self, name, configured=True, download_result=None, raises=None):
+        self.name = name
+        self._configured = configured
+        self._download_result = download_result
+        self._raises = raises
+        self.download_calls = []
+
+    def is_configured(self):
+        return self._configured
+
+    async def download(self, username, filename, file_size):
+        self.download_calls.append((username, filename, file_size))
+        if self._raises:
+            raise self._raises
+        return self._download_result
+
+
+def test_search_with_fallback_returns_first_non_empty_result():
+    engine = DownloadEngine()
+    yt = _FakeSearchPlugin('youtube', search_result=([], []))
+    td = _FakeSearchPlugin('tidal', search_result=(['track1'], []))
+    qz = _FakeSearchPlugin('qobuz', search_result=(['track2'], []))
+    engine.register_plugin('youtube', yt)
+    engine.register_plugin('tidal', td)
+    engine.register_plugin('qobuz', qz)
+
+    tracks, _ = _run_async(engine.search_with_fallback('q', ['youtube', 'tidal', 'qobuz']))
+    assert tracks == ['track1']
+    # Tidal short-circuits — qobuz never queried.
+    assert yt.search_calls == 1
+    assert td.search_calls == 1
+    assert qz.search_calls == 0
+
+
+def test_search_with_fallback_skips_unconfigured_plugins():
+    engine = DownloadEngine()
+    yt = _FakeSearchPlugin('youtube', configured=False)
+    td = _FakeSearchPlugin('tidal', configured=True, search_result=(['hit'], []))
+    engine.register_plugin('youtube', yt)
+    engine.register_plugin('tidal', td)
+
+    tracks, _ = _run_async(engine.search_with_fallback('q', ['youtube', 'tidal']))
+    assert tracks == ['hit']
+    assert yt.search_calls == 0  # skipped
+
+
+def test_search_with_fallback_continues_after_per_source_exception():
+    engine = DownloadEngine()
+    yt = _FakeSearchPlugin('youtube', raises=RuntimeError("yt down"))
+    td = _FakeSearchPlugin('tidal', search_result=(['fallback-hit'], []))
+    engine.register_plugin('youtube', yt)
+    engine.register_plugin('tidal', td)
+
+    tracks, _ = _run_async(engine.search_with_fallback('q', ['youtube', 'tidal']))
+    assert tracks == ['fallback-hit']
+
+
+def test_search_with_fallback_returns_empty_when_chain_exhausted():
+    engine = DownloadEngine()
+    yt = _FakeSearchPlugin('youtube', search_result=([], []))
+    td = _FakeSearchPlugin('tidal', search_result=([], []))
+    engine.register_plugin('youtube', yt)
+    engine.register_plugin('tidal', td)
+
+    tracks, _ = _run_async(engine.search_with_fallback('q', ['youtube', 'tidal']))
+    assert tracks == []
+    assert yt.search_calls == 1
+    assert td.search_calls == 1
+
+
+def test_download_with_fallback_returns_first_accepted_download_id():
+    """Phase F bug fix: legacy hybrid download routed to one source
+    via username hint with no retry. Engine now falls through chain."""
+    engine = DownloadEngine()
+    yt = _FakeDownloadPlugin('youtube', download_result=None)  # refuses
+    td = _FakeDownloadPlugin('tidal', download_result='td-id')
+    engine.register_plugin('youtube', yt)
+    engine.register_plugin('tidal', td)
+
+    result = _run_async(engine.download_with_fallback(
+        'youtube', 'v||t', 0, ['youtube', 'tidal'],
+    ))
+    assert result == 'td-id'
+    assert len(yt.download_calls) == 1  # tried first
+    assert len(td.download_calls) == 1  # took over
+
+
+def test_download_with_fallback_promotes_username_hint_to_head():
+    """A username hint that matches a source-chain entry tries that
+    source FIRST regardless of declared chain order."""
+    engine = DownloadEngine()
+    yt = _FakeDownloadPlugin('youtube', download_result='yt-id')
+    td = _FakeDownloadPlugin('tidal', download_result='td-id')
+    engine.register_plugin('youtube', yt)
+    engine.register_plugin('tidal', td)
+
+    # Chain says tidal-first, but username hint promotes youtube.
+    result = _run_async(engine.download_with_fallback(
+        'youtube', 'v||t', 0, ['tidal', 'youtube'],
+    ))
+    assert result == 'yt-id'
+    assert len(yt.download_calls) == 1
+    assert len(td.download_calls) == 0  # never reached
+
+
+def test_download_with_fallback_returns_none_when_all_refuse():
+    engine = DownloadEngine()
+    yt = _FakeDownloadPlugin('youtube', download_result=None)
+    td = _FakeDownloadPlugin('tidal', download_result=None)
+    engine.register_plugin('youtube', yt)
+    engine.register_plugin('tidal', td)
+
+    result = _run_async(engine.download_with_fallback(
+        'youtube', 'v||t', 0, ['youtube', 'tidal'],
+    ))
+    assert result is None
+    assert len(yt.download_calls) == 1
+    assert len(td.download_calls) == 1
+
+
+def test_download_with_fallback_continues_past_exception():
+    engine = DownloadEngine()
+    yt = _FakeDownloadPlugin('youtube', raises=RuntimeError("yt died"))
+    td = _FakeDownloadPlugin('tidal', download_result='td-id')
+    engine.register_plugin('youtube', yt)
+    engine.register_plugin('tidal', td)
+
+    result = _run_async(engine.download_with_fallback(
+        'youtube', 'v||t', 0, ['youtube', 'tidal'],
+    ))
+    assert result == 'td-id'