feat(web): three-phase progress with live download % on The Floor

Replace the misleading 6-micro-step progress bar with three honest
phases (Searching -> Downloading -> Finishing). Downloading now shows
a live determinate percentage sourced from job.downloadProgress (Slice
A), with a "NN% · done/total tracks" caption when the track count is
known. Searching and Finishing stay indeterminate.

- status.ts: describeJob collapses to {label, kind}; drop step/totalSteps
  and the STAGES array.
- progress-bar.tsx: ProgressBar takes {kind, indeterminate?, percent?,
  caption?} instead of step/total/stageLabel.
- job-row.tsx: JobRow takes a `bar?: ReactNode` slot instead of
  step/total/indeterminate/stageLabel props.
- queue.tsx: builds the per-state bar + meta; downloading meta drops
  the track count (now shown in the bar). Retry button + error note
  from the retry feature preserved exactly.

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

web/prisma/migrations/20260713220000_add_job_download_progress/migration.sql

@@ -0,0 +1,2 @@
+-- AlterTable
+ALTER TABLE "Job" ADD COLUMN "downloadProgress" DOUBLE PRECISION NOT NULL DEFAULT 0;

web/prisma/schema.prisma

@@ -43,17 +43,18 @@ model Request {
 }
 
 model Job {
-  id           String      @id @default(cuid())
-  request      Request     @relation(fields: [requestId], references: [id], onDelete: Cascade)
-  requestId    String      @unique
-  state        JobState    @default(requested)
-  currentStage String      @default("intake")
-  attempts     Int         @default(0)
-  error        String?
-  claimedAt    DateTime?
-  createdAt    DateTime    @default(now())
-  updatedAt    DateTime    @updatedAt
-  candidates   Candidate[]
+  id               String      @id @default(cuid())
+  request          Request     @relation(fields: [requestId], references: [id], onDelete: Cascade)
+  requestId        String      @unique
+  state            JobState    @default(requested)
+  currentStage     String      @default("intake")
+  attempts         Int         @default(0)
+  error            String?
+  claimedAt        DateTime?
+  createdAt        DateTime    @default(now())
+  updatedAt        DateTime    @updatedAt
+  downloadProgress Float       @default(0)
+  candidates       Candidate[]
 }
 
 model Candidate {

web/src/app/_ui/job-row.tsx

@@ -1,6 +1,5 @@
 import type { ReactNode } from "react";
 import type { Kind } from "./status";
-import { ProgressBar } from "./progress-bar";
 import { StatusChip } from "./status-chip";
 
 // Map the severity kind to the row's visual class (stripe/bar color).
@@ -15,23 +14,17 @@ export function JobRow({
   album,
   kind,
   label,
-  step,
-  total,
   meta,
   note,
-  indeterminate,
-  stageLabel,
+  bar,
 }: {
   artist: string;
   album: string;
   kind: Kind;
   label: string;
-  step: number;
-  total: number;
   meta?: ReactNode;
   note?: ReactNode;
-  indeterminate?: boolean;
-  stageLabel?: string;
+  bar?: ReactNode;
 }) {
   return (
     <article className={`job ${rowClass(kind)}`}>
@@ -41,11 +34,7 @@ export function JobRow({
           {album} <span className="artist">· {artist}</span>
         </h3>
         {meta ? <div className="meta">{meta}</div> : null}
-        {note ? (
-          <div className="note">{note}</div>
-        ) : kind !== "attention" ? (
-          <ProgressBar kind={kind} step={step} total={total} indeterminate={indeterminate} stageLabel={stageLabel} />
-        ) : null}
+        {note ? <div className="note">{note}</div> : kind !== "attention" ? bar : null}
       </div>
       <StatusChip kind={kind} label={label} />
     </article>

web/src/app/_ui/progress-bar.tsx

@@ -1,19 +1,17 @@
 import type { Kind } from "./status";
 
-/** A thin ruled progress bar. `indeterminate` for open-ended stages (searching);
- *  otherwise a discrete step/total fill with a step readout. */
+/** A thin ruled progress bar. `indeterminate` for open-ended phases (searching,
+ *  finishing); otherwise a determinate fill to `percent` with a live readout. */
 export function ProgressBar({
   kind,
-  step,
-  total,
   indeterminate,
-  stageLabel,
+  percent,
+  caption,
 }: {
   kind: Kind;
-  step: number;
-  total: number;
   indeterminate?: boolean;
-  stageLabel?: string;
+  percent?: number;
+  caption?: string;
 }) {
   if (indeterminate) {
     return (
@@ -21,17 +19,17 @@ export function ProgressBar({
         <div className="bar indet">
           <span />
         </div>
-        <div className="pct muted">searching</div>
+        <div className="pct muted">{caption ?? "…"}</div>
       </div>
     );
   }
-  const pct = total > 0 ? Math.round((step / total) * 100) : 0;
+  const pct = Math.min(100, Math.max(0, percent ?? 0));
   return (
     <div className="prog">
       <div className="bar">
         <span style={{ width: `${pct}%` }} />
       </div>
-      <div className="pct">{stageLabel ? `${stageLabel} · step ${step} of ${total}` : `${step}/${total}`}</div>
+      <div className="pct">{caption ?? `${Math.round(percent ?? 0)}%`}</div>
     </div>
   );
 }

web/src/app/_ui/status.test.ts

@@ -5,21 +5,16 @@ describe("describeJob", () => {
   it("maps pipeline states to literal labels + severity kind", () => {
     expect(describeJob("requested", "intake")).toMatchObject({ label: "Queued", kind: "idle" });
     expect(describeJob("matching", "match")).toMatchObject({ label: "Searching", kind: "working" });
+    expect(describeJob("matched", "match")).toMatchObject({ label: "Searching", kind: "working" });
     expect(describeJob("downloading", "download")).toMatchObject({ label: "Downloading", kind: "working" });
-    expect(describeJob("tagging", "tag")).toMatchObject({ label: "Tagging", kind: "verify" });
+    expect(describeJob("tagging", "tag")).toMatchObject({ label: "Finishing", kind: "verify" });
     expect(describeJob("imported", "import")).toMatchObject({ label: "Pressed", kind: "done" });
     expect(describeJob("needs_attention", "download")).toMatchObject({ label: "Needs attention", kind: "attention" });
   });
 
-  it("derives a stepped progress from the pipeline stage", () => {
-    expect(describeJob("downloading", "download")).toMatchObject({ step: 4, totalSteps: 6 });
-    expect(describeJob("matching", "intake")).toMatchObject({ step: 1, totalSteps: 6 });
-  });
-
-  it("falls back gracefully for unknown states/stages", () => {
+  it("falls back gracefully for unknown states", () => {
     expect(describeJob("weird", "intake").kind).toBe("idle");
     expect(describeJob("weird", "intake").label).toBe("Weird");
-    expect(describeJob("downloading", "???").step).toBe(0);
   });
 });
 

web/src/app/_ui/status.ts

@@ -1,16 +1,14 @@
 export type Kind = "idle" | "working" | "verify" | "done" | "attention";
 
-// Pipeline stage order (worker Job.currentStage), used to derive a stepped progress.
-const STAGES = ["intake", "match", "rank", "download", "tag", "import"];
-
-// Worker Job.state -> a literal, legible label + severity kind. The record-label
-// metaphor stays in the chrome; these labels never obscure what's happening.
+// Worker Job.state -> a literal, legible label + severity kind, collapsed to three
+// honest phases: Searching -> Downloading -> Finishing. The record-label metaphor
+// stays in the chrome; these labels never obscure what's happening.
 const STATE_MAP: Record<string, { label: string; kind: Kind }> = {
   requested: { label: "Queued", kind: "idle" },
   matching: { label: "Searching", kind: "working" },
-  matched: { label: "Matched", kind: "working" },
+  matched: { label: "Searching", kind: "working" },
   downloading: { label: "Downloading", kind: "working" },
-  tagging: { label: "Tagging", kind: "verify" },
+  tagging: { label: "Finishing", kind: "verify" },
   imported: { label: "Pressed", kind: "done" },
   needs_attention: { label: "Needs attention", kind: "attention" },
 };
@@ -29,16 +27,12 @@ export function timeAgo(value: string | Date | null | undefined, now: number): s
   return `${Math.floor(h / 24)}d`;
 }
 
-export function describeJob(
-  state: string,
-  stage: string,
-): { label: string; kind: Kind; step: number; totalSteps: number } {
-  const base =
-    STATE_MAP[state] ??
-    ({ label: state ? state[0].toUpperCase() + state.slice(1) : "Unknown", kind: "idle" } as {
-      label: string;
-      kind: Kind;
-    });
-  const idx = STAGES.indexOf(stage);
-  return { ...base, step: idx >= 0 ? idx + 1 : 0, totalSteps: STAGES.length };
+export function describeJob(state: string, stage: string): { label: string; kind: Kind } {
+  void stage; // stage no longer drives the label; kept for call-site compatibility
+  return (
+    STATE_MAP[state] ?? {
+      label: state ? state[0].toUpperCase() + state.slice(1) : "Unknown",
+      kind: "idle",
+    }
+  );
 }

web/src/app/api/requests/route.test.ts

@@ -62,7 +62,15 @@ describe("requests API", () => {
       data: {
         artist: "Chosen Artist",
         album: "Chosen Album",
-        job: { create: { state: "downloading", currentStage: "download", attempts: 2, error: "boom" } },
+        job: {
+          create: {
+            state: "downloading",
+            currentStage: "download",
+            attempts: 2,
+            error: "boom",
+            downloadProgress: 0.42,
+          },
+        },
       },
       include: { job: true },
     });
@@ -99,6 +107,7 @@ describe("requests API", () => {
     expect(found.job.chosen).toEqual({ source: "qobuz", format: "flac", trackCount: 12 });
     expect(found.job.error).toBe("boom");
     expect(found.job.attempts).toBe(2);
+    expect(found.job.downloadProgress).toBe(0.42);
   });
 
   it("reports no chosen candidate and zero count when a job has none", async () => {
@@ -109,5 +118,6 @@ describe("requests API", () => {
     const found = body.requests.find((r: { id: string }) => r.id === created.id);
     expect(found.job.candidateCount).toBe(0);
     expect(found.job.chosen).toBeNull();
+    expect(found.job.downloadProgress).toBe(0);
   });
 });

web/src/app/api/requests/route.ts

@@ -63,6 +63,7 @@ export async function GET() {
               error: r.job.error,
               claimedAt: r.job.claimedAt,
               updatedAt: r.job.updatedAt,
+              downloadProgress: r.job.downloadProgress,
               candidateCount: cands.length,
               chosen: chosen ? { source: chosen.source, format: chosen.format, trackCount: chosen.trackCount } : null,
             };

web/src/app/queue.tsx

@@ -5,6 +5,7 @@ import { describeJob, timeAgo } from "./_ui/status";
 import { StatTiles } from "./_ui/stat-tiles";
 import { SectionHeader } from "./_ui/section-header";
 import { JobRow } from "./_ui/job-row";
+import { ProgressBar } from "./_ui/progress-bar";
 import { PressedList } from "./_ui/pressed-list";
 import { toast } from "./_ui/toast";
 
@@ -24,6 +25,7 @@ type Row = {
     updatedAt: string | null;
     candidateCount: number;
     chosen: { source: string; format: string; trackCount: number } | null;
+    downloadProgress: number;
   } | null;
 };
 
@@ -38,6 +40,7 @@ function jobOf(r: Row) {
       updatedAt: null,
       candidateCount: 0,
       chosen: null,
+      downloadProgress: 0,
     }
   );
 }
@@ -139,7 +142,7 @@ export function Queue() {
               let meta: string | undefined;
               if (j.state === "downloading" || j.state === "tagging") {
                 const parts: string[] = [];
-                if (j.chosen) parts.push(`${j.chosen.source} · ${j.chosen.format} · ${j.chosen.trackCount} tracks`);
+                if (j.chosen) parts.push(`${j.chosen.source} · ${j.chosen.format}`);
                 const elapsed = timeAgo(j.claimedAt ?? j.updatedAt, now);
                 if (elapsed) parts.push(elapsed);
                 meta = parts.join(" · ") || undefined;
@@ -151,8 +154,6 @@ export function Queue() {
               } else if (j.state === "requested") {
                 const elapsed = timeAgo(r.createdAt, now);
                 meta = elapsed ? `In the queue · ${elapsed}` : "In the queue";
-              } else if (j.state === "needs_attention") {
-                meta = undefined;
               } else {
                 meta = undefined;
               }
@@ -166,8 +167,25 @@ export function Queue() {
                 </>
               ) : undefined;
 
-              const stageLabel =
-                attention || j.state === "matching" || j.state === "matched" ? undefined : j.currentStage;
+              let bar: React.ReactNode;
+              if (j.state === "matching" || j.state === "matched") {
+                bar = <ProgressBar kind="working" indeterminate caption="searching" />;
+              } else if (j.state === "downloading") {
+                const pct = Math.round((j.downloadProgress || 0) * 100);
+                const total = j.chosen?.trackCount ?? 0;
+                const done = total ? Math.round((pct / 100) * total) : 0;
+                bar = (
+                  <ProgressBar
+                    kind="working"
+                    percent={pct}
+                    caption={total ? `${pct}% · ${done}/${total} tracks` : `${pct}%`}
+                  />
+                );
+              } else if (j.state === "tagging") {
+                bar = <ProgressBar kind="verify" indeterminate caption="finishing" />;
+              } else {
+                bar = undefined;
+              }
 
               return (
                 <JobRow
@@ -176,12 +194,9 @@ export function Queue() {
                   album={r.album}
                   kind={d.kind}
                   label={d.label}
-                  step={d.step}
-                  total={d.totalSteps}
-                  indeterminate={j.state === "matching" || j.state === "matched"}
                   meta={meta}
                   note={note}
-                  stageLabel={stageLabel}
+                  bar={bar}
                 />
               );
             });

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