fix: make intake dedupe cutoff-aware so monitor upgrade path works end-to-end

The monitor's quality-upgrade path was inert: run_pipeline's intake
short-circuit treated any existing LibraryItem as "done", so a
monitor-driven upgrade job for a below-cutoff copy jumped straight to
imported without downloading. Intake now checks whether the job is
monitor-driven (Request.monitoredReleaseId IS NOT NULL) and, if so,
only short-circuits when the existing copy already meets the quality
cutoff; plain manual requests keep exact slice-1 behavior. Also drops
an unused `field` import in browser.py.

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

worker/lyra_worker/browser.py

@@ -1,4 +1,4 @@
-from dataclasses import dataclass, field
+from dataclasses import dataclass
 from typing import Protocol
 
 

worker/lyra_worker/main.py

@@ -37,7 +37,8 @@ def run_forever() -> None:
                 time.sleep(IDLE_SLEEP)
                 continue
             print(f"worker: claimed job {job_id}", flush=True)
-            run_pipeline(conn, job_id, adapters, resolver=resolver, tagger=tagger)
+            run_pipeline(conn, job_id, adapters, resolver=resolver, tagger=tagger,
+                         upgrade_cutoff=mcfg.quality_cutoff)
             if mcfg.enabled:
                 try:
                     reconcile(conn, job_id, mcfg)

worker/lyra_worker/pipeline.py

@@ -58,6 +58,27 @@ def _already_in_library(conn: psycopg.Connection, target: MBTarget) -> bool:
         return cur.fetchone() is not None
 
 
+def _is_upgrade_job(conn: psycopg.Connection, job_id: str) -> bool:
+    with conn.cursor() as cur:
+        cur.execute(
+            'SELECT "monitoredReleaseId" IS NOT NULL FROM "Request" '
+            'WHERE id = (SELECT "requestId" FROM "Job" WHERE id = %s)',
+            (job_id,),
+        )
+        row = cur.fetchone()
+        return bool(row and row[0])
+
+
+def _already_in_library_at_cutoff(conn: psycopg.Connection, target: MBTarget, cutoff: int) -> bool:
+    with conn.cursor() as cur:
+        cur.execute(
+            'SELECT 1 FROM "LibraryItem" WHERE artist = %s AND album = %s '
+            'AND "qualityClass" >= %s LIMIT 1',
+            (target.artist, target.album, cutoff),
+        )
+        return cur.fetchone() is not None
+
+
 def _persist_candidates(conn: psycopg.Connection, job_id: str, candidates: list[Candidate]) -> None:
     with conn.cursor() as cur:
         for c in candidates:
@@ -110,6 +131,7 @@ def run_pipeline(
     tagger=None,
     min_confidence: float = 0.7,
     dest_root: str = "/music",
+    upgrade_cutoff: int | None = None,
 ) -> None:
     """Real staged acquisition using source-agnostic adapters. Fakes in this plan."""
     names = [a.name for a in adapters]
@@ -123,7 +145,13 @@ def run_pipeline(
         resolved = resolver.resolve(target.artist, target.album)
         if resolved is not None:
             target = resolved
-    if _already_in_library(conn, target):
+    # dedupe: a monitor-driven upgrade job proceeds unless the existing copy already meets
+    # the cutoff; a plain request short-circuits on any existing copy (slice-1 behavior).
+    if upgrade_cutoff is not None and _is_upgrade_job(conn, job_id):
+        _dedupe_hit = _already_in_library_at_cutoff(conn, target, upgrade_cutoff)
+    else:
+        _dedupe_hit = _already_in_library(conn, target)
+    if _dedupe_hit:
         _set_state(conn, job_id, "imported", "import")
         with conn.cursor() as cur:
             cur.execute(