diff --git a/worker/lyra_worker/pipeline.py b/worker/lyra_worker/pipeline.py
index 7dfbbe4..c3dba71 100644
--- a/worker/lyra_worker/pipeline.py
+++ b/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")
diff --git a/worker/tests/test_pipeline.py b/worker/tests/test_pipeline.py
index 4ef265d..0a5bc28 100644
--- a/worker/tests/test_pipeline.py
+++ b/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)