docs: add staged-download + atomic-import implementation plan

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

docs/superpowers/plans/2026-07-11-lyra-staged-download-import.md

@@ -0,0 +1,434 @@
+# Lyra Staged Download + Atomic Import 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:** Stop downloads from writing into the live library. Download into a per-job staging dir; only a complete, verified album is promoted (clean audio + one cover) into `/music`, replacing any prior copy — so a partial/failed download or a retry can never corrupt or pollute the library.
+
+**Architecture:** New `library.py` helpers — `staging_dir`, `clear_staging_root`, `import_album` (clean audio-only move). `pipeline.py`'s download stage writes into the staging dir (cleared before each candidate); after the completeness check passes on staging, `import_album` moves the album into the final `album_dir` (replacing any existing folder), the tagger runs on the final dir, and a `try/finally` removes the staging dir whether the job succeeded or failed. `main.run_forever` sweeps orphaned staging dirs on startup.
+
+**Tech Stack:** Python 3.12 worker (`os`/`shutil`, psycopg). Tests: pytest — pure/fs helper tests with `tmp_path` (offline), plus pipeline tests with file-writing fake adapters (no real downloads).
+
+## Global Constraints
+
+- Python 3.12 worker. No new dependency (`os`, `shutil` are stdlib).
+- Adapters and the tagger are UNCHANGED — they still operate on whatever directory they are handed (now the staging dir for download; the final dir for tagging).
+- **Order matters (avoids library corruption AND minimizes test churn):** download → **verify on staging** → **promote to final via `import_album`** → **tag on final** → DB import. The library folder is created/touched ONLY after the completeness check passes, so a partial download never reaches it.
+- The final album folder must contain EXACTLY the album's audio files (by extension) plus at most one `cover.jpg` — no streamrip nesting, no non-audio artifacts.
+- The job's staging dir is removed in a `try/finally` around the download→import section (success and failure).
+- Backward compatibility: every existing worker test stays green. Only `test_tag_pipeline.py`'s file-path assertions change (the tagger now runs on the final dir, not the staging download dir); `test_tag_stage.py` is unaffected because the tagger still receives the final `album_dir`.
+- Every task ends with a commit.
+- Run worker commands from `/home/jonathan/Projects/lyra/worker` with `export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra` and `.venv/bin/python`.
+
+## Shared interfaces (defined by tasks below)
+
+```python
+# worker/lyra_worker/library.py (Task 1)
+def staging_dir(dest_root: str, job_id: str) -> str: ...      # "{dest_root}/.staging/{job_id}"
+def clear_staging_root(dest_root: str) -> None: ...           # rm -rf "{dest_root}/.staging"
+def import_album(staging: str, final: str) -> str: ...        # clean audio(+cover) move -> fresh final
+
+# worker/lyra_worker/pipeline.py (Task 2): download into staging_dir; verify; import_album -> final;
+#   tag on final; _import(..., final); try/finally clears the job's staging dir.
+# worker/lyra_worker/main.py (Task 3): clear_staging_root(DEST_ROOT) once before the claim loop.
+```
+
+---
+
+### Task 1: `library.py` staging helpers
+
+**Files:**
+- Modify: `worker/lyra_worker/library.py`
+- Test: `worker/tests/test_staging.py`
+
+**Interfaces:**
+- Consumes: nothing.
+- Produces: `staging_dir(dest_root, job_id)`, `clear_staging_root(dest_root)`, `import_album(staging, final) -> final`. `import_album` removes an existing `final`, creates a fresh one, moves every root-level audio file (by extension) from `staging` into it, then moves one cover image (`cover|folder|front.{jpg,jpeg,png}` found anywhere under staging) to `final/cover.jpg`; tolerates a missing `staging`.
+
+- [ ] **Step 1: Write the failing test**
+
+`worker/tests/test_staging.py`:
+```python
+import os
+
+from lyra_worker.library import clear_staging_root, import_album, staging_dir
+
+
+def test_staging_dir_path():
+    assert staging_dir("/music", "job-1") == "/music/.staging/job-1"
+
+
+def test_import_album_moves_audio_and_cover_into_fresh_final(tmp_path):
+    staging = tmp_path / "stg"
+    (staging / "sub").mkdir(parents=True)
+    (staging / "01 A.flac").write_bytes(b"a")
+    (staging / "02 B.flac").write_bytes(b"b")
+    (staging / "sub" / "cover.jpg").write_bytes(b"img")  # streamrip leaves art in a subfolder
+    (staging / "notes.txt").write_text("junk")
+    final = tmp_path / "lib" / "Artist" / "Album"
+
+    result = import_album(str(staging), str(final))
+
+    assert result == str(final)
+    assert sorted(os.listdir(final)) == ["01 A.flac", "02 B.flac", "cover.jpg"]  # audio + cover only
+
+
+def test_import_album_replaces_existing_final(tmp_path):
+    final = tmp_path / "lib" / "Artist" / "Album"
+    final.mkdir(parents=True)
+    (final / "old.flac").write_bytes(b"old")  # a stale lower-quality copy
+    staging = tmp_path / "stg"
+    staging.mkdir()
+    (staging / "01 New.flac").write_bytes(b"new")
+
+    import_album(str(staging), str(final))
+
+    assert sorted(os.listdir(final)) == ["01 New.flac"]  # old copy fully replaced
+
+
+def test_import_album_tolerates_missing_staging(tmp_path):
+    final = tmp_path / "Artist" / "Album"
+    import_album(str(tmp_path / "does-not-exist"), str(final))
+    assert os.path.isdir(final) and os.listdir(final) == []
+
+
+def test_clear_staging_root_removes_staging(tmp_path):
+    stg = tmp_path / ".staging" / "j1"
+    stg.mkdir(parents=True)
+    (stg / "x").write_bytes(b"")
+    clear_staging_root(str(tmp_path))
+    assert not (tmp_path / ".staging").exists()
+```
+
+- [ ] **Step 2: Run the test to verify it fails**
+
+```bash
+cd /home/jonathan/Projects/lyra/worker
+export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
+.venv/bin/python -m pytest tests/test_staging.py -v
+```
+Expected: FAIL — `ImportError: cannot import name 'staging_dir'`.
+
+- [ ] **Step 3: Implement the helpers**
+
+Append to `worker/lyra_worker/library.py` (keep the existing `safe_name`/`album_dir`; add the import at the top):
+```python
+import os
+import shutil
+
+_AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"}
+_COVER_STEMS = {"cover", "folder", "front"}
+_COVER_EXT = {".jpg", ".jpeg", ".png"}
+
+
+def staging_dir(dest_root: str, job_id: str) -> str:
+    """A per-job download staging directory, on the same filesystem as the library."""
+    return f"{dest_root}/.staging/{job_id}"
+
+
+def clear_staging_root(dest_root: str) -> None:
+    """Remove every staging dir (called at worker startup to sweep crash orphans)."""
+    shutil.rmtree(f"{dest_root}/.staging", ignore_errors=True)
+
+
+def _find_cover(staging: str) -> str | None:
+    for root, _dirs, files in os.walk(staging):
+        for name in sorted(files):
+            stem, ext = os.path.splitext(name)
+            if ext.lower() in _COVER_EXT and stem.lower() in _COVER_STEMS:
+                return os.path.join(root, name)
+    return None
+
+
+def import_album(staging: str, final: str) -> str:
+    """Move the album's audio files (+ one cover image) from `staging` into a fresh `final`,
+    replacing any existing folder there. Non-audio artifacts and nested subfolders are left
+    behind (the caller removes `staging`). Tolerates a missing `staging`. Returns `final`."""
+    if os.path.isdir(final):
+        shutil.rmtree(final)
+    os.makedirs(final, exist_ok=True)
+    if not os.path.isdir(staging):
+        return final
+    for name in sorted(os.listdir(staging)):
+        src = os.path.join(staging, name)
+        if os.path.isfile(src) and os.path.splitext(name)[1].lower() in _AUDIO_EXT:
+            shutil.move(src, os.path.join(final, name))
+    cover = _find_cover(staging)
+    if cover is not None:
+        shutil.move(cover, os.path.join(final, "cover.jpg"))
+    return final
+```
+
+- [ ] **Step 4: Run the test to verify it passes**
+
+```bash
+cd /home/jonathan/Projects/lyra/worker
+export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
+.venv/bin/python -m pytest tests/test_staging.py -v
+```
+Expected: PASS — all five tests green.
+
+- [ ] **Step 5: Run the full worker suite (no regressions)**
+
+```bash
+cd /home/jonathan/Projects/lyra/worker
+export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
+.venv/bin/python -m pytest -q
+```
+Expected: PASS — the new module-level `import os`/`import shutil` in `library.py` doesn't disturb `safe_name`/`album_dir`; every existing test still green.
+
+- [ ] **Step 6: Commit**
+
+```bash
+cd /home/jonathan/Projects/lyra
+git add worker/lyra_worker/library.py worker/tests/test_staging.py
+git commit -m "feat: staging-dir + clean import_album library helpers"
+```
+
+---
+
+### Task 2: Pipeline — download to staging, promote to library, isolate failures
+
+**Files:**
+- Modify: `worker/lyra_worker/pipeline.py`
+- Modify: `worker/tests/test_tag_pipeline.py` (tagger now runs on the final dir; file promoted; staging cleaned)
+- Test: `worker/tests/test_staging_pipeline.py` (partial-then-retry regression + library-untouched-on-failure)
+
+**Interfaces:**
+- Consumes: `staging_dir`, `import_album` (Task 1); existing `album_dir`, `_import`, `_fail`, adapters, tagger.
+- Produces: `run_pipeline` downloads into `staging_dir(dest_root, job_id)` (cleared before each candidate), verifies completeness on staging, `import_album(staging, final)` into `album_dir`, tags the final dir, `_import(..., final)`, and always removes the job's staging dir.
+
+- [ ] **Step 1: Write the failing regression test**
+
+`worker/tests/test_staging_pipeline.py`:
+```python
+import os
+
+from lyra_worker.claim import claim_next
+from lyra_worker.pipeline import run_pipeline
+from lyra_worker.types import Candidate, DownloadResult, Quality
+from tests.conftest import insert_request
+
+_Q = Quality(fmt="FLAC", lossless=True, bit_depth=24, sample_rate=96000)
+
+
+class FlakyAdapter:
+    """Writes a partial (1-track, oddly-named) set on the first download, the full 3-track set
+    after. The first-attempt file is named UNLIKE the real tracks — mirroring streamrip's
+    inconsistent per-attempt naming, so an un-isolated download would leave a visible orphan."""
+
+    name = "qobuz"
+    tier = 0
+
+    def __init__(self):
+        self.calls = 0
+
+    def health(self):
+        return True
+
+    def search(self, target):
+        return [Candidate(source="qobuz", source_ref="r", matched_artist=target.artist,
+                          matched_album=target.album, quality=_Q, track_count=3, source_tier=0)]
+
+    def download(self, candidate, dest, on_progress):
+        self.calls += 1
+        os.makedirs(dest, exist_ok=True)
+        if self.calls == 1:
+            open(os.path.join(dest, "orphan-a.flac"), "wb").close()  # partial, distinct name
+            return DownloadResult(ok=True, path=dest, track_count=1)
+        for i in range(3):
+            open(os.path.join(dest, f"{i} track.flac"), "wb").close()
+        return DownloadResult(ok=True, path=dest, track_count=3)
+
+
+def test_partial_download_leaves_library_untouched(conn, tmp_path):
+    adapter = FlakyAdapter()  # first call is partial
+    job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
+    claim_next(conn)
+    run_pipeline(conn, job_id, [adapter], dest_root=str(tmp_path))
+
+    with conn.cursor() as cur:
+        cur.execute('SELECT state FROM "Job" WHERE id = %s', (job_id,))
+        assert cur.fetchone()[0] == "needs_attention"  # completeness check rejected the partial
+    assert not (tmp_path / "Radiohead" / "In Rainbows").exists()  # library never touched
+    staging_root = tmp_path / ".staging"
+    assert not staging_root.exists() or list(staging_root.iterdir()) == []  # staging cleaned
+
+
+def test_retry_imports_clean_with_no_orphans(conn, tmp_path):
+    adapter = FlakyAdapter()
+    # attempt 1: partial -> needs_attention (no library write)
+    job1 = insert_request(conn, artist="Radiohead", album="In Rainbows")
+    claim_next(conn)
+    run_pipeline(conn, job1, [adapter], dest_root=str(tmp_path))
+    # attempt 2: full -> imported
+    job2 = insert_request(conn, artist="Radiohead", album="In Rainbows")
+    claim_next(conn)
+    run_pipeline(conn, job2, [adapter], dest_root=str(tmp_path))
+
+    with conn.cursor() as cur:
+        cur.execute('SELECT state FROM "Job" WHERE id = %s', (job2,))
+        assert cur.fetchone()[0] == "imported"
+    final = tmp_path / "Radiohead" / "In Rainbows"
+    # exactly the 3 real tracks — the attempt-1 "orphan-a.flac" is NOT present
+    assert sorted(os.listdir(final)) == ["0 track.flac", "1 track.flac", "2 track.flac"]
+```
+
+- [ ] **Step 2: Run it to verify it fails**
+
+```bash
+cd /home/jonathan/Projects/lyra/worker
+export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
+.venv/bin/python -m pytest tests/test_staging_pipeline.py -v
+```
+Expected: FAIL — against the current direct-to-library pipeline, `test_partial_download_leaves_library_untouched` fails (the partial download wrote into `.../Radiohead/In Rainbows`) and `test_retry_imports_clean_with_no_orphans` fails (the folder also contains `orphan-a.flac`).
+
+- [ ] **Step 3: Rewrite the download→import section of `run_pipeline`**
+
+In `worker/lyra_worker/pipeline.py`, update the import line:
+```python
+from lyra_worker.library import album_dir, import_album, staging_dir
+```
+and add `import shutil` at the top (with the other imports).
+
+Replace everything from the `# 4. download (fall-through)` comment through the end of the function (the download, tag, and import blocks) with:
+```python
+    # 4. download (fall-through) into an isolated per-job staging dir
+    _set_state(conn, job_id, "downloading", "download")
+    by_source = {a.name: a for a in adapters}
+    staging = staging_dir(dest_root, job_id)
+    winner = None
+    result = None
+    try:
+        for candidate in ranked:
+            adapter = by_source.get(candidate.source)
+            if adapter is None:
+                continue
+            shutil.rmtree(staging, ignore_errors=True)  # clean slate per attempt
+            result = adapter.download(candidate, staging, lambda _pct: None)
+            if result.ok:
+                winner = candidate
+                _mark_chosen(conn, job_id, candidate.source_ref)
+                break
+        if winner is None or result is None or not result.ok:
+            _fail(conn, job_id, "all downloads failed")
+            return
+
+        # 5. verify completeness on the staging copy, then promote + tag
+        _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
+
+        final = album_dir(dest_root, target)
+        import_album(staging, final)  # clean audio(+cover) move into the library, replacing any prior copy
+
+        if tagger is not None:
+            try:
+                tagger.tag_album(final, 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 (DB)
+        _set_state(conn, job_id, "imported", "import")
+        _import(conn, job_id, target, winner, final)
+    finally:
+        shutil.rmtree(staging, ignore_errors=True)  # a partial/failed download never persists
+```
+
+- [ ] **Step 4: Run the regression test to verify it passes**
+
+```bash
+cd /home/jonathan/Projects/lyra/worker
+export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
+.venv/bin/python -m pytest tests/test_staging_pipeline.py -v
+```
+Expected: PASS — partial → needs_attention + untouched library + cleaned staging; retry → imported with exactly the 3 real tracks, no orphan.
+
+- [ ] **Step 5: Update `test_tag_pipeline.py` for the new tag target**
+
+In `worker/tests/test_tag_pipeline.py`, the tagger now runs on the FINAL library dir (not the staging download dir), and the downloaded file is promoted into it. Replace the two trailing assertions (currently `assert tagger.seen == written["dir"]` and `assert os.path.isdir(tagger.seen)`) with:
+```python
+    final = os.path.join(str(tmp_path), "John Mayer", "Continuum")
+    assert tagger.seen == final                                  # tagger runs on the final library dir
+    assert os.path.isfile(os.path.join(final, "track.opus"))     # the downloaded file was promoted into it
+    assert not os.path.exists(written["dir"])                    # the staging download dir was cleaned up
+```
+(The fake writes `track.opus` — a real audio extension — so `import_album` moves it into `final`.)
+
+- [ ] **Step 6: Run the full worker suite**
+
+```bash
+cd /home/jonathan/Projects/lyra/worker
+export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
+.venv/bin/python -m pytest -q
+```
+Expected: PASS — all suites green (`test_tag_stage.py` is unaffected because the tagger still receives `album_dir`; only `test_tag_pipeline.py` needed the path-assertion update), live tests skipped, output pristine.
+
+Note (expected, not a regression): the other pipeline tests (`test_pipeline.py`, `test_qobuz_pipeline.py`, etc.) use fake adapters that don't write files and pass `dest_root="/tmp/lib"`. `import_album` now always creates the (empty) final dir, so these runs leave harmless empty `/tmp/lib/<Artist>/<Album>` directories. They still assert only on job state / DB rows, so they stay green. Do NOT rewrite them to `tmp_path` in this task — that's out of scope churn.
+
+- [ ] **Step 7: Commit**
+
+```bash
+cd /home/jonathan/Projects/lyra
+git add worker/lyra_worker/pipeline.py worker/tests/test_staging_pipeline.py worker/tests/test_tag_pipeline.py
+git commit -m "feat: download to staging + atomically promote complete album into the library"
+```
+
+---
+
+### Task 3: Sweep orphaned staging dirs on worker startup
+
+**Files:**
+- Modify: `worker/lyra_worker/main.py`
+- Test: `worker/tests/test_staging.py` (add a startup-sweep behavior note via the helper — already covered)
+
+**Interfaces:**
+- Consumes: `clear_staging_root` (Task 1).
+- Produces: `run_forever()` calls `clear_staging_root(DEST_ROOT)` once before the claim loop, where `DEST_ROOT` matches `run_pipeline`'s default library root (`"/music"`).
+
+- [ ] **Step 1: Add the startup sweep**
+
+In `worker/lyra_worker/main.py`, add the import and a module constant, and call the sweep once after `wait_for_db()` and before the `while True` loop:
+```python
+from lyra_worker.library import clear_staging_root
+```
+Add near the other module constants (after `MONITOR_TICK_SECONDS`):
+```python
+DEST_ROOT = "/music"  # must match run_pipeline's default library root
+```
+In `run_forever()`, immediately after `conn = wait_for_db()`:
+```python
+    clear_staging_root(DEST_ROOT)  # sweep any staging dirs orphaned by a prior crash
+```
+
+- [ ] **Step 2: Verify import stays offline and the suite is green**
+
+```bash
+cd /home/jonathan/Projects/lyra/worker
+export DATABASE_URL=postgresql://lyra:lyra@localhost:5432/lyra
+.venv/bin/python -c "from lyra_worker import main; print('main imports ok; DEST_ROOT =', main.DEST_ROOT)"
+.venv/bin/python -m pytest -q
+```
+Expected: prints `main imports ok; DEST_ROOT = /music`; full suite passes (the `clear_staging_root` helper's own behavior is covered by `test_staging.py::test_clear_staging_root_removes_staging`).
+
+- [ ] **Step 3: Commit**
+
+```bash
+cd /home/jonathan/Projects/lyra
+git add worker/lyra_worker/main.py
+git commit -m "feat: sweep orphaned staging dirs on worker startup"
+```
+
+---
+
+## Notes: pipeline hardened
+
+Downloads are now isolated: they land in `{/music}/.staging/{job_id}`, are verified there, and only a
+complete album is promoted — clean audio plus one `cover.jpg` — into the library, replacing any prior
+copy; partial/failed downloads and retries leave the library untouched and self-clean. This closes the
+corruption exposed by the monitor's retry (the *In Rainbows* case) and the leftover-`__artwork` nesting.
+After this lands: re-enable the monitor (`monitor.enabled=true`), re-grab *In Rainbows* (already reset to
+`wanted`), and confirm a clean 10-track import. Remaining deferred hardening (crash-resume mid-pipeline,
+per-track duration matching, cover-art embedding, same-quality upgrade skip) is unchanged and out of scope.
+```