# Lyra Qobuz 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 Qobuz source adapter (lossless/hi-res FLAC via streamrip), implementing the existing `SourceAdapter` contract, and make the registry **config-aware** so Qobuz is included only when the user has entered credentials (via the settings page from Plan 3a). Adapter logic is unit-tested offline against a fake client; the real streamrip path is exercised only by an opt-in live test the user runs with their Qobuz account. **Architecture:** `QobuzAdapter` depends on an injectable `QobuzClient` seam (like `YtClient`). Tests inject a `FakeQobuzClient`; the real `StreamripClient` (wraps streamrip's async `QobuzClient`, bridged to sync via `asyncio.run`) is constructed only in `registry.build_adapters(config)`. The registry now takes the resolved config dict, builds each real adapter, and includes it only if `adapter.health()` is true — so an unconfigured source is skipped. The worker loop reads config once at startup and passes it in. **Tech Stack:** Python 3.12 worker; `streamrip` (new dependency, async, v2 API). Tests: pytest with a fake client (offline) + an opt-in live test gated by `LYRA_LIVE_TESTS` **and** real Qobuz creds in env. ## Global Constraints - Python 3.12 in the worker container; version-flexible local deps. - **New dependency allowed:** `streamrip` (v2, async). No other new deps. - The `SourceAdapter` protocol, `Candidate`/`Quality`/`DownloadResult`/`MBTarget` types, ranker, confidence, and pipeline are UNCHANGED. The adapter only implements the protocol. - Qobuz: `name="qobuz"`, `tier=0`. Quality is lossless FLAC: `Quality(fmt="FLAC", lossless=True, bit_depth=, sample_rate=)`. Qobuz reports `maximum_sampling_rate` in **kHz** (e.g. 44.1, 96) — convert to Hz (`int(khz*1000)`). - `registry.build_adapters(config: dict) -> list[SourceAdapter]` — NEW signature takes the resolved config; returns only adapters whose `health()` is true. Qobuz `health()` = credentials present; YouTube `health()` = always true. - The real `StreamripClient` is NOT unit-tested offline; it imports `streamrip` LAZILY (inside methods) so importing the module and constructing the client stay offline/cheap. Its network path is covered only by the opt-in live test (`LYRA_LIVE_TESTS` + `QOBUZ_EMAIL`/`QOBUZ_PASSWORD` env). Normal runs never hit the network. - **Live-validation note (carry into the live test):** streamrip's embedding API is version-sensitive. The `StreamripClient` here follows streamrip v2 docs (`Config.defaults()`, `config.session.qobuz.*`, `QobuzClient.search`/`login`, `PendingAlbum.resolve()`/`album.rip()`); exact attribute paths (`downloads.folder`, `album.tracks`) and password handling may need adjustment against the installed streamrip version — that is the expected purpose of the live test. - Config is read ONCE at worker startup; changing credentials requires a worker restart to take effect (document; this matches current UX). - Every task ends with a commit. ## Shared interfaces (defined by tasks below) ```python # worker/lyra_worker/adapters/qobuz.py (Task 1) class QobuzClient(Protocol): def is_configured(self) -> bool: ... def search_album(self, artist: str, album: str) -> list[dict]: ... # {source_ref, title, artist, track_count, bit_depth, sampling_rate} def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict: ... # {track_count, path}; raises on failure class QobuzAdapter: # implements SourceAdapter name = "qobuz"; tier = 0 def __init__(self, client: QobuzClient): ... 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/_streamrip.py (Task 2) class StreamripClient: # real QobuzClient backed by streamrip (async bridged to sync) 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, changed signature) def build_adapters(config: dict) -> list[SourceAdapter]: ... ``` --- ### Task 1: QobuzAdapter + client seam (offline, fake client) **Files:** - Create: `worker/lyra_worker/adapters/qobuz.py` - Test: `worker/tests/test_qobuz_adapter.py` **Interfaces:** - Consumes: `MBTarget`, `Candidate`, `Quality`, `DownloadResult`; `SourceAdapter`. - Produces: `QobuzClient` Protocol (with `is_configured`) and `QobuzAdapter`. `search` maps each result dict to `Candidate(source="qobuz", source_ref, matched_artist=r["artist"], matched_album=r["title"], quality=FLAC with bit_depth + sample_rate=int(r["sampling_rate"]*1000), track_count, source_tier=0)`. `health()` delegates to `client.is_configured()`. `download` wraps `client.download` (exception → `ok=False`). - [ ] **Step 1: Write the failing test** `worker/tests/test_qobuz_adapter.py`: ```python from lyra_worker.adapters.qobuz import QobuzAdapter from lyra_worker.quality import quality_class from lyra_worker.types import MBTarget TARGET = MBTarget(artist="John Mayer", album="Continuum", track_count=12) class FakeQobuzClient: def __init__(self, configured=True, results=None, fail=False): self._configured = configured self._results = results if results is not None else [ {"source_ref": "qz123", "title": "Continuum", "artist": "John Mayer", "track_count": 12, "bit_depth": 24, "sampling_rate": 96.0} ] 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("qobuz boom") on_progress(1.0) return {"track_count": 12, "path": dest} def test_name_and_tier(): a = QobuzAdapter(FakeQobuzClient()) assert a.name == "qobuz" assert a.tier == 0 def test_health_reflects_configuration(): assert QobuzAdapter(FakeQobuzClient(configured=True)).health() is True assert QobuzAdapter(FakeQobuzClient(configured=False)).health() is False def test_search_maps_to_lossless_hires_candidate(): cands = QobuzAdapter(FakeQobuzClient()).search(TARGET) assert len(cands) == 1 c = cands[0] assert c.source == "qobuz" assert c.source_ref == "qz123" assert c.matched_artist == "John Mayer" assert c.matched_album == "Continuum" assert c.track_count == 12 assert c.source_tier == 0 assert c.quality.lossless is True assert c.quality.fmt == "FLAC" assert c.quality.sample_rate == 96000 # 96.0 kHz -> Hz assert c.quality.bit_depth == 24 assert quality_class(c.quality) == 3 # hi-res def test_search_empty_when_no_results(): assert QobuzAdapter(FakeQobuzClient(results=[])).search(TARGET) == [] def test_download_success_and_failure(): ok = QobuzAdapter(FakeQobuzClient()).download( QobuzAdapter(FakeQobuzClient()).search(TARGET)[0], "/tmp/x", lambda _p: None) assert ok.ok is True and ok.track_count == 12 bad = QobuzAdapter(FakeQobuzClient(fail=True)).download( QobuzAdapter(FakeQobuzClient()).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_qobuz_adapter.py -v ``` Expected: FAIL — `ModuleNotFoundError: No module named 'lyra_worker.adapters.qobuz'`. - [ ] **Step 3: Implement the adapter** `worker/lyra_worker/adapters/qobuz.py`: ```python from typing import Callable, Protocol from lyra_worker.types import Candidate, DownloadResult, MBTarget, Quality class QobuzClient(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, bit_depth, sampling_rate(kHz).""" ... 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(bit_depth: int, sampling_rate_khz: float) -> Quality: return Quality( fmt="FLAC", lossless=True, bit_depth=bit_depth, sample_rate=int(sampling_rate_khz * 1000), ) class QobuzAdapter: name = "qobuz" tier = 0 def __init__(self, client: QobuzClient): 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="qobuz", source_ref=r["source_ref"], matched_artist=r.get("artist", ""), matched_album=r.get("title", ""), quality=_quality(r.get("bit_depth", 16), r.get("sampling_rate", 44.1)), 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_qobuz_adapter.py -v ``` Expected: PASS — all six tests green. - [ ] **Step 5: Commit** ```bash git add worker/lyra_worker/adapters/qobuz.py worker/tests/test_qobuz_adapter.py git commit -m "feat: add QobuzAdapter with injectable client seam" ``` --- ### Task 2: Real StreamripClient + streamrip dependency + opt-in live test **Files:** - Create: `worker/lyra_worker/adapters/_streamrip.py` - Create: `worker/tests/test_qobuz_live.py` - Modify: `worker/requirements.txt` **Interfaces:** - Consumes: `streamrip` (lazily imported inside methods). - Produces: `StreamripClient(config: dict)` implementing the `QobuzClient` seam. Reads `qobuz.email`/`qobuz.password` from config; `is_configured()` = both present; `search_album`/`download` bridge streamrip's async API via `asyncio.run`. - [ ] **Step 1: Add the dependency** Append to `worker/requirements.txt`: ``` streamrip>=2.0 ``` Install into the venv: `worker/.venv/bin/pip install -r worker/requirements.txt` - [ ] **Step 2: Implement the real client** `worker/lyra_worker/adapters/_streamrip.py`: ```python import asyncio import os from typing import Callable class StreamripClient: """Real Qobuz client backed by streamrip (v2, async → sync via asyncio.run). NOT unit-tested offline; see test_qobuz_live.py. streamrip is imported lazily inside the async helpers so importing/constructing this class stays offline. """ def __init__(self, config: dict): self._email = config.get("qobuz.email", "") self._password = config.get("qobuz.password", "") def is_configured(self) -> bool: return bool(self._email and self._password) def _make_config(self, download_folder: str | None = None): from streamrip.config import Config cfg = Config.defaults() cfg.session.qobuz.use_auth_token = False cfg.session.qobuz.email_or_userid = self._email cfg.session.qobuz.password_or_token = self._password cfg.session.qobuz.quality = 3 if download_folder is not None: cfg.session.downloads.folder = download_folder return cfg def search_album(self, artist: str, album: str) -> list[dict]: return asyncio.run(self._search(f"{artist} {album}")) async def _search(self, query: str) -> list[dict]: from streamrip.client import QobuzClient client = QobuzClient(self._make_config()) await client.login() try: pages = await client.search("album", query, limit=5) finally: await client.session.close() results: list[dict] = [] for page in pages: for item in page.get("albums", {}).get("items", []): results.append( { "source_ref": str(item.get("id", "")), "title": item.get("title", ""), "artist": (item.get("artist") or {}).get("name", ""), "track_count": item.get("tracks_count", 0) or 0, "bit_depth": item.get("maximum_bit_depth", 16) or 16, "sampling_rate": item.get("maximum_sampling_rate", 44.1) or 44.1, } ) return results def download(self, source_ref: str, dest: str, on_progress: Callable[[float], None]) -> dict: return asyncio.run(self._download(source_ref, dest, on_progress)) async def _download(self, album_id: str, dest: str, on_progress: Callable[[float], None]) -> dict: os.makedirs(dest, exist_ok=True) from streamrip.client import QobuzClient from streamrip.db import Database, Dummy from streamrip.media import PendingAlbum client = QobuzClient(self._make_config(download_folder=dest)) await client.login() try: on_progress(0.0) db = Database(downloads=Dummy(), failed=Dummy()) pending = PendingAlbum(album_id, client, self._make_config(download_folder=dest), db) album = await pending.resolve() if album is None: raise RuntimeError(f"could not resolve Qobuz album {album_id}") await album.rip() track_count = len(getattr(album, "tracks", []) or []) on_progress(1.0) finally: await client.session.close() return {"track_count": track_count, "path": dest} ``` - [ ] **Step 3: Write the opt-in live test** `worker/tests/test_qobuz_live.py`: ```python import os import pytest from lyra_worker.adapters._streamrip import StreamripClient pytestmark = pytest.mark.skipif( not (os.environ.get("LYRA_LIVE_TESTS") and os.environ.get("QOBUZ_EMAIL") and os.environ.get("QOBUZ_PASSWORD")), reason="live Qobuz test; set LYRA_LIVE_TESTS=1 + QOBUZ_EMAIL + QOBUZ_PASSWORD to run", ) def test_live_search_returns_results(): client = StreamripClient( {"qobuz.email": os.environ["QOBUZ_EMAIL"], "qobuz.password": os.environ["QOBUZ_PASSWORD"]} ) results = client.search_album("John Mayer", "Continuum") assert len(results) >= 1 assert all(r["source_ref"] for r in results) # sanity: at least one result is a plausible lossless match assert any(r["bit_depth"] >= 16 for r in results) ``` - [ ] **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._streamrip import StreamripClient; print('import ok', StreamripClient({}).is_configured())" worker/.venv/bin/python -m pytest tests/test_qobuz_live.py -v ``` Expected: prints `import ok False` (empty config → not configured, and importing the module does NOT import streamrip); the live test is reported **skipped**. Do NOT set the live env vars here. - [ ] **Step 5: Commit** ```bash git add worker/lyra_worker/adapters/_streamrip.py worker/tests/test_qobuz_live.py worker/requirements.txt git commit -m "feat: add real streamrip Qobuz client and opt-in live test" ``` --- ### Task 3: Config-aware registry + worker wiring **Files:** - Modify: `worker/lyra_worker/registry.py` - Modify: `worker/lyra_worker/main.py` - Modify: `worker/tests/test_registry.py` **Interfaces:** - Consumes: `QobuzAdapter` + `StreamripClient` (Tasks 1-2), `YouTubeAdapter` + `YtDlpClient` (Plan 3b), `get_config` (Plan 3a). - Produces: `build_adapters(config: dict) -> list[SourceAdapter]` returning `[QobuzAdapter(StreamripClient(config)), YouTubeAdapter(YtDlpClient())]` filtered to those whose `health()` is true. `main.run_forever()` reads config once via `get_config(conn)` and passes it in. - [ ] **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.youtube import YouTubeAdapter from lyra_worker.registry import build_adapters QOBUZ_CFG = {"qobuz.email": "me@example.com", "qobuz.password": "secret"} def test_youtube_always_present_qobuz_skipped_without_config(): adapters = build_adapters({}) names = [a.name for a in adapters] assert "youtube" in names assert "qobuz" not in names # no creds -> health() false -> excluded def test_qobuz_included_when_configured(): adapters = build_adapters(QOBUZ_CFG) names = [a.name for a in adapters] assert "qobuz" in names assert "youtube" in names qz = next(a for a in adapters if a.name == "qobuz") yt = next(a for a in adapters if a.name == "youtube") assert isinstance(qz, QobuzAdapter) assert isinstance(yt, YouTubeAdapter) def test_adapter_names_unique_and_qobuz_first(): adapters = build_adapters(QOBUZ_CFG) names = [a.name for a in adapters] assert len(names) == len(set(names)) assert names[0] == "qobuz" # tier 0 listed first ``` - [ ] **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 — old `build_adapters()` takes no args (TypeError when called with a config dict). - [ ] **Step 3: Update the registry** Replace `worker/lyra_worker/registry.py` with: ```python 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.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 (e.g. Qobuz requires credentials in config). Plan 3d adds Soulseek (slskd) the same way. """ candidates: list[SourceAdapter] = [ QobuzAdapter(StreamripClient(config)), YouTubeAdapter(YtDlpClient()), ] return [a for a in candidates if a.health()] ``` - [ ] **Step 4: Wire config into the worker loop** In `worker/lyra_worker/main.py`, update the imports and `run_forever()` so config is read once and passed to `build_adapters`. The function becomes: ```python import time from lyra_worker.claim import claim_next from lyra_worker.config import get_config from lyra_worker.db import wait_for_db from lyra_worker.pipeline import run_pipeline from lyra_worker.registry import build_adapters IDLE_SLEEP = 2.0 def run_forever() -> None: conn = wait_for_db() adapters = build_adapters(get_config(conn)) print(f"worker: {len(adapters)} adapter(s) enabled: {[a.name for a in adapters]}", flush=True) print("worker: waiting for jobs", flush=True) try: while True: job_id = claim_next(conn) if job_id is None: time.sleep(IDLE_SLEEP) continue print(f"worker: claimed job {job_id}", flush=True) run_pipeline(conn, job_id, adapters) print(f"worker: finished job {job_id}", flush=True) finally: conn.close() if __name__ == "__main__": run_forever() ``` (Keep the `if __name__ == "__main__"` guard.) - [ ] **Step 5: Run the tests to verify they pass** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest tests/test_registry.py -v ``` Expected: PASS — all three tests green. - [ ] **Step 6: Commit** ```bash git add worker/lyra_worker/registry.py worker/lyra_worker/main.py worker/tests/test_registry.py git commit -m "feat: config-aware registry gating Qobuz on credentials; wire config into worker" ``` --- ### Task 4: Pipeline integration with the real Qobuz adapter class **Files:** - Test: `worker/tests/test_qobuz_pipeline.py` **Interfaces:** - Consumes: `run_pipeline`, `QobuzAdapter` (Task 1), `FakeQobuzClient` (Task 1's test), `conn` fixture + `insert_request`. - Produces: proof that the real `QobuzAdapter` class (driven by a fake client, offline) flows through the pipeline to `imported`, choosing a `qobuz` candidate and writing a `qobuz` LibraryItem. - [ ] **Step 1: Write the test** `worker/tests/test_qobuz_pipeline.py`: ```python from lyra_worker.adapters.qobuz import QobuzAdapter 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 def test_qobuz_adapter_flows_through_pipeline(conn): job_id = insert_request(conn, artist="John Mayer", album="Continuum") claim_next(conn) run_pipeline(conn, job_id, [QobuzAdapter(FakeQobuzClient())], 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 FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) assert cur.fetchone()[0] == "qobuz" cur.execute('SELECT source, format FROM "LibraryItem" WHERE artist = %s', ("John Mayer",)) row = cur.fetchone() assert row == ("qobuz", "FLAC") def test_qobuz_and_youtube_ranked_qobuz_wins(conn): from lyra_worker.adapters.youtube import YouTubeAdapter from tests.test_youtube_adapter import FakeYtClient # Both sources return an exact-title match; Qobuz (lossless, tier 0) must outrank YouTube (lossy). 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, [QobuzAdapter(FakeQobuzClient()), 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] == "qobuz" # lossless beats lossy ``` - [ ] **Step 2: Run the test** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest tests/test_qobuz_pipeline.py -v ``` Expected: PASS immediately — exercises already-implemented code (adapter from Task 1 + existing pipeline/ranker). If it FAILS, investigate a real integration mismatch and report it rather than altering assertions. (Note: `FakeQobuzClient` echoes the target, so confidence clears the gate; Qobuz outranks YouTube via `quality_class` 2/3 > 1 and tier 0.) - [ ] **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 (qobuz adapter/pipeline, youtube, registry, config, crypto, pipeline, ranker, etc.), both live tests skipped, output pristine. - [ ] **Step 4: Commit** ```bash git add worker/tests/test_qobuz_pipeline.py git commit -m "test: verify real QobuzAdapter flows through the pipeline and outranks YouTube" ``` --- ## Notes for the next plans `registry.build_adapters(config)` is now config-aware and returns Qobuz (when configured) + YouTube. Plan 3d adds `SoulseekAdapter` (slskd REST client + an `slskd` service in docker-compose), following the same injectable-client + fake + opt-in-live pattern, gated on `slskd.url`/`slskd.api_key` in config via `health()`. Plan 4 then adds real MusicBrainz resolution in `intake` (filling `MBTarget.track_count`/`total_duration_s` — which makes the confidence gate and the tag-stage completeness check meaningful for all sources) plus real tagging/file-move in `tag`/`import`, and should also add the persistent `/music` volume mount so downloads survive container recreation. The **live validation of `StreamripClient`** (Task 2) against a real Qobuz account is a user step: run `LYRA_LIVE_TESTS=1 QOBUZ_EMAIL=… QOBUZ_PASSWORD=… pytest tests/test_qobuz_live.py`, and adjust the streamrip Config/attribute paths if the installed version differs. ```