feat: track download progress on Job (migration + worker poller + API)

Add Job.downloadProgress (0.0-1.0), derived from a worker background
poller that counts audio files landing in staging / expected track
count, since streamrip only reports 0/1. No UI yet (slice B).

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

worker/lyra_worker/pipeline.py

@@ -1,4 +1,6 @@
+import os
 import shutil
+import threading
 from dataclasses import replace
 from typing import Sequence
 
@@ -11,6 +13,39 @@ from lyra_worker.quality import quality_class
 from lyra_worker.ranker import rank_candidates
 from lyra_worker.types import Candidate, MBTarget
 
+_AUDIO_EXT = {".flac", ".mp3", ".m4a", ".opus", ".ogg", ".aac", ".wav"}
+
+
+def _count_staged_audio(staging: str) -> int:
+    """Count audio files anywhere under the staging dir (streamrip nests them in a subfolder)."""
+    n = 0
+    for _root, _dirs, files in os.walk(staging):
+        for f in files:
+            if os.path.splitext(f)[1].lower() in _AUDIO_EXT:
+                n += 1
+    return n
+
+
+def _poll_download_progress(job_id: str, staging: str, expected: int, stop: "threading.Event") -> None:
+    """Until `stop`, periodically write Job.downloadProgress = files-in-staging / expected
+    (capped 0.99). Uses its own short-lived connection (psycopg conns aren't shareable across
+    threads). A missing DSN or expected<=0 just no-ops."""
+    dsn = os.environ.get("DATABASE_URL")
+    if not dsn or expected <= 0:
+        return
+    conn = psycopg.connect(dsn)
+    try:
+        while not stop.is_set():
+            frac = min(_count_staged_audio(staging) / expected, 0.99)
+            with conn.cursor() as cur:
+                cur.execute('UPDATE "Job" SET "downloadProgress" = %s WHERE id = %s', (frac, job_id))
+            conn.commit()
+            stop.wait(1.5)
+    except Exception as e:  # a progress poller must never affect the job
+        print(f"pipeline: download progress poller error: {e}", flush=True)
+    finally:
+        conn.close()
+
 
 def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) -> None:
     with conn.cursor() as cur:
@@ -21,6 +56,12 @@ def _set_state(conn: psycopg.Connection, job_id: str, state: str, stage: str) ->
     conn.commit()
 
 
+def _set_download_progress(conn: psycopg.Connection, job_id: str, frac: float) -> None:
+    with conn.cursor() as cur:
+        cur.execute('UPDATE "Job" SET "downloadProgress" = %s WHERE id = %s', (frac, job_id))
+    conn.commit()
+
+
 def _request_id(conn: psycopg.Connection, job_id: str) -> str:
     with conn.cursor() as cur:
         cur.execute('SELECT "requestId" FROM "Job" WHERE id = %s', (job_id,))
@@ -201,20 +242,32 @@ def run_pipeline(
     staging = staging_dir(staging_root or f"{dest_root}/.staging", job_id)
     winner = None
     result = None
+    _set_download_progress(conn, job_id, 0.0)  # reset for this run
+    expected = target.track_count or (ranked[0].track_count if ranked else 0)
+    _stop = threading.Event()
+    _poller = threading.Thread(
+        target=_poll_download_progress, args=(job_id, staging, expected, _stop), daemon=True
+    )
+    _poller.start()
     try:
-        for candidate in ranked:
-            adapter = by_source.get(candidate.source)
-            if adapter is None:
-                continue
-            _mark_chosen(conn, job_id, candidate.source_ref)  # reflect the source now in flight
-            shutil.rmtree(staging, ignore_errors=True)  # clean slate per attempt
-            result = adapter.download(candidate, staging, lambda _pct: None)
-            if result.ok:
-                winner = candidate
-                break
+        try:
+            for candidate in ranked:
+                adapter = by_source.get(candidate.source)
+                if adapter is None:
+                    continue
+                _mark_chosen(conn, job_id, candidate.source_ref)  # reflect the source now in flight
+                shutil.rmtree(staging, ignore_errors=True)  # clean slate per attempt
+                result = adapter.download(candidate, staging, lambda _pct: None)
+                if result.ok:
+                    winner = candidate
+                    break
+        finally:
+            _stop.set()
+            _poller.join(timeout=3)
         if winner is None or result is None or not result.ok:
             _fail(conn, job_id, "all downloads failed")
             return
+        _set_download_progress(conn, job_id, 1.0)  # download done — UI shows 100% into Finishing
 
         # 5. verify completeness on the staging copy, then promote + tag
         _set_state(conn, job_id, "tagging", "tag")

worker/tests/test_download_progress.py

@@ -0,0 +1,22 @@
+import os
+
+from lyra_worker.pipeline import _count_staged_audio
+
+
+def test_count_staged_audio_counts_nested_audio_files_only(tmp_path):
+    nested = tmp_path / "Artist" / "Album"
+    nested.mkdir(parents=True)
+    (nested / "01.flac").write_bytes(b"")
+    (nested / "02.flac").write_bytes(b"")
+    (nested / "03.flac").write_bytes(b"")
+    (nested / "cover.jpg").write_bytes(b"")
+
+    assert _count_staged_audio(str(tmp_path)) == 3
+
+
+def test_count_staged_audio_empty_dir_is_zero(tmp_path):
+    assert _count_staged_audio(str(tmp_path)) == 0
+
+
+def test_count_staged_audio_missing_dir_is_zero():
+    assert _count_staged_audio("/nonexistent/path/for/sure") == 0

worker/tests/test_pipeline.py

@@ -19,6 +19,12 @@ def _request_status(conn, job_id):
         return cur.fetchone()[0]
 
 
+def _download_progress(conn, job_id):
+    with conn.cursor() as cur:
+        cur.execute('SELECT "downloadProgress" FROM "Job" WHERE id = %s', (job_id,))
+        return cur.fetchone()[0]
+
+
 def test_success_picks_best_source_and_imports(conn):
     job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
     claim_next(conn)
@@ -26,6 +32,7 @@ def test_success_picks_best_source_and_imports(conn):
 
     assert _job_state(conn, job_id) == ("imported", "import")
     assert _request_status(conn, job_id) == "completed"
+    assert _download_progress(conn, job_id) == 1.0  # a successful download completes at 100%
     with conn.cursor() as cur:
         cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
         assert cur.fetchone()[0] == "qobuz"  # highest quality won
@@ -41,6 +48,7 @@ def test_falls_through_when_best_download_fails(conn):
     run_pipeline(conn, job_id, [FailingAdapter(), FakeSoulseek()], dest_root="/tmp/lib")
 
     assert _job_state(conn, job_id) == ("imported", "import")
+    assert _download_progress(conn, job_id) == 1.0  # falls through, but still completes at 100%
     with conn.cursor() as cur:
         cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
         assert cur.fetchone()[0] == "soulseek"
@@ -75,6 +83,7 @@ def test_all_downloads_fail_goes_to_needs_attention(conn):
     run_pipeline(conn, job_id, [FailingAdapter()], dest_root="/tmp/lib")
 
     assert _job_state(conn, job_id)[0] == "needs_attention"
+    assert _download_progress(conn, job_id) == 0.0  # never reached 1.0 — all attempts failed
 
 
 def test_incomplete_download_goes_to_needs_attention(conn):