manual search: INJECT the exact pasted-link track, don't rely on text search surfacing it (#932)

reopened by diegocade1: pasting a Qobuz track link still showed unrelated results. the earlier
fix (b1f061a) only BUBBLED the linked track to the top — but a pasted link is resolved to an
"artist title" text query and searched, and for an obscure track ("foreign lavennew" by colacola)
that text search returns broad lookalikes ("Foreign Bird", "Foreign Spies", …) and never the
actual track. nothing to bubble → user sees junk.

fix: since the link is already resolved via get_track(id), fetch that exact track AS a downloadable
result and inject it at the top (Qobuz downloads by id, so the result is fully usable). the text
search still runs for alternatives.

- QobuzClient.get_track_result(id): get_track + _qobuz_to_track_result; None on any failure.
- _qobuz_to_track_result gains require_streamable (default True for bulk search). the link fetch
  passes False: track/get may OMIT the streamable flag, which would default-False and wrongly drop
  the exact track the user explicitly asked for. (this closes the one shape assumption that
  couldn't be verified against a live Qobuz API — the track is no longer gated on it.)
- track_link.inject_linked_track_first(tracks, linked_result, id): pure seam — prepend the fetched
  result + drop any search duplicate; falls back to the bubble when no result was fetched.
- manual-search endpoint fetches linked_result defensively (getattr 'get_track_result') and calls
  the seam. Tidal/HiFi (get_track returns a dict but the converter wants an object — shape
  mismatch) have no get_track_result, so they keep the existing bubble path: NO regression.

14 tests: inject puts the fetched track first when search missed it / dedups a search copy / falls
back to bubble / str-safe id / noop; get_track_result convert/none/exception; and the REAL
converter builds a valid downloadable result from a track/get dict that OMITS streamable (search
path still rejects it). 85 track-link/qobuz tests green, ruff clean.

core/downloads/track_link.py

@@ -38,6 +38,23 @@ def bubble_linked_track_first(tracks: List[Any], link_track_id: str) -> List[Any
     target = str(link_track_id)
     return sorted(tracks, key=lambda t: linked_track_id(t) != target)
 
+
+def inject_linked_track_first(
+    tracks: List[Any], linked_result: Any, link_track_id: str
+) -> List[Any]:
+    """Put the EXACT linked track first.
+
+    When ``linked_result`` is the track fetched directly by id, prepend it and
+    drop any search duplicate of it — so an obscure track a text search never
+    surfaced is still present and downloadable (#932). When it's None (the source
+    can't fetch one), fall back to bubbling a matching search result. Pure."""
+    if not link_track_id:
+        return tracks
+    target = str(link_track_id)
+    if linked_result is not None:
+        return [linked_result] + [t for t in tracks if linked_track_id(t) != target]
+    return bubble_linked_track_first(tracks, target)
+
 # host substring → download source id. Only ID-downloadable streaming sources.
 _HOSTS = (
     ('tidal.com', 'tidal'),

core/qobuz_client.py

@@ -715,6 +715,26 @@ class QobuzClient(DownloadSourcePlugin):
             logger.error(f"Error getting Qobuz track {track_id}: {e}")
             return None
 
+    def get_track_result(self, track_id):
+        """Fetch ONE track by ID and convert it to a downloadable TrackResult.
+
+        Used when a pasted Qobuz link must inject the EXACT track: a text search
+        for an obscure track's name often doesn't surface it at all, so bubbling
+        the matching id is a no-op (#932). Returns None on any failure so the
+        caller falls back to the text-search path."""
+        try:
+            track = self.get_track(track_id)
+            if not track:
+                return None
+            quality_key = quality_tier_for_source('qobuz', default='lossless')
+            quality_info = QOBUZ_QUALITY_MAP.get(quality_key, QOBUZ_QUALITY_MAP['lossless'])
+            # require_streamable=False: this is the exact track the user linked — don't
+            # let a missing/absent 'streamable' flag in track/get drop it (#932).
+            return self._qobuz_to_track_result(track, quality_info, require_streamable=False)
+        except Exception as e:
+            logger.debug(f"get_track_result failed for Qobuz {track_id}: {e}")
+            return None
+
     # ===================== Playlists & Favorites =====================
     #
     # Qobuz playlist sync surface — mirrors the Tidal client contract
@@ -1012,14 +1032,20 @@ class QobuzClient(DownloadSourcePlugin):
             traceback.print_exc()
             return ([], [])
 
-    def _qobuz_to_track_result(self, track: Dict, quality_info: dict) -> Optional[TrackResult]:
-        """Convert Qobuz track dict to TrackResult (Soulseek-compatible format)."""
+    def _qobuz_to_track_result(self, track: Dict, quality_info: dict,
+                               require_streamable: bool = True) -> Optional[TrackResult]:
+        """Convert Qobuz track dict to TrackResult (Soulseek-compatible format).
+
+        ``require_streamable`` gates bulk SEARCH results on the ``streamable`` flag
+        (filters noise). For a track the user explicitly pasted a link to, pass
+        False: ``track/get`` may omit the flag (which would default-False and wrongly
+        drop the exact track they asked for), and they should get to try it (#932)."""
         track_id = track.get('id')
         if not track_id:
             return None
 
-        # Check if track is streamable
-        if not track.get('streamable', False):
+        # Check if track is streamable (skipped for explicit by-id link fetches).
+        if require_streamable and not track.get('streamable', False):
             return None
 
         performer = track.get('performer', {})

tests/downloads/test_track_link.py

@@ -122,3 +122,112 @@ def test_bubble_noop_when_nothing_matches_or_empty():
     assert bubble_linked_track_first([a, b], '296427754') == [a, b]   # unchanged
     assert bubble_linked_track_first([], '296427754') == []
     assert bubble_linked_track_first([a, b], '') == [a, b]
+
+
+# ── inject the EXACT fetched track first (#932 reopen: text search misses obscure tracks) ──
+
+from core.downloads.track_link import inject_linked_track_first
+
+
+def test_inject_puts_fetched_track_first_when_search_missed_it():
+    # The reported case: the linked track isn't among the fuzzy text-search
+    # results at all, so the directly-fetched result is injected at the top.
+    fetched = _result('296427754')
+    search = [_result('111'), _result('222')]   # unrelated lookalikes
+    out = inject_linked_track_first(search, fetched, '296427754')
+    assert out[0] is fetched
+    assert len(out) == 3
+
+
+def test_inject_dedups_a_search_copy_of_the_linked_track():
+    fetched = _result('296427754')
+    dup = _result('296427754')                  # search also returned it
+    search = [_result('111'), dup, _result('222')]
+    out = inject_linked_track_first(search, fetched, '296427754')
+    assert out[0] is fetched
+    # exactly one element carries the linked id, and it's the injected one
+    assert [linked_track_id(t) for t in out].count('296427754') == 1
+    assert all(t is not dup for t in out)        # the search copy (by identity) was dropped
+    assert len(out) == 3
+
+
+def test_inject_falls_back_to_bubble_without_a_fetched_result():
+    exact = _result('296427754')
+    out = inject_linked_track_first([_result('111'), exact, _result('222')], None, '296427754')
+    assert linked_track_id(out[0]) == '296427754'   # bubbled, not injected
+    assert len(out) == 3
+
+
+def test_inject_is_noop_without_a_link_id():
+    search = [_result('111')]
+    assert inject_linked_track_first(search, _result('x'), '') == search
+
+
+def test_inject_int_track_id_is_str_safe():
+    fetched = _result('296427754')
+    out = inject_linked_track_first([_result('111')], fetched, 296427754)
+    assert out[0] is fetched
+
+
+# ── QobuzClient.get_track_result: fetch by id → downloadable TrackResult ──
+
+from core.qobuz_client import QobuzClient
+
+
+def _bare_qobuz():
+    return QobuzClient.__new__(QobuzClient)   # no network / __init__
+
+
+def test_get_track_result_converts_fetched_track(monkeypatch):
+    client = _bare_qobuz()
+    sentinel = object()
+    monkeypatch.setattr(client, 'get_track', lambda tid: {'id': tid, 'streamable': True})
+    monkeypatch.setattr(client, '_qobuz_to_track_result',
+                        lambda track, qi, require_streamable=True: sentinel)
+    assert client.get_track_result('296427754') is sentinel
+
+
+def test_get_track_result_none_when_track_unavailable(monkeypatch):
+    client = _bare_qobuz()
+    monkeypatch.setattr(client, 'get_track', lambda tid: None)
+    assert client.get_track_result('nope') is None
+
+
+def test_get_track_result_none_on_exception(monkeypatch):
+    client = _bare_qobuz()
+    def _boom(_tid):
+        raise RuntimeError('api down')
+    monkeypatch.setattr(client, 'get_track', _boom)
+    assert client.get_track_result('x') is None
+
+
+# ── #932 hardening: a pasted-link fetch must not be dropped by a missing
+#    'streamable' flag (track/get may omit it). Exercises the REAL converter. ──
+
+_QOBUZ_TRACK = {'id': 296427754, 'title': 'foreign lavennew',
+                'performer': {'name': 'colacola'}, 'duration': 180}
+
+
+def test_converter_rejects_non_streamable_on_search_path():
+    # Default (bulk search) still filters on streamable — unchanged behaviour.
+    client = _bare_qobuz()
+    qi = {'codec': 'flac', 'bitrate': 1411}
+    assert client._qobuz_to_track_result(dict(_QOBUZ_TRACK), qi) is None  # no streamable flag
+
+
+def test_converter_builds_for_link_fetch_without_streamable():
+    client = _bare_qobuz()
+    qi = {'codec': 'flac', 'bitrate': 1411}
+    r = client._qobuz_to_track_result(dict(_QOBUZ_TRACK), qi, require_streamable=False)
+    assert r is not None
+    assert linked_track_id(r) == '296427754'   # carries the id → injectable + downloadable
+
+
+def test_get_track_result_survives_missing_streamable_flag(monkeypatch):
+    # The feared track/get shape (no 'streamable') must STILL yield the track —
+    # this is the end-to-end gap that couldn't be confirmed against a live API.
+    client = _bare_qobuz()
+    monkeypatch.setattr(client, 'get_track', lambda _tid: dict(_QOBUZ_TRACK))
+    r = client.get_track_result('296427754')
+    assert r is not None
+    assert linked_track_id(r) == '296427754'

web_server.py

@@ -7506,6 +7506,7 @@ def manual_search_for_task(task_id):
         link = parse_download_track_link(query)
         link_source = None
         link_track_id = None
+        linked_result = None  # the EXACT track fetched by id, injected into results
         # A pasted SoundCloud link can't be turned into an "artist title" query
         # and searched — unlisted/private tracks aren't searchable. Instead force
         # the SoundCloud source and keep the URL as the query; the SoundCloud
@@ -7535,6 +7536,18 @@ def manual_search_for_task(task_id):
             query = clean_q
             source = _src
             link_source, link_track_id = _src, _tid
+            # Fetch the EXACT linked track as a downloadable result to inject —
+            # a text search for an obscure track's name often doesn't surface it
+            # at all, so we can't rely on it being in the search results (#932).
+            # Defensive: only sources that expose get_track_result (Qobuz today);
+            # others fall back to the bubble path below — never worse than before.
+            _link_client = download_orchestrator.client(_src) if download_orchestrator else None
+            if _link_client is not None and hasattr(_link_client, 'get_track_result'):
+                try:
+                    linked_result = _link_client.get_track_result(_tid)
+                except Exception as _lr_err:
+                    logger.debug("[Manual Search] get_track_result failed for %s %s: %s",
+                                 _src, _tid, _lr_err)
 
         if source != 'all':
             if source not in valid_source_ids:
@@ -7595,13 +7608,15 @@ def manual_search_for_task(task_id):
                             "error": error,
                         }) + "\n"
                         continue
-                    # Pasted-link exact match: bubble the track whose source id
-                    # matches the link to the top so the user sees the exact version
-                    # first. Reads _source_metadata['track_id'] (TrackResult has no
-                    # top-level id) — the old getattr(t,'id') always missed (#932).
-                    if src_name == link_source and link_track_id and tracks:
-                        from core.downloads.track_link import bubble_linked_track_first
-                        tracks = bubble_linked_track_first(tracks, link_track_id)
+                    # Pasted-link exact match (#932): the linked source's search may
+                    # not surface an obscure linked track at all, so INJECT the exact
+                    # track (fetched by id) at the top and drop any search duplicate of
+                    # it. If we couldn't fetch it (source has no get_track_result, or it
+                    # failed), fall back to bubbling a matching search result — never
+                    # worse than before.
+                    if src_name == link_source and link_track_id:
+                        from core.downloads.track_link import inject_linked_track_first
+                        tracks = inject_linked_track_first(tracks, linked_result, link_track_id)
                     serialized = []
                     for t in tracks:
                         s = _serialize_candidate(t, source_override=src_name)