docs: add Soulseek adapter plan (slice 1, plan 3d)

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-10 22:39:34 +02:00
parent 7510e7718b
commit 9804f3bb90
1 changed files with 629 additions and 0 deletions
@@ -0,0 +1,629 @@
 # Lyra Soulseek Adapter Implementation Plan
 > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
 **Goal:** Add the Soulseek source adapter (P2P, via a running slskd daemon's REST API), implementing the existing `SourceAdapter` contract, gated in the registry on the user's slskd URL + API key. Adapter logic is unit-tested offline against a fake client; the real slskd HTTP path is exercised only by an opt-in live test the user runs against their slskd.
 **Architecture:** `SoulseekAdapter` depends on an injectable `SoulseekClient` seam. Tests inject a `FakeSlskdClient`; the real `SlskdClient` (talks to slskd's REST API with `requests`) is constructed only in `registry.build_adapters(config)` and included only when configured (`health()`). Soulseek search returns files scattered across peers; the real client groups a peer's audio files by directory into per-directory album candidates, infers format (FLAC vs MP3) from extensions, and encodes the peer+file list into the candidate's `source_ref`. Download enqueues those files via slskd and polls until complete. Nothing in the pipeline/ranker/confidence/types changes.
 **Tech Stack:** Python 3.12 worker; `requests` (new dependency, lightweight). Tests: pytest with a fake client (offline) + an opt-in live test gated by `LYRA_LIVE_TESTS` + `SLSKD_URL` + `SLSKD_API_KEY`.
 ## Global Constraints
 - Python 3.12 in the worker container; version-flexible local deps.
 - **New dependency:** `requests` (lightweight, installs on all Python versions). Imported normally at module top (no lazy import needed — importing `requests` does no I/O). No other new deps.
 - The `SourceAdapter` protocol, `Candidate`/`Quality`/`DownloadResult`/`MBTarget` types, ranker, confidence, and pipeline are UNCHANGED.
 - Soulseek: `name="soulseek"`, `tier=1`. Quality: FLAC → `Quality(fmt="FLAC", lossless=True, bit_depth=16, sample_rate=44100)` (CD, quality_class 2); MP3 → `Quality(fmt="MP3", lossless=False, bitrate_kbps=<or 320>)` (quality_class 1). This yields the design order: Qobuz lossless (class 2, tier 0) > Soulseek FLAC (class 2, tier 1) > Soulseek MP3 (class 1, tier 1) > YouTube (class 1, tier 2).
 - `registry.build_adapters(config)` gains `SoulseekAdapter(SlskdClient(config))`, included only when `slskd.url` + `slskd.api_key` are present (via `health()`). **No slskd service is added to docker-compose** — the user points `slskd.url` at their existing slskd; the worker container must be able to reach that URL.
 - The real `SlskdClient` is NOT unit-tested offline; covered only by the opt-in live test (`LYRA_LIVE_TESTS` + `SLSKD_URL` + `SLSKD_API_KEY`). Normal runs never hit the network.
 - **Live-validation note (carry into the live test):** the slskd response/transfer JSON shapes (`isComplete`, `state` strings like "Completed, Succeeded", `responses[].files[]`, `transfers` grouping) follow slskd's documented REST API but exact field names/state strings may differ by slskd version — that is what the live test validates.
 - Soulseek candidates have weak artist/album metadata (parsed from the peer's directory name); like YouTube, they may score below the confidence gate until Plan 4's MusicBrainz intake. Noted, not fixed here.
 - Every task ends with a commit.
 ## Shared interfaces (defined by tasks below)
 ```python
 # worker/lyra_worker/adapters/soulseek.py (Task 1)
 class SoulseekClient(Protocol):
    def is_configured(self) -> bool: ...
    def search_album(self, artist: str, album: str) -> list[dict]: ...   # {source_ref, title, artist, track_count, format, bitrate}
    def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict: ...  # {track_count, path}; raises on failure
 class SoulseekAdapter:   # implements SourceAdapter
    name = "soulseek"; tier = 1
    def __init__(self, client: SoulseekClient): ...
    def health(self) -> bool: ...            # -> client.is_configured()
    def search(self, target) -> list[Candidate]: ...
    def download(self, candidate, dest, on_progress) -> DownloadResult: ...
 # worker/lyra_worker/adapters/_slskd.py (Task 2)
 class SlskdClient:   # real SoulseekClient talking to slskd REST via requests
    def __init__(self, config: dict): ...
    def is_configured(self) -> bool: ...
    def search_album(self, artist, album) -> list[dict]: ...
    def download(self, source_ref, dest, on_progress) -> dict: ...
 # worker/lyra_worker/registry.py (Task 3): build_adapters(config) now also includes Soulseek when configured
 ```
 ---
 ### Task 1: SoulseekAdapter + client seam (offline, fake client)
 **Files:**
 - Create: `worker/lyra_worker/adapters/soulseek.py`
 - Test: `worker/tests/test_soulseek_adapter.py`
 **Interfaces:**
 - Consumes: `MBTarget`, `Candidate`, `Quality`, `DownloadResult`; `SourceAdapter`.
 - Produces: `SoulseekClient` Protocol (with `is_configured`) and `SoulseekAdapter`. `search` maps each result dict to a `Candidate(source="soulseek", source_ref, matched_artist=r["artist"], matched_album=r["title"], quality=<FLAC or MP3 per r["format"]>, track_count, source_tier=1)`. Format→Quality: `"FLAC"` → lossless FLAC 16/44100; anything else → MP3 lossy with `bitrate_kbps=r.get("bitrate") or 320`. `health()` delegates to `client.is_configured()`.
 - [ ] **Step 1: Write the failing test**
 `worker/tests/test_soulseek_adapter.py`:
 ```python
 from lyra_worker.adapters.soulseek import SoulseekAdapter
 from lyra_worker.quality import quality_class
 from lyra_worker.types import MBTarget
 TARGET = MBTarget(artist="John Mayer", album="Continuum", track_count=12)
 class FakeSlskdClient:
    def __init__(self, configured=True, results=None, fail=False):
        self._configured = configured
        self._results = results if results is not None else [
            {"source_ref": "ref-flac", "title": "Continuum", "artist": "John Mayer",
             "track_count": 12, "format": "FLAC", "bitrate": None},
        ]
        self._fail = fail
    def is_configured(self):
        return self._configured
    def search_album(self, artist, album):
        return self._results
    def download(self, source_ref, dest, on_progress):
        if self._fail:
            raise RuntimeError("slskd boom")
        on_progress(1.0)
        return {"track_count": 12, "path": dest}
 def test_name_and_tier():
    a = SoulseekAdapter(FakeSlskdClient())
    assert a.name == "soulseek"
    assert a.tier == 1
 def test_health_reflects_configuration():
    assert SoulseekAdapter(FakeSlskdClient(configured=True)).health() is True
    assert SoulseekAdapter(FakeSlskdClient(configured=False)).health() is False
 def test_flac_maps_to_lossless_candidate():
    c = SoulseekAdapter(FakeSlskdClient()).search(TARGET)[0]
    assert c.source == "soulseek"
    assert c.source_ref == "ref-flac"
    assert c.matched_artist == "John Mayer"
    assert c.matched_album == "Continuum"
    assert c.track_count == 12
    assert c.source_tier == 1
    assert c.quality.fmt == "FLAC"
    assert c.quality.lossless is True
    assert quality_class(c.quality) == 2  # CD lossless
 def test_mp3_maps_to_lossy_candidate():
    results = [{"source_ref": "ref-mp3", "title": "Continuum", "artist": "John Mayer",
                "track_count": 12, "format": "MP3", "bitrate": 320}]
    c = SoulseekAdapter(FakeSlskdClient(results=results)).search(TARGET)[0]
    assert c.quality.fmt == "MP3"
    assert c.quality.lossless is False
    assert c.quality.bitrate_kbps == 320
    assert quality_class(c.quality) == 1  # lossy
 def test_search_empty_and_download_failure():
    assert SoulseekAdapter(FakeSlskdClient(results=[])).search(TARGET) == []
    bad = SoulseekAdapter(FakeSlskdClient(fail=True)).download(
        SoulseekAdapter(FakeSlskdClient()).search(TARGET)[0], "/tmp/x", lambda _p: None)
    assert bad.ok is False and bad.error
 ```
 - [ ] **Step 2: Run the test to verify it fails**
 Run:
 ```bash
 cd worker
 worker/.venv/bin/python -m pytest tests/test_soulseek_adapter.py -v
 ```
 Expected: FAIL — `ModuleNotFoundError: No module named 'lyra_worker.adapters.soulseek'`.
 - [ ] **Step 3: Implement the adapter**
 `worker/lyra_worker/adapters/soulseek.py`:
 ```python
 from typing import Callable, Protocol
 from lyra_worker.types import Candidate, DownloadResult, MBTarget, Quality
 class SoulseekClient(Protocol):
    def is_configured(self) -> bool:
        ...
    def search_album(self, artist: str, album: str) -> list[dict]:
        """Album matches: dicts with source_ref, title, artist, track_count, format, bitrate."""
        ...
    def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
        """Download into dest; return {track_count, path}. Raise on failure."""
        ...
 def _quality(fmt: str, bitrate: int | None) -> Quality:
    if fmt.upper() == "FLAC":
        return Quality(fmt="FLAC", lossless=True, bit_depth=16, sample_rate=44100)
    return Quality(fmt="MP3", lossless=False, bitrate_kbps=bitrate or 320)
 class SoulseekAdapter:
    name = "soulseek"
    tier = 1
    def __init__(self, client: SoulseekClient):
        self._client = client
    def health(self) -> bool:
        return self._client.is_configured()
    def search(self, target: MBTarget) -> list[Candidate]:
        candidates: list[Candidate] = []
        for r in self._client.search_album(target.artist, target.album):
            candidates.append(
                Candidate(
                    source="soulseek",
                    source_ref=r["source_ref"],
                    matched_artist=r.get("artist", ""),
                    matched_album=r.get("title", ""),
                    quality=_quality(r.get("format", "MP3"), r.get("bitrate")),
                    track_count=r.get("track_count", 0) or 0,
                    source_tier=self.tier,
                )
            )
        return candidates
    def download(
        self, candidate: Candidate, dest: str, on_progress: Callable[[float], None]
    ) -> DownloadResult:
        try:
            info = self._client.download(candidate.source_ref, dest, on_progress)
        except Exception as e:
            return DownloadResult(ok=False, error=str(e))
        return DownloadResult(
            ok=True,
            path=info.get("path", dest),
            track_count=info.get("track_count", candidate.track_count),
        )
 ```
 - [ ] **Step 4: Run the test to verify it passes**
 Run:
 ```bash
 cd worker
 worker/.venv/bin/python -m pytest tests/test_soulseek_adapter.py -v
 ```
 Expected: PASS — all six tests green.
 - [ ] **Step 5: Commit**
 ```bash
 git add worker/lyra_worker/adapters/soulseek.py worker/tests/test_soulseek_adapter.py
 git commit -m "feat: add SoulseekAdapter with injectable client seam"
 ```
 ---
 ### Task 2: Real SlskdClient (slskd REST) + requests dependency + opt-in live test
 **Files:**
 - Create: `worker/lyra_worker/adapters/_slskd.py`
 - Create: `worker/tests/test_soulseek_live.py`
 - Modify: `worker/requirements.txt`
 **Interfaces:**
 - Consumes: `requests`.
 - Produces: `SlskdClient(config: dict)` implementing the `SoulseekClient` seam against slskd's REST API. Reads `slskd.url`/`slskd.api_key`; `is_configured()` = both present. `search_album` starts a search, polls to completion, groups each peer's audio files by directory into candidates. `download` decodes the `source_ref`, enqueues the files, and polls transfers to completion.
 - [ ] **Step 1: Add the dependency**
 Append to `worker/requirements.txt`:
 ```
 requests>=2.31
 ```
 Install into the venv: `worker/.venv/bin/pip install -r worker/requirements.txt`
 - [ ] **Step 2: Implement the real client**
 `worker/lyra_worker/adapters/_slskd.py`:
 ```python
 import json
 import os
 import time
 from typing import Callable
 import requests
 _AUDIO_EXT = {"flac", "mp3", "m4a", "ogg", "opus", "wav", "aac"}
 _LOSSLESS_EXT = {"flac", "wav"}
 _SEARCH_POLLS = 40   # * 3s ≈ 2 min
 _XFER_POLLS = 200    # * 3s ≈ 10 min
 _POLL_SECONDS = 3
 def _basename(path: str) -> str:
    return path.rsplit("\\", 1)[-1].rsplit("/", 1)[-1]
 def _dirname(path: str) -> str:
    if "\\" in path:
        return path.rsplit("\\", 1)[0]
    if "/" in path:
        return path.rsplit("/", 1)[0]
    return ""
 class SlskdClient:
    """Real Soulseek client talking to an slskd daemon's REST API.
    NOT unit-tested offline; see test_soulseek_live.py. slskd response/transfer
    JSON shapes may vary by version — validate with the live test.
    """
    def __init__(self, config: dict):
        self._url = (config.get("slskd.url", "") or "").rstrip("/")
        self._key = config.get("slskd.api_key", "")
    def is_configured(self) -> bool:
        return bool(self._url and self._key)
    def _headers(self) -> dict:
        return {"X-API-Key": self._key, "Content-Type": "application/json"}
    def _get(self, path: str) -> dict | list:
        r = requests.get(f"{self._url}{path}", headers=self._headers(), timeout=30)
        r.raise_for_status()
        return r.json()
    def search_album(self, artist: str, album: str) -> list[dict]:
        r = requests.post(
            f"{self._url}/api/v0/searches",
            headers=self._headers(),
            json={"searchText": f"{artist} {album}"},
            timeout=30,
        )
        r.raise_for_status()
        search_id = r.json()["id"]
        for _ in range(_SEARCH_POLLS):
            time.sleep(_POLL_SECONDS)
            state = self._get(f"/api/v0/searches/{search_id}")
            if state.get("isComplete") or "Completed" in str(state.get("state", "")):
                break
        responses = self._get(f"/api/v0/searches/{search_id}/responses")
        candidates: list[dict] = []
        for resp in responses:
            username = resp.get("username", "")
            by_dir: dict[str, list] = {}
            for f in resp.get("files", []) or []:
                name = f.get("filename", "")
                ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
                if ext not in _AUDIO_EXT:
                    continue
                by_dir.setdefault(_dirname(name), []).append(
                    {"filename": name, "size": f.get("size", 0), "ext": ext}
                )
            for directory, dfiles in by_dir.items():
                if not dfiles:
                    continue
                lossless = all(f["ext"] in _LOSSLESS_EXT for f in dfiles)
                dirname = _basename(directory)
                if " - " in dirname:
                    guess_artist, guess_album = (p.strip() for p in dirname.split(" - ", 1))
                else:
                    guess_artist, guess_album = "", dirname
                candidates.append(
                    {
                        "source_ref": json.dumps(
                            {
                                "username": username,
                                "files": [{"filename": f["filename"], "size": f["size"]} for f in dfiles],
                            }
                        ),
                        "title": guess_album,
                        "artist": guess_artist,
                        "track_count": len(dfiles),
                        "format": "FLAC" if lossless else "MP3",
                        "bitrate": None,
                    }
                )
        return candidates
    def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict:
        ref = json.loads(source_ref)
        username = ref["username"]
        files = ref["files"]
        os.makedirs(dest, exist_ok=True)
        r = requests.post(
            f"{self._url}/api/v0/transfers/downloads/{username}",
            headers=self._headers(),
            json=[{"filename": f["filename"], "size": f["size"]} for f in files],
            timeout=30,
        )
        r.raise_for_status()
        on_progress(0.0)
        total = len(files)
        for _ in range(_XFER_POLLS):
            time.sleep(_POLL_SECONDS)
            data = self._get(f"/api/v0/transfers/downloads/{username}")
            states: list[str] = []
            for directory in data.get("directories", []) if isinstance(data, dict) else []:
                for f in directory.get("files", []):
                    states.append(str(f.get("state", "")))
            if states:
                done = sum(1 for s in states if "Completed" in s)
                on_progress(min(1.0, done / max(total, 1)))
                if any(bad in s for s in states for bad in ("Errored", "Rejected", "Cancelled")):
                    raise RuntimeError("soulseek transfer failed")
                if done >= total:
                    break
        return {"track_count": total, "path": dest}
 ```
 - [ ] **Step 3: Write the opt-in live test**
 `worker/tests/test_soulseek_live.py`:
 ```python
 import os
 import pytest
 from lyra_worker.adapters._slskd import SlskdClient
 pytestmark = pytest.mark.skipif(
    not (os.environ.get("LYRA_LIVE_TESTS") and os.environ.get("SLSKD_URL") and os.environ.get("SLSKD_API_KEY")),
    reason="live slskd test; set LYRA_LIVE_TESTS=1 + SLSKD_URL + SLSKD_API_KEY to run",
 )
 def test_live_search_returns_results():
    client = SlskdClient({"slskd.url": os.environ["SLSKD_URL"], "slskd.api_key": os.environ["SLSKD_API_KEY"]})
    results = client.search_album("John Mayer", "Continuum")
    assert isinstance(results, list)
    # Soulseek results depend on who's online; assert shape, not count.
    for r in results:
        assert r["source_ref"] and r["track_count"] >= 1
        assert r["format"] in ("FLAC", "MP3")
 ```
 - [ ] **Step 4: Verify (offline) the module imports and the live test is skipped**
 Run:
 ```bash
 cd worker
 worker/.venv/bin/python -c "from lyra_worker.adapters._slskd import SlskdClient; print('import ok', SlskdClient({}).is_configured())"
 worker/.venv/bin/python -m pytest tests/test_soulseek_live.py -v
 ```
 Expected: prints `import ok False`; the live test is reported **skipped**. Do NOT set the live env vars here.
 - [ ] **Step 5: Commit**
 ```bash
 git add worker/lyra_worker/adapters/_slskd.py worker/tests/test_soulseek_live.py worker/requirements.txt
 git commit -m "feat: add real slskd Soulseek client and opt-in live test"
 ```
 ---
 ### Task 3: Add Soulseek to the config-aware registry
 **Files:**
 - Modify: `worker/lyra_worker/registry.py`
 - Modify: `worker/tests/test_registry.py`
 **Interfaces:**
 - Consumes: `SoulseekAdapter` + `SlskdClient` (Tasks 1-2), plus the existing Qobuz/YouTube adapters.
 - Produces: `build_adapters(config)` now also constructs `SoulseekAdapter(SlskdClient(config))`, included only when `health()` (slskd.url + slskd.api_key present). Order best-tier first: Qobuz (0), Soulseek (1), YouTube (2).
 - [ ] **Step 1: Update the registry test (new expectations first)**
 Replace `worker/tests/test_registry.py` with:
 ```python
 from lyra_worker.adapters.qobuz import QobuzAdapter
 from lyra_worker.adapters.soulseek import SoulseekAdapter
 from lyra_worker.adapters.youtube import YouTubeAdapter
 from lyra_worker.registry import build_adapters
 QOBUZ_CFG = {"qobuz.email": "me@example.com", "qobuz.password": "secret"}
 SLSKD_CFG = {"slskd.url": "http://slskd:5030", "slskd.api_key": "key"}
 def test_only_youtube_without_any_config():
    names = [a.name for a in build_adapters({})]
    assert names == ["youtube"]  # qobuz + soulseek both gated off
 def test_soulseek_included_when_configured():
    adapters = build_adapters(SLSKD_CFG)
    names = [a.name for a in adapters]
    assert "soulseek" in names and "youtube" in names
    assert "qobuz" not in names
    assert isinstance(next(a for a in adapters if a.name == "soulseek"), SoulseekAdapter)
 def test_all_three_configured_ordered_by_tier():
    adapters = build_adapters({**QOBUZ_CFG, **SLSKD_CFG})
    names = [a.name for a in adapters]
    assert names == ["qobuz", "soulseek", "youtube"]  # tiers 0,1,2
    assert len(names) == len(set(names))
    assert isinstance(next(a for a in adapters if a.name == "qobuz"), QobuzAdapter)
    assert isinstance(next(a for a in adapters if a.name == "youtube"), YouTubeAdapter)
 ```
 - [ ] **Step 2: Run the test to verify it fails**
 Run:
 ```bash
 cd worker
 worker/.venv/bin/python -m pytest tests/test_registry.py -v
 ```
 Expected: FAIL — the current registry has no Soulseek adapter, so `test_soulseek_included_when_configured` and `test_all_three_configured_ordered_by_tier` fail.
 - [ ] **Step 3: Update the registry**
 Replace `worker/lyra_worker/registry.py` with:
 ```python
 from lyra_worker.adapters._slskd import SlskdClient
 from lyra_worker.adapters._streamrip import StreamripClient
 from lyra_worker.adapters._ytdlp import YtDlpClient
 from lyra_worker.adapters.base import SourceAdapter
 from lyra_worker.adapters.qobuz import QobuzAdapter
 from lyra_worker.adapters.soulseek import SoulseekAdapter
 from lyra_worker.adapters.youtube import YouTubeAdapter
 def build_adapters(config: dict) -> list[SourceAdapter]:
    """Real source adapters, best-tier first, filtered to those that are configured.
    An adapter is included only if its health() is true (Qobuz needs credentials,
    Soulseek needs an slskd URL + API key). YouTube needs no config.
    """
    candidates: list[SourceAdapter] = [
        QobuzAdapter(StreamripClient(config)),
        SoulseekAdapter(SlskdClient(config)),
        YouTubeAdapter(YtDlpClient()),
    ]
    return [a for a in candidates if a.health()]
 ```
 - [ ] **Step 4: Run the tests to verify they pass**
 Run:
 ```bash
 cd worker
 worker/.venv/bin/python -m pytest tests/test_registry.py -v
 ```
 Expected: PASS — all three tests green.
 - [ ] **Step 5: Commit**
 ```bash
 git add worker/lyra_worker/registry.py worker/tests/test_registry.py
 git commit -m "feat: add Soulseek to the config-aware registry"
 ```
 ---
 ### Task 4: Pipeline integration + full ranking order
 **Files:**
 - Test: `worker/tests/test_soulseek_pipeline.py`
 **Interfaces:**
 - Consumes: `run_pipeline`, `SoulseekAdapter` (Task 1), `FakeSlskdClient` (Task 1's test), the Qobuz/YouTube adapters + their fakes, `conn` fixture + `insert_request`.
 - Produces: proof that the real `SoulseekAdapter` class (fake client) flows through the pipeline to `imported` with a `soulseek` FLAC LibraryItem, and that the full four-way ranking holds (Qobuz > Soulseek FLAC > YouTube).
 - [ ] **Step 1: Write the tests**
 `worker/tests/test_soulseek_pipeline.py`:
 ```python
 from lyra_worker.adapters.qobuz import QobuzAdapter
 from lyra_worker.adapters.soulseek import SoulseekAdapter
 from lyra_worker.adapters.youtube import YouTubeAdapter
 from lyra_worker.claim import claim_next
 from lyra_worker.pipeline import run_pipeline
 from tests.conftest import insert_request
 from tests.test_qobuz_adapter import FakeQobuzClient
 from tests.test_soulseek_adapter import FakeSlskdClient
 from tests.test_youtube_adapter import FakeYtClient
 def test_soulseek_flows_through_pipeline(conn):
    job_id = insert_request(conn, artist="John Mayer", album="Continuum")
    claim_next(conn)
    run_pipeline(conn, job_id, [SoulseekAdapter(FakeSlskdClient())], dest_root="/tmp/lib")
    with conn.cursor() as cur:
        cur.execute('SELECT state, "currentStage" FROM "Job" WHERE id = %s', (job_id,))
        assert cur.fetchone() == ("imported", "import")
        cur.execute('SELECT source, format FROM "LibraryItem" WHERE artist = %s', ("John Mayer",))
        assert cur.fetchone() == ("soulseek", "FLAC")
 def test_soulseek_flac_beats_youtube(conn):
    yt_results = [{"source_ref": "yt1", "title": "Continuum", "artist": "John Mayer", "track_count": 12}]
    job_id = insert_request(conn, artist="John Mayer", album="Continuum")
    claim_next(conn)
    run_pipeline(
        conn, job_id,
        [SoulseekAdapter(FakeSlskdClient()), YouTubeAdapter(FakeYtClient(results=yt_results))],
        dest_root="/tmp/lib",
    )
    with conn.cursor() as cur:
        cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
        assert cur.fetchone()[0] == "soulseek"  # FLAC (class 2) beats YouTube lossy (class 1)
 def test_qobuz_beats_soulseek(conn):
    job_id = insert_request(conn, artist="John Mayer", album="Continuum")
    claim_next(conn)
    run_pipeline(
        conn, job_id,
        [QobuzAdapter(FakeQobuzClient()), SoulseekAdapter(FakeSlskdClient())],
        dest_root="/tmp/lib",
    )
    with conn.cursor() as cur:
        cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
        # Qobuz hi-res (class 3) OR same-class-lower-tier both put Qobuz first; Qobuz wins.
        assert cur.fetchone()[0] == "qobuz"
 ```
 - [ ] **Step 2: Run the tests**
 Run:
 ```bash
 cd worker
 export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
 worker/.venv/bin/python -m pytest tests/test_soulseek_pipeline.py -v
 ```
 Expected: PASS immediately — exercises already-implemented code. If any FAILS, investigate a real integration mismatch and report it rather than weakening assertions. (`FakeSlskdClient`/`FakeQobuzClient`/`FakeYtClient` all echo the target, so all clear the confidence gate; ranking then decides.)
 - [ ] **Step 3: Run the full worker suite**
 Run:
 ```bash
 cd worker
 export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
 worker/.venv/bin/python -m pytest -v
 ```
 Expected: PASS — every suite green (soulseek adapter/pipeline, qobuz, youtube, registry, config, crypto, pipeline, ranker, etc.), all three live tests skipped, output pristine.
 - [ ] **Step 4: Commit**
 ```bash
 git add worker/tests/test_soulseek_pipeline.py
 git commit -m "test: verify SoulseekAdapter through the pipeline and full ranking order"
 ```
 ---
 ## Notes for the next plan
 All three real download adapters (Qobuz, Soulseek, YouTube) are now in place and config-gated. Plan 4 is the priority: real MusicBrainz resolution in `intake` (fills `MBTarget.track_count`/`total_duration_s` + canonical artist/album — which makes the confidence gate and the tag-stage completeness check meaningful for ALL sources, fixing the "grabbed a single track / weak metadata scores below the gate" issues on YouTube and Soulseek), real tagging/file-move in `tag`/`import`, and the persistent `/music` volume mount in docker-compose (currently downloads land in ephemeral container storage). The **live validation of `SlskdClient`** (Task 2) against the user's slskd is a user step: run `LYRA_LIVE_TESTS=1 SLSKD_URL=… SLSKD_API_KEY=… pytest tests/test_soulseek_live.py`, and adjust the response/transfer JSON field names/state strings if the installed slskd version differs.
 ```