Lyra/docs/superpowers/plans/2026-07-10-lyra-musicbrainz-intake.md

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)

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

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:

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

    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:

    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:

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:

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

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

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

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:

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:

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:

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:

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

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

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:

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

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:

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:

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:

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

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:

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:

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:

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:

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:

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:

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

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.