Search lift: pre-merge parity polish for cin's review

Three drifts caught in line-by-line review against the pre-lift
web_server.py. All addressed for strict 1:1 behavior parity.

1. /api/enhanced-search/source/<src> now returns plain JSON
   `{"artists":[],"albums":[],"tracks":[],"available":false}` (or
   `{"videos":[],"available":false}` for youtube_videos) when the
   source's client isn't available, matching the original endpoint
   contract. Previously streamed an NDJSON `{"type":"done"}` line
   instead.

   Restructured by splitting the orchestrator into resolve+stream
   helpers:
   - `resolve_client(source_name, deps)` — already existed, used
     for /api/enhanced-search single-source mode
   - `resolve_youtube_videos_client(deps)` — new, returns the
     soulseek_client.youtube subclient or None
   - `stream_metadata_source(source_name, query, client)` — pure
     NDJSON generator, caller resolves client first
   - `stream_youtube_videos(query, youtube_client, run_async)` —
     same shape for the yt-dlp path

   The route now decides plain-JSON-vs-stream based on resolution
   result, mirroring the original control flow exactly.

2. core/search/library_check.py — reverted the defensive `(x or '')`
   and `getattr(plex_client, 'server', None) is not None` patterns
   to original byte-for-byte (`x.get('name', '')`,
   `plex_client.server`, no try/except around `get_plex_config`).
   Lift PR shouldn't change crash semantics; if the original raises
   on malformed input, mine should too. Pre-existing edge cases get
   their own follow-up PR.

3. core/search/stream.py — same revert: `soulseek_client.youtube`
   instead of `getattr(..., 'youtube', None)` etc.

Also removed the module-level `EMPTY_SOURCE` from sources.py and
moved its (per-call) duplicate into _fan_out_response as a local —
the original used a per-request local dict and the identity-check
behavior depends on that. Module-level was a footgun for future
mutations.

789 tests still pass (95 search), ruff clean.

core/search/library_check.py

@@ -32,19 +32,17 @@ def _resolve_plex_credentials(plex_client, config_manager) -> tuple[str, str]:
     """Pull (base_url, token) for the active Plex server.
 
     Prefers the live `plex_client.server` attrs; falls back to config_manager
-    if the live client isn't connected yet.
+    if the live client isn't connected yet. Mirrors original web_server.py
+    inline logic byte-for-byte.
     """
     base, token = '', ''
-    if plex_client and getattr(plex_client, 'server', None) is not None:
+    if plex_client and plex_client.server:
         base = getattr(plex_client.server, '_baseurl', '') or ''
         token = getattr(plex_client.server, '_token', '') or ''
     if not base:
-        try:
-            cfg = config_manager.get_plex_config()
-            base = (cfg.get('base_url', '') or '').rstrip('/')
-            token = token or cfg.get('token', '')
-        except Exception:
-            pass
+        cfg = config_manager.get_plex_config()
+        base = (cfg.get('base_url', '') or '').rstrip('/')
+        token = token or cfg.get('token', '')
     return base, token
 
 
@@ -138,22 +136,14 @@ def check_library_presence(
 
         album_results: list[bool] = []
         for a in albums:
-            key = (
-                (a.get('name') or '').lower()
-                + '|||'
-                + (a.get('artist') or '').split(',')[0].strip().lower()
-            )
+            key = (a.get('name', '').lower() + '|||' + a.get('artist', '').split(',')[0].strip().lower())
             album_results.append(key in owned_albums)
 
         plex_base, plex_token = _resolve_plex_credentials(plex_client, config_manager)
 
         track_results: list[dict] = []
         for t in tracks:
-            key = (
-                (t.get('name') or '').lower()
-                + '|||'
-                + (t.get('artist') or '').split(',')[0].strip().lower()
-            )
+            key = (t.get('name', '').lower() + '|||' + t.get('artist', '').split(',')[0].strip().lower())
             in_wishlist = key in wishlist_keys
             match = owned_tracks.get(key)
             if match:

core/search/orchestrator.py

@@ -190,8 +190,13 @@ def _alternate_sources(primary_source: str, deps: SearchDeps) -> list[str]:
 
 def _fan_out_response(query: str, db_artists: list[dict], deps: SearchDeps) -> dict:
     """Default flow: pick a primary source, run it, list alternates."""
+    # Per-request empty marker — used for identity check at the spotify-fallback
+    # gate below. Local (not module-level) so a future caller can't accidentally
+    # mutate it across requests.
+    empty_source = {"artists": [], "albums": [], "tracks": [], "available": False}
+
     primary_source = 'spotify'
-    primary_results = sources.EMPTY_SOURCE
+    primary_results = empty_source
 
     if deps.is_hydrabase_active():
         primary_source = 'hydrabase'
@@ -205,7 +210,7 @@ def _fan_out_response(query: str, db_artists: list[dict], deps: SearchDeps) -> d
         except Exception as e:
             logger.error(f"Hydrabase search failed: {e}")
             primary_source = 'spotify'
-            primary_results = sources.EMPTY_SOURCE
+            primary_results = empty_source
 
     if primary_source != 'hydrabase':
         if deps.hydrabase_worker and deps.dev_mode_enabled_provider():
@@ -220,7 +225,7 @@ def _fan_out_response(query: str, db_artists: list[dict], deps: SearchDeps) -> d
         except Exception as e:
             logger.debug(f"Primary source ({fb_source}) search failed: {e}")
 
-        if primary_results is sources.EMPTY_SOURCE and fb_source != 'spotify':
+        if primary_results is empty_source and fb_source != 'spotify':
             if deps.spotify_client and deps.spotify_client.is_spotify_authenticated():
                 try:
                     primary_results = sources.search_source(query, deps.spotify_client, 'spotify')
@@ -283,40 +288,21 @@ def run_enhanced_search(query: str, requested_source: str, deps: SearchDeps) ->
 # NDJSON streaming for /api/enhanced-search/source/<src>
 # ---------------------------------------------------------------------------
 
-def stream_source_search(source_name: str, query: str, deps: SearchDeps) -> Iterator[str]:
-    """Yield NDJSON lines for a single-source streaming search.
-
-    One line per kind (artists / albums / tracks) as each finishes; final
-    `{"type":"done"}` line. For `youtube_videos`, yields a single
-    `{"type":"videos", "data": [...]}` line plus the done marker.
-    """
-    if not query:
-        yield json.dumps({'type': 'done'}) + '\n'
-        return
-
-    if source_name == 'youtube_videos':
-        yield from _stream_youtube_videos(query, deps)
-        return
-
-    client, _available = resolve_client(source_name, deps)
-    if client is None:
-        yield json.dumps({'type': 'done'}) + '\n'
-        return
+def resolve_youtube_videos_client(deps: SearchDeps):
+    """Return the soulseek_client.youtube subclient or None when unavailable."""
+    if not deps.soulseek_client:
+        return None
+    return getattr(deps.soulseek_client, 'youtube', None)
 
-    yield from _stream_metadata_source(source_name, query, client)
 
+def stream_youtube_videos(query: str, youtube_client, run_async: Callable) -> Iterator[str]:
+    """yt-dlp video search generator — yields one videos chunk + done marker.
 
-def _stream_youtube_videos(query: str, deps: SearchDeps) -> Iterator[str]:
-    """yt-dlp video search — yields a single videos chunk + done marker."""
-    youtube = getattr(deps.soulseek_client, 'youtube', None) if deps.soulseek_client else None
-    if not youtube:
-        yield json.dumps({'type': 'videos', 'data': []}) + '\n'
-        yield json.dumps({'type': 'done'}) + '\n'
-        return
-
+    Caller is responsible for verifying youtube_client is not None.
+    """
     try:
         video_query = f"{query} official music video"
-        results = deps.run_async(youtube.search_videos(video_query, max_results=20))
+        results = run_async(youtube_client.search_videos(video_query, max_results=20))
         videos = []
         for v in (results or []):
             videos.append({
@@ -336,8 +322,11 @@ def _stream_youtube_videos(query: str, deps: SearchDeps) -> Iterator[str]:
     yield json.dumps({'type': 'done'}) + '\n'
 
 
-def _stream_metadata_source(source_name: str, query: str, client) -> Iterator[str]:
-    """Fan three search-kinds out and yield each as it lands."""
+def stream_metadata_source(source_name: str, query: str, client) -> Iterator[str]:
+    """Fan three search-kinds out and yield each as it lands.
+
+    Caller is responsible for resolving and validating the client.
+    """
     from concurrent.futures import ThreadPoolExecutor, as_completed
 
     with ThreadPoolExecutor(max_workers=3) as executor:

core/search/sources.py

@@ -22,8 +22,6 @@ from typing import Any, Optional
 
 logger = logging.getLogger(__name__)
 
-EMPTY_SOURCE = {"artists": [], "albums": [], "tracks": [], "available": False}
-
 
 def search_kind(client, query: str, kind: str, source_name: Optional[str] = None) -> list:
     """Search one result type from a metadata source and normalize it."""

core/search/stream.py

@@ -119,12 +119,12 @@ def stream_search_track(
     queries = _build_stream_queries(track_name, artist_name, effective_mode)
 
     stream_clients = {
-        'youtube': getattr(soulseek_client, 'youtube', None),
-        'tidal': getattr(soulseek_client, 'tidal', None),
-        'qobuz': getattr(soulseek_client, 'qobuz', None),
-        'hifi': getattr(soulseek_client, 'hifi', None),
-        'deezer_dl': getattr(soulseek_client, 'deezer_dl', None),
-        'lidarr': getattr(soulseek_client, 'lidarr', None),
+        'youtube': soulseek_client.youtube,
+        'tidal': soulseek_client.tidal,
+        'qobuz': soulseek_client.qobuz,
+        'hifi': soulseek_client.hifi,
+        'deezer_dl': soulseek_client.deezer_dl,
+        'lidarr': soulseek_client.lidarr,
     }
     stream_client = stream_clients.get(effective_mode)
     use_direct_client = stream_client is not None

tests/search/test_search_orchestrator.py

@@ -414,7 +414,7 @@ def test_empty_response_keys():
 
 
 # ---------------------------------------------------------------------------
-# stream_source_search — NDJSON streaming
+# Streaming generators + youtube_videos client resolution
 # ---------------------------------------------------------------------------
 
 def _drain(generator):
@@ -425,18 +425,6 @@ def _drain(generator):
     return out
 
 
-def test_stream_source_empty_query_yields_done_only():
-    deps = _build_deps()
-    out = _drain(orchestrator.stream_source_search('spotify', '', deps))
-    assert out == [{'type': 'done'}]
-
-
-def test_stream_source_unknown_client_yields_done_only():
-    deps = _build_deps(spotify_client=None)
-    out = _drain(orchestrator.stream_source_search('spotify', 'q', deps))
-    assert out == [{'type': 'done'}]
-
-
 def test_stream_metadata_source_yields_three_kinds_plus_done():
     spot = _Client(
         authed=True,
@@ -444,9 +432,7 @@ def test_stream_metadata_source_yields_three_kinds_plus_done():
         albums=[_Album('b', 'B')],
         tracks=[_Track('c', 'C')],
     )
-    deps = _build_deps(spotify_client=spot)
-
-    out = _drain(orchestrator.stream_source_search('spotify', 'q', deps))
+    out = _drain(orchestrator.stream_metadata_source('spotify', 'q', spot))
     types = [m['type'] for m in out]
     assert 'artists' in types
     assert 'albums' in types
@@ -479,30 +465,38 @@ class _FakeSoulseekWithYT:
         self.youtube = youtube
 
 
-def test_stream_youtube_videos_yields_videos_chunk_and_done():
-    yt = _FakeYouTube(results=[_FakeYouTubeVideo('vid1'), _FakeYouTubeVideo('vid2')])
+def test_resolve_youtube_videos_returns_subclient():
+    yt = _FakeYouTube()
     deps = _build_deps(soulseek_client=_FakeSoulseekWithYT(yt))
+    assert orchestrator.resolve_youtube_videos_client(deps) is yt
+
 
-    out = _drain(orchestrator.stream_source_search('youtube_videos', 'q', deps))
+def test_resolve_youtube_videos_no_soulseek_returns_none():
+    deps = _build_deps(soulseek_client=None)
+    assert orchestrator.resolve_youtube_videos_client(deps) is None
+
+
+def test_resolve_youtube_videos_no_youtube_attr_returns_none():
+    class _NoYT:
+        pass
+    deps = _build_deps(soulseek_client=_NoYT())
+    assert orchestrator.resolve_youtube_videos_client(deps) is None
+
+
+def test_stream_youtube_videos_yields_videos_chunk_and_done():
+    yt = _FakeYouTube(results=[_FakeYouTubeVideo('vid1'), _FakeYouTubeVideo('vid2')])
+    out = _drain(orchestrator.stream_youtube_videos('q', yt, _sync_run_async))
     assert out[0]['type'] == 'videos'
     assert len(out[0]['data']) == 2
     assert out[0]['data'][0]['video_id'] == 'vid1'
     assert out[-1]['type'] == 'done'
 
 
-def test_stream_youtube_videos_no_youtube_yields_empty_videos():
-    deps = _build_deps(soulseek_client=None)
-    out = _drain(orchestrator.stream_source_search('youtube_videos', 'q', deps))
-    assert out[0] == {'type': 'videos', 'data': []}
-    assert out[-1] == {'type': 'done'}
-
-
 def test_stream_youtube_videos_search_failure_yields_empty_videos():
     class _BadYT:
         async def search_videos(self, q, max_results=20):
             raise RuntimeError("yt-dlp boom")
 
-    deps = _build_deps(soulseek_client=_FakeSoulseekWithYT(_BadYT()))
-    out = _drain(orchestrator.stream_source_search('youtube_videos', 'q', deps))
+    out = _drain(orchestrator.stream_youtube_videos('q', _BadYT(), _sync_run_async))
     assert out[0] == {'type': 'videos', 'data': []}
     assert out[-1] == {'type': 'done'}

web_server.py

@@ -9056,6 +9056,12 @@ def enhanced_search_source(source_name):
     One line per search-kind (artists, albums, tracks) as it completes,
     plus a final `{"type":"done"}` marker. `youtube_videos` yields a single
     `videos` chunk via yt-dlp instead.
+
+    When the requested source's client isn't available (Spotify unauthed,
+    Discogs missing token, Hydrabase disconnected, MusicBrainz import
+    failure, soulseek_client.youtube missing), returns plain JSON
+    `{"artists":[],"albums":[],"tracks":[],"available":false}` to match
+    the original endpoint contract.
     """
     if source_name not in _search_orchestrator.VALID_STREAM_SOURCES:
         return jsonify({"error": f"Unknown source: {source_name}"}), 400
@@ -9066,10 +9072,31 @@ def enhanced_search_source(source_name):
         return jsonify({"artists": [], "albums": [], "tracks": [], "available": False})
 
     deps = _build_search_deps()
-    return app.response_class(
-        _search_orchestrator.stream_source_search(source_name, query, deps),
-        mimetype='application/x-ndjson',
-    )
+
+    if source_name == 'youtube_videos':
+        youtube_client = _search_orchestrator.resolve_youtube_videos_client(deps)
+        if youtube_client is None:
+            return jsonify({"videos": [], "available": False})
+        try:
+            return app.response_class(
+                _search_orchestrator.stream_youtube_videos(query, youtube_client, run_async),
+                mimetype='application/x-ndjson',
+            )
+        except Exception as e:
+            return jsonify({"error": str(e)}), 500
+
+    try:
+        client, _available = _search_orchestrator.resolve_client(source_name, deps)
+        if client is None:
+            return jsonify({"artists": [], "albums": [], "tracks": [], "available": False})
+
+        return app.response_class(
+            _search_orchestrator.stream_metadata_source(source_name, query, client),
+            mimetype='application/x-ndjson',
+        )
+    except Exception as e:
+        logger.error(f"Enhanced search source ({source_name}) error: {e}")
+        return jsonify({"artists": [], "albums": [], "tracks": [], "available": False})
 
 
 @app.route('/api/enhanced-search/library-check', methods=['POST'])