Lyra/docs/superpowers/plans/2026-07-10-lyra-qobuz-adapter.md

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=<from Qobuz>, sample_rate=<Hz>). 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)

# 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:

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:

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:

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:

cd worker
worker/.venv/bin/python -m pytest tests/test_qobuz_adapter.py -v

Expected: PASS — all six tests green.

• Step 5: Commit

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:

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:

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:

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

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:

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:

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:

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:

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:

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

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:

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:

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:

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

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.