fix(worker): mark the chosen candidate at download-start, not after success

_mark_chosen ran only after a download succeeded, so throughout the 'downloading'
state no candidate was chosen — the Floor row's "source · format · tracks" line was
blank (only elapsed showed). Now mark the candidate exclusively at each download
attempt so `chosen` reflects the source in flight (and updates on fall-through).

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

worker/lyra_worker/pipeline.py

@@ -104,10 +104,12 @@ def _persist_candidates(conn: psycopg.Connection, job_id: str, candidates: list[
 
 
 def _mark_chosen(conn: psycopg.Connection, job_id: str, source_ref: str) -> None:
+    """Mark exactly one candidate (the one now being downloaded) as chosen, clearing any
+    prior choice. Called at each download attempt so `chosen` reflects the source in flight."""
     with conn.cursor() as cur:
         cur.execute(
-            'UPDATE "Candidate" SET chosen = true WHERE "jobId" = %s AND "sourceRef" = %s',
-            (job_id, source_ref),
+            'UPDATE "Candidate" SET chosen = ("sourceRef" = %s) WHERE "jobId" = %s',
+            (source_ref, job_id),
         )
     conn.commit()
 
@@ -204,11 +206,11 @@ def run_pipeline(
             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
-                _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")

worker/tests/test_pipeline.py

@@ -46,6 +46,20 @@ def test_falls_through_when_best_download_fails(conn):
         assert cur.fetchone()[0] == "soulseek"
 
 
+def test_fallthrough_marks_exactly_one_chosen(conn):
+    # `chosen` now marks the source in flight at each attempt; after falling through from a
+    # failed source to a working one, exactly ONE candidate stays chosen (the winner) — the
+    # attempted-then-failed source must be un-marked (exclusive _mark_chosen).
+    job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
+    claim_next(conn)
+    run_pipeline(conn, job_id, [FailingAdapter(), FakeSoulseek()], dest_root="/tmp/lib")
+    with conn.cursor() as cur:
+        cur.execute('SELECT count(*) FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
+        assert cur.fetchone()[0] == 1
+        cur.execute('SELECT source FROM "Candidate" WHERE "jobId" = %s AND chosen = true', (job_id,))
+        assert cur.fetchone()[0] == "soulseek"
+
+
 def test_no_match_goes_to_needs_attention(conn):
     job_id = insert_request(conn, artist="Radiohead", album="In Rainbows")
     claim_next(conn)