feat: Tagger seam and pipeline tag-stage integration

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

worker/lyra_worker/pipeline.py

@@ -103,6 +103,7 @@ def run_pipeline(
     job_id: str,
     adapters: Sequence[SourceAdapter],
     resolver=None,
+    tagger=None,
     min_confidence: float = 0.7,
     dest_root: str = "/music",
 ) -> None:
@@ -174,6 +175,12 @@ def run_pipeline(
         _fail(conn, job_id, "incomplete download")
         return
 
+    if tagger is not None:
+        try:
+            tagger.tag_album(dest, target)
+        except Exception as e:  # a tagging failure must not discard a good download
+            print(f"pipeline: tagging failed for job {job_id}: {e}", flush=True)
+
     # 6. import
     _set_state(conn, job_id, "imported", "import")
     _import(conn, job_id, target, winner, result.path or dest)

worker/lyra_worker/tagger.py

@@ -0,0 +1,9 @@
+from typing import Protocol
+
+from lyra_worker.types import MBTarget
+
+
+class Tagger(Protocol):
+    def tag_album(self, album_dir: str, target: MBTarget) -> None:
+        """Tag the album's audio files with target's metadata and rename them."""
+        ...

worker/lyra_worker/types.py

@@ -8,6 +8,7 @@ class MBTarget:
     track_count: int | None = None
     total_duration_s: int | None = None
     year: int | None = None
+    tracklist: tuple[str, ...] = ()
 
 
 @dataclass(frozen=True)

worker/tests/test_tag_stage.py

@@ -0,0 +1,52 @@
+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
+
+
+class RecordingTagger:
+    def __init__(self, fail=False):
+        self.calls = []
+        self._fail = fail
+
+    def tag_album(self, album_dir, target):
+        self.calls.append((album_dir, target))
+        if self._fail:
+            raise RuntimeError("tagging blew up")
+
+
+def test_tagger_called_with_album_dir_and_target(conn):
+    tagger = RecordingTagger()
+    job_id = insert_request(conn, artist="John Mayer", album="Continuum")
+    claim_next(conn)
+    run_pipeline(conn, job_id, [QobuzAdapter(FakeQobuzClient())], tagger=tagger, 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"
+    assert len(tagger.calls) == 1
+    album_dir, target = tagger.calls[0]
+    assert album_dir == "/tmp/lib/John Mayer/Continuum"
+    assert target.album == "Continuum"
+
+
+def test_tagger_failure_does_not_fail_the_job(conn):
+    tagger = RecordingTagger(fail=True)
+    job_id = insert_request(conn, artist="John Mayer", album="Continuum")
+    claim_next(conn)
+    run_pipeline(conn, job_id, [QobuzAdapter(FakeQobuzClient())], tagger=tagger, 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"  # tagging error is logged, not fatal
+
+
+def test_no_tagger_still_imports(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")  # no tagger
+
+    with conn.cursor() as cur:
+        cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,))
+        assert cur.fetchone()[0] == "imported"