# Lyra MusicBrainz Intake + Search Hardening 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:** Make Lyra's matching accurate by resolving each request against MusicBrainz in the `intake` stage — filling the canonical artist/album, track count, year, and total duration — and use the canonical track count in the tag-stage completeness check so a partial download (e.g. a single-track YouTube "full album" video for a 12-track album) is correctly rejected to `needs_attention`. Also harden the match stage so one source's search failure can't crash the worker. **Architecture:** The pipeline's `intake` stage gains an optional injectable `MbResolver`. When present and it returns a canonical `MBTarget`, the pipeline uses that (with track count) for confidence scoring, dedupe, and the completeness check; when absent or no match, it falls back to the raw request exactly as today (fully backward-compatible — `resolver` defaults to `None`). The real `MusicBrainzResolver` uses `musicbrainzngs` (lazily imported), covered by an opt-in live test. The match loop wraps each adapter's `search` in try/except so a failing source contributes no candidates instead of crashing the job. **Tech Stack:** Python 3.12 worker; `musicbrainzngs` (new dependency). Tests: pytest with a fake resolver (offline) + an opt-in live test gated by `LYRA_LIVE_TESTS`. ## Global Constraints - Python 3.12 in the worker container; version-flexible local deps. - **New dependency:** `musicbrainzngs` (pure-Python, lightweight). Imported LAZILY inside the resolver's method so importing the module / constructing the resolver stays offline. No other new deps. - `ranker`, `confidence`, `quality`, and the `SourceAdapter`/adapter code are UNCHANGED. Changes are confined to `types.py` (one added field), `pipeline.py` (intake enrichment + tag check + search try/except), a new resolver module + seam, `registry`/`main` wiring, and tests. - **Backward compatibility is mandatory:** `run_pipeline`'s new `resolver` parameter defaults to `None`; with no resolver the pipeline behaves exactly as before (target track count `None`, tag check uses the winning candidate's count). All existing pipeline tests must keep passing unchanged. - `MBTarget` gains `year: int | None = None` (defaulted; existing constructions unaffected). - The real `MusicBrainzResolver` is NOT unit-tested offline; covered only by the opt-in live test (`LYRA_LIVE_TESTS`). Normal runs never hit the network. MusicBrainz requires a descriptive User-Agent and rate-limits to ~1 req/sec (musicbrainzngs enforces this by default). - Every task ends with a commit. ## Shared interfaces (defined by tasks below) ```python # worker/lyra_worker/types.py (Task 2): MBTarget gains `year: int | None = None` # worker/lyra_worker/resolver.py (Task 2) class MbResolver(Protocol): def resolve(self, artist: str, album: str) -> MBTarget | None: ... # canonical target, or None if no confident match # worker/lyra_worker/pipeline.py (Tasks 1-2): run_pipeline(conn, job_id, adapters, resolver=None, min_confidence=0.7, dest_root="/music") # worker/lyra_worker/_musicbrainz.py (Task 3) class MusicBrainzResolver: # real MbResolver using musicbrainzngs (lazily imported) def __init__(self, app_name="Lyra", version="0.1", contact="lyra@localhost"): ... def resolve(self, artist, album) -> MBTarget | None: ... # worker/lyra_worker/registry.py (Task 4): build_resolver() -> MbResolver ``` --- ### Task 1: Harden the match stage against a failing source **Files:** - Modify: `worker/lyra_worker/pipeline.py` (match loop) - Test: `worker/tests/test_pipeline_hardening.py` **Interfaces:** - Consumes: existing `run_pipeline`, the fakes. - Produces: the match loop wraps each `adapter.search(target)` in `try/except Exception`, logging and continuing on failure, so one source raising does not crash the job. - [ ] **Step 1: Write the failing test** `worker/tests/test_pipeline_hardening.py`: ```python 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_youtube_adapter import FakeYtClient class ExplodingAdapter: name = "boom" tier = 0 def health(self): return True def search(self, target): raise RuntimeError("search backend down") def download(self, candidate, dest, on_progress): raise AssertionError("should never be called") def test_one_failing_source_does_not_crash_the_job(conn): job_id = insert_request(conn, artist="John Mayer", album="Continuum") claim_next(conn) yt = [{"source_ref": "yt1", "title": "Continuum", "artist": "John Mayer", "track_count": 12}] # The exploding adapter must not abort the pipeline; YouTube should still win. run_pipeline(conn, job_id, [ExplodingAdapter(), YouTubeAdapter(FakeYtClient(results=yt))], dest_root="/tmp/lib") with conn.cursor() as cur: cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,)) assert cur.fetchone()[0] == "imported" cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,)) assert cur.fetchone()[0] == "youtube" ``` - [ ] **Step 2: Run the test to verify it fails** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest tests/test_pipeline_hardening.py -v ``` Expected: FAIL — the current match loop calls `adapter.search` without try/except, so `RuntimeError` propagates out of `run_pipeline` and the test errors. - [ ] **Step 3: Wrap per-adapter search in the match loop** In `worker/lyra_worker/pipeline.py`, find the match stage loop (currently): ```python found: list[Candidate] = [] for adapter in adapters: for c in adapter.search(target): found.append(replace(c, source_tier=adapter.tier, confidence=score_confidence(target, c))) ``` Replace it with: ```python found: list[Candidate] = [] for adapter in adapters: try: results = adapter.search(target) except Exception as e: # a down source contributes no candidates, never crashes the job print(f"pipeline: adapter {adapter.name} search failed: {e}", flush=True) continue for c in results: found.append(replace(c, source_tier=adapter.tier, confidence=score_confidence(target, c))) ``` - [ ] **Step 4: Run the test to verify it passes** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest tests/test_pipeline_hardening.py -v ``` Expected: PASS. - [ ] **Step 5: Run the full worker suite (nothing regressed)** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest -q ``` Expected: PASS — all prior tests still green, output pristine. - [ ] **Step 6: Commit** ```bash git add worker/lyra_worker/pipeline.py worker/tests/test_pipeline_hardening.py git commit -m "fix: isolate per-adapter search failures in the pipeline match loop" ``` --- ### Task 2: MBTarget.year + resolver seam + intake enrichment + completeness check **Files:** - Modify: `worker/lyra_worker/types.py` (add `year` to MBTarget) - Create: `worker/lyra_worker/resolver.py` (the `MbResolver` Protocol) - Modify: `worker/lyra_worker/pipeline.py` (intake enrichment + tag-stage completeness check) - Test: `worker/tests/test_intake.py` **Interfaces:** - Consumes: `MBTarget`, `score_confidence`, the fakes. - Produces: - `MBTarget` with `year: int | None = None`. - `resolver.MbResolver` Protocol: `resolve(artist, album) -> MBTarget | None`. - `run_pipeline(conn, job_id, adapters, resolver=None, ...)`: in intake, if `resolver` is not None, call `resolver.resolve(raw.artist, raw.album)`; if it returns a non-None `MBTarget`, use it as `target` (else keep the raw target). The dedupe check, confidence scoring, download, and completeness check all use this `target`. - Tag-stage completeness check uses the canonical count when known: `expected = target.track_count if target.track_count is not None else winner.track_count`; fail with "incomplete download" when `result.track_count < expected`. - [ ] **Step 1: Add `year` to MBTarget** In `worker/lyra_worker/types.py`, the `MBTarget` dataclass gains a `year` field (after `total_duration_s`): ```python @dataclass(frozen=True) class MBTarget: artist: str album: str track_count: int | None = None total_duration_s: int | None = None year: int | None = None ``` - [ ] **Step 2: Add the resolver Protocol** `worker/lyra_worker/resolver.py`: ```python from typing import Protocol from lyra_worker.types import MBTarget class MbResolver(Protocol): def resolve(self, artist: str, album: str) -> MBTarget | None: """Return a canonical MBTarget (with track_count) or None if no confident match.""" ... ``` - [ ] **Step 3: Write the failing tests** `worker/tests/test_intake.py`: ```python from lyra_worker.adapters.youtube import YouTubeAdapter from lyra_worker.claim import claim_next from lyra_worker.pipeline import run_pipeline from lyra_worker.types import MBTarget from tests.conftest import insert_request from tests.test_youtube_adapter import FakeYtClient class FakeResolver: def __init__(self, target): self._target = target def resolve(self, artist, album): return self._target def _job_state(conn, job_id): with conn.cursor() as cur: cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,)) return cur.fetchone()[0] def test_single_track_download_rejected_against_musicbrainz_count(conn): # MusicBrainz says Continuum has 12 tracks; a 1-track YouTube "full album" video must be rejected. resolver = FakeResolver(MBTarget(artist="John Mayer", album="Continuum", track_count=12)) yt = [{"source_ref": "yt1", "title": "Continuum", "artist": "John Mayer", "track_count": 1}] # FakeYtClient.download reports the same 1-track count for the single-video candidate: client = FakeYtClient(results=yt) client.download = lambda source_ref, dest, on_progress: (on_progress(1.0), {"track_count": 1, "path": dest})[1] job_id = insert_request(conn, artist="john mayer", album="continuum") claim_next(conn) run_pipeline(conn, job_id, [YouTubeAdapter(client)], resolver=resolver, dest_root="/tmp/lib") assert _job_state(conn, job_id) == "needs_attention" def test_complete_download_imports_with_musicbrainz_count(conn): resolver = FakeResolver(MBTarget(artist="John Mayer", album="Continuum", track_count=12)) yt = [{"source_ref": "yt1", "title": "Continuum", "artist": "John Mayer", "track_count": 12}] client = FakeYtClient(results=yt) client.download = lambda source_ref, dest, on_progress: (on_progress(1.0), {"track_count": 12, "path": dest})[1] job_id = insert_request(conn, artist="john mayer", album="continuum") claim_next(conn) run_pipeline(conn, job_id, [YouTubeAdapter(client)], resolver=resolver, dest_root="/tmp/lib") assert _job_state(conn, job_id) == "imported" with conn.cursor() as cur: # canonical artist/album from MusicBrainz are what get stored cur.execute('SELECT artist, album FROM "LibraryItem"') assert cur.fetchone() == ("John Mayer", "Continuum") def test_no_resolver_preserves_legacy_behavior(conn): # Without a resolver, target.track_count stays None and the candidate's own count is used. yt = [{"source_ref": "yt1", "title": "Continuum", "artist": "John Mayer", "track_count": 1}] client = FakeYtClient(results=yt) client.download = lambda source_ref, dest, on_progress: (on_progress(1.0), {"track_count": 1, "path": dest})[1] job_id = insert_request(conn, artist="John Mayer", album="Continuum") claim_next(conn) run_pipeline(conn, job_id, [YouTubeAdapter(client)], dest_root="/tmp/lib") # no resolver assert _job_state(conn, job_id) == "imported" # 1 == 1, passes as before def test_resolver_returning_none_falls_back_to_request(conn): resolver = FakeResolver(None) yt = [{"source_ref": "yt1", "title": "Continuum", "artist": "John Mayer", "track_count": 1}] client = FakeYtClient(results=yt) client.download = lambda source_ref, dest, on_progress: (on_progress(1.0), {"track_count": 1, "path": dest})[1] job_id = insert_request(conn, artist="John Mayer", album="Continuum") claim_next(conn) run_pipeline(conn, job_id, [YouTubeAdapter(client)], resolver=resolver, dest_root="/tmp/lib") assert _job_state(conn, job_id) == "imported" # falls back to raw target, 1==1 ``` - [ ] **Step 4: Run the tests to verify they fail** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest tests/test_intake.py -v ``` Expected: FAIL — `run_pipeline` does not yet accept a `resolver` argument (TypeError), and the completeness check does not use the target count. - [ ] **Step 5: Enrich intake and update the completeness check** In `worker/lyra_worker/pipeline.py`: (a) Change the signature: ```python def run_pipeline( conn: psycopg.Connection, job_id: str, adapters: Sequence[SourceAdapter], resolver=None, min_confidence: float = 0.7, dest_root: str = "/music", ) -> None: ``` (b) In the intake stage, after loading the raw target, enrich it via the resolver: ```python # 1. intake _set_state(conn, job_id, "matching", "intake") target = _load_target(conn, job_id) if resolver is not None: resolved = resolver.resolve(target.artist, target.album) if resolved is not None: target = resolved if _already_in_library(conn, target): ... ``` (Everything after — dedupe, match, rank, download — already uses `target`; leave those unchanged.) (c) In the tag stage, replace the completeness check: ```python # 5. tag (integrity check against the canonical track count when known) _set_state(conn, job_id, "tagging", "tag") expected = target.track_count if target.track_count is not None else winner.track_count if result.track_count < expected: _fail(conn, job_id, "incomplete download") return ``` - [ ] **Step 6: 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_intake.py -v ``` Expected: PASS — all four tests green (single-track rejected, complete imports with canonical names, legacy behavior preserved, None-resolver fallback). - [ ] **Step 7: Run the full worker suite (backward compatibility)** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest -q ``` Expected: PASS — every existing test still green (they call `run_pipeline` without a resolver, so target.track_count stays None and behavior is unchanged), output pristine. - [ ] **Step 8: Commit** ```bash git add worker/lyra_worker/types.py worker/lyra_worker/resolver.py worker/lyra_worker/pipeline.py worker/tests/test_intake.py git commit -m "feat: MusicBrainz-aware intake and completeness check against canonical track count" ``` --- ### Task 3: Real MusicBrainzResolver + dependency + opt-in live test **Files:** - Create: `worker/lyra_worker/_musicbrainz.py` - Create: `worker/tests/test_musicbrainz_live.py` - Modify: `worker/requirements.txt` **Interfaces:** - Consumes: `musicbrainzngs` (lazily imported), `MBTarget`. - Produces: `MusicBrainzResolver` implementing `MbResolver`: searches release-groups by artist+album, picks the best title match (fuzzy ≥ 0.6), then fetches a release's recordings for the track count + total duration; returns a canonical `MBTarget(artist, album, track_count, total_duration_s, year)` or `None`. - [ ] **Step 1: Add the dependency** Append to `worker/requirements.txt`: ``` musicbrainzngs>=0.7.1 ``` Install into the venv: `worker/.venv/bin/pip install -r worker/requirements.txt` - [ ] **Step 2: Implement the real resolver** `worker/lyra_worker/_musicbrainz.py`: ```python from difflib import SequenceMatcher from lyra_worker.types import MBTarget _MIN_TITLE_RATIO = 0.6 def _ratio(a: str, b: str) -> float: return SequenceMatcher(None, a.strip().casefold(), b.strip().casefold()).ratio() class MusicBrainzResolver: """Real MbResolver using the MusicBrainz webservice via musicbrainzngs. NOT unit-tested offline; see test_musicbrainz_live.py. musicbrainzngs is imported lazily so importing/constructing this class stays offline. """ def __init__(self, app_name: str = "Lyra", version: str = "0.1", contact: str = "lyra@localhost"): self._app = app_name self._version = version self._contact = contact def resolve(self, artist: str, album: str) -> MBTarget | None: import musicbrainzngs musicbrainzngs.set_useragent(self._app, self._version, self._contact) res = musicbrainzngs.search_release_groups(query=album, artist=artist, limit=5) groups = res.get("release-group-list", []) if not groups: return None rg = max(groups, key=lambda g: _ratio(album, g.get("title", ""))) if _ratio(album, rg.get("title", "")) < _MIN_TITLE_RATIO: return None canonical_album = rg.get("title", album) canonical_artist = artist credit = rg.get("artist-credit") if isinstance(credit, list) and credit and isinstance(credit[0], dict): canonical_artist = (credit[0].get("artist") or {}).get("name", artist) year = None frd = rg.get("first-release-date", "") or "" if frd[:4].isdigit(): year = int(frd[:4]) track_count = None total_duration_s = None rgid = rg.get("id") if rgid: rgfull = musicbrainzngs.get_release_group_by_id(rgid, includes=["releases"]) releases = rgfull.get("release-group", {}).get("release-list", []) if releases: rel = musicbrainzngs.get_release_by_id( releases[0]["id"], includes=["recordings"] ).get("release", {}) tracks = [] for medium in rel.get("medium-list", []): tracks.extend(medium.get("track-list", [])) if tracks: track_count = len(tracks) total_ms = sum(int((t.get("recording") or {}).get("length") or 0) for t in tracks) total_duration_s = total_ms // 1000 if total_ms else None return MBTarget( artist=canonical_artist, album=canonical_album, track_count=track_count, total_duration_s=total_duration_s, year=year, ) ``` - [ ] **Step 3: Write the opt-in live test** `worker/tests/test_musicbrainz_live.py`: ```python import os import pytest from lyra_worker._musicbrainz import MusicBrainzResolver pytestmark = pytest.mark.skipif( not os.environ.get("LYRA_LIVE_TESTS"), reason="live MusicBrainz test; set LYRA_LIVE_TESTS=1 to run", ) def test_live_resolves_continuum(): target = MusicBrainzResolver(contact="lyra-tests@localhost").resolve("John Mayer", "Continuum") assert target is not None assert target.track_count and target.track_count >= 10 # Continuum has 12 tracks assert "continuum" in target.album.casefold() assert target.year == 2006 ``` - [ ] **Step 4: Verify (offline) the module imports and the live test is skipped** Run: ```bash cd worker worker/.venv/bin/python -c "import sys; from lyra_worker._musicbrainz import MusicBrainzResolver; print('import ok; mb loaded:', 'musicbrainzngs' in sys.modules)" worker/.venv/bin/python -m pytest tests/test_musicbrainz_live.py -v ``` Expected: prints `import ok; mb loaded: False` (musicbrainzngs imported lazily, not at module import); the live test is reported **skipped**. Do NOT set LYRA_LIVE_TESTS here. - [ ] **Step 5: Commit** ```bash git add worker/lyra_worker/_musicbrainz.py worker/tests/test_musicbrainz_live.py worker/requirements.txt git commit -m "feat: add real MusicBrainz resolver and opt-in live test" ``` --- ### Task 4: Wire the resolver into the worker + end-to-end integration **Files:** - Modify: `worker/lyra_worker/registry.py` (add `build_resolver()`) - Modify: `worker/lyra_worker/main.py` (build + pass the resolver) - Test: `worker/tests/test_intake_pipeline.py` **Interfaces:** - Consumes: `MusicBrainzResolver` (Task 3), `run_pipeline` (Task 2). - Produces: `registry.build_resolver() -> MbResolver` returning a `MusicBrainzResolver()`; `main.run_forever()` builds it once and passes it to `run_pipeline`. - [ ] **Step 1: Write the failing test** `worker/tests/test_intake_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 lyra_worker.registry import build_resolver from lyra_worker.types import MBTarget from tests.conftest import insert_request from tests.test_qobuz_adapter import FakeQobuzClient def test_build_resolver_returns_a_resolver(): r = build_resolver() assert hasattr(r, "resolve") def test_enriched_target_flows_end_to_end(conn): class FakeResolver: def resolve(self, artist, album): return MBTarget(artist="John Mayer", album="Continuum", track_count=12, year=2006) # FakeQobuzClient defaults to a 12-track match and 12-track download -> imports cleanly. job_id = insert_request(conn, artist="j. mayer", album="continuum (deluxe)") claim_next(conn) run_pipeline(conn, job_id, [QobuzAdapter(FakeQobuzClient())], resolver=FakeResolver(), dest_root="/tmp/lib") with conn.cursor() as cur: cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,)) assert cur.fetchone()[0] == "imported" cur.execute('SELECT artist, album FROM "LibraryItem"') assert cur.fetchone() == ("John Mayer", "Continuum") # canonicalized, not the messy request ``` - [ ] **Step 2: Run the test to verify it fails** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest tests/test_intake_pipeline.py -v ``` Expected: FAIL — `build_resolver` does not exist in `registry`. - [ ] **Step 3: Add build_resolver and wire it into main** In `worker/lyra_worker/registry.py`, add: ```python from lyra_worker._musicbrainz import MusicBrainzResolver from lyra_worker.resolver import MbResolver def build_resolver() -> MbResolver: """The MusicBrainz resolver used to canonicalize requests in intake.""" return MusicBrainzResolver() ``` (Keep the existing `build_adapters`; add these alongside it.) In `worker/lyra_worker/main.py`, update `run_forever()` to build and pass the resolver: ```python def run_forever() -> None: conn = wait_for_db() adapters = build_adapters(get_config(conn)) resolver = build_resolver() 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, resolver=resolver) print(f"worker: finished job {job_id}", flush=True) finally: conn.close() ``` Update the import line to include `build_resolver`: `from lyra_worker.registry import build_adapters, build_resolver`. - [ ] **Step 4: Run the test to verify it passes** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest tests/test_intake_pipeline.py -v ``` Expected: PASS — both tests green. - [ ] **Step 5: Run the full worker suite** Run: ```bash cd worker export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra worker/.venv/bin/python -m pytest -q ``` Expected: PASS — every suite green, all live tests skipped, output pristine. - [ ] **Step 6: Commit** ```bash git add worker/lyra_worker/registry.py worker/lyra_worker/main.py worker/tests/test_intake_pipeline.py git commit -m "feat: wire MusicBrainz resolver into the worker loop" ``` --- ## Notes for the next plan (4b: tagging, import & persistent library) With intake now producing a canonical `MBTarget` (artist, album, `track_count`, `year`, `total_duration_s`), Plan 4b can: tag downloaded files with the canonical metadata (mutagen), organize them into `Artist/Album (Year)/## Title.ext` using `target.year`, and add the persistent `/music` volume mount to docker-compose (currently downloads land in ephemeral container storage). The completeness check now rejects partial downloads, so the tag/import stage can trust that an `imported` job has the full tracklist. The **live validation of `MusicBrainzResolver`** (Task 3) is a user step: run `LYRA_LIVE_TESTS=1 pytest tests/test_musicbrainz_live.py` and adjust the release-group/release traversal if the MB response shape differs. ```