From 1d847e2aebe63d645d757f9e3f5ebb8c5fb66c1f Mon Sep 17 00:00:00 2001 From: Jonathan Date: Sat, 11 Jul 2026 15:07:01 +0200 Subject: [PATCH] docs: add staged-download + atomic-import implementation plan Co-Authored-By: Claude Opus 4.8 (1M context) --- .../2026-07-11-lyra-staged-download-import.md | 434 ++++++++++++++++++ 1 file changed, 434 insertions(+) create mode 100644 docs/superpowers/plans/2026-07-11-lyra-staged-download-import.md diff --git a/docs/superpowers/plans/2026-07-11-lyra-staged-download-import.md b/docs/superpowers/plans/2026-07-11-lyra-staged-download-import.md new file mode 100644 index 0000000..edd1815 --- /dev/null +++ b/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//` 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. +```