feat(worker): per-seed discovery contributions so score sums across chunks/sweeps

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

worker/lyra_worker/discovery.py

@@ -143,47 +143,93 @@ def _derive_albums(conn: psycopg.Connection, browser: MbBrowser, surfaced: list[
     return albums
 
 
+def _record_contributions(conn: psycopg.Connection, seed_mbid: str, live, followed: set[str],
+                          cfg: DiscoveryConfig) -> set[str]:
+    """Replace seed_mbid's contribution rows with its current similar-artist scores.
+    Returns the set of candidate mbids whose aggregate may have changed (old contributors
+    of this seed ∪ newly inserted), so callers recompute drop-offs too."""
+    with conn.cursor() as cur:
+        cur.execute('SELECT "candidateMbid" FROM "DiscoverySeedContribution" WHERE "seedMbid" = %s',
+                    (seed_mbid,))
+        affected = {r[0] for r in cur.fetchall()}
+        cur.execute('DELETE FROM "DiscoverySeedContribution" WHERE "seedMbid" = %s', (seed_mbid,))
+
+    per_cand: dict[str, dict] = {}
+    for src in live:
+        try:
+            similar = src.similar_artists(seed_mbid)
+        except Exception as e:  # a bad source/seed must not abort the sweep
+            print(f"worker: discovery source {src.name} failed for {seed_mbid}: {e}", flush=True)
+            continue
+        for sa in sorted(similar, key=lambda a: a.score, reverse=True)[: cfg.similar_per_seed]:
+            if sa.mbid in followed:
+                continue
+            c = per_cand.setdefault(sa.mbid, {"name": sa.name, "score": 0.0, "sources": set()})
+            c["score"] += sa.score
+            c["sources"].add(src.name)
+            if sa.name and not c["name"]:
+                c["name"] = sa.name
+
+    with conn.cursor() as cur:
+        for mbid, c in per_cand.items():
+            cur.execute(
+                'INSERT INTO "DiscoverySeedContribution" (id, "candidateMbid", "candidateName", '
+                '"seedMbid", score, sources, "updatedAt") '
+                'VALUES (gen_random_uuid()::text, %s, %s, %s, %s, %s, now())',
+                (mbid, c["name"], seed_mbid, c["score"], sorted(c["sources"])),
+            )
+            affected.add(mbid)
+    return affected
+
+
+def _recompute_suggestions(conn: psycopg.Connection, affected: set[str],
+                           cfg: DiscoveryConfig) -> list[dict]:
+    """Recompute each affected candidate's DiscoverySuggestion from the sum of its
+    contributions. Returns the candidates that produced a live pending suggestion."""
+    if not affected:
+        return []
+    with conn.cursor() as cur:
+        cur.execute(
+            'SELECT "candidateMbid", "candidateName", score, "seedMbid", sources '
+            'FROM "DiscoverySeedContribution" WHERE "candidateMbid" = ANY(%s)',
+            (list(affected),))
+        rows = cur.fetchall()
+
+    agg: dict[str, dict] = {}
+    for mbid, name, score, seed, sources in rows:
+        a = agg.setdefault(mbid, {"mbid": mbid, "name": name, "score": 0.0,
+                                  "seeds": set(), "sources": set()})
+        a["score"] += score
+        a["seeds"].add(seed)
+        a["sources"].update(sources or [])
+        if name and not a["name"]:
+            a["name"] = name
+
+    surfaced: list[dict] = []
+    for a in sorted(agg.values(), key=lambda a: a["score"], reverse=True):
+        if a["score"] < cfg.min_score:
+            continue
+        cand = {"mbid": a["mbid"], "name": a["name"], "score": a["score"],
+                "seed_count": len(a["seeds"]), "sources": a["sources"]}
+        if _upsert_artist(conn, cand):  # rowcount 1 => a live pending suggestion
+            surfaced.append(cand)
+    return surfaced
+
+
 def run_discovery(conn: psycopg.Connection, sources: list[SimilaritySource],
                   browser: MbBrowser, cfg: DiscoveryConfig) -> DiscoveryResult:
-    """Aggregate similar artists across the library's seed artists; upsert suggestions."""
+    """Record each seed's similar-artist contributions, then recompute affected
+    suggestions from the summed contributions — so a candidate's score is the sum of
+    every seed's latest contribution, independent of chunk and sweep boundaries."""
     seeds = _seed_artists(conn, cfg)
     followed = _followed_mbids(conn)
     live = _healthy_sources(sources)
 
-    agg: dict[str, dict] = {}
+    affected: set[str] = set()
     for seed_mbid, _seed_name in seeds:
-        for src in live:
-            try:
-                similar = src.similar_artists(seed_mbid)
-            except Exception as e:  # a bad source/seed must not abort the sweep
-                print(f"worker: discovery source {src.name} failed for {seed_mbid}: {e}", flush=True)
-                continue
-            for sa in sorted(similar, key=lambda a: a.score, reverse=True)[: cfg.similar_per_seed]:
-                if sa.mbid in followed:
-                    continue
-                c = agg.setdefault(sa.mbid, {"mbid": sa.mbid, "name": sa.name,
-                                             "score": 0.0, "seeds": set(), "sources": set()})
-                c["score"] += sa.score
-                c["seeds"].add(seed_mbid)
-                c["sources"].add(src.name)
-                if sa.name and not c["name"]:
-                    c["name"] = sa.name
-
-    artists = 0
-    surfaced: list[dict] = []
-    for c in sorted(agg.values(), key=lambda c: c["score"], reverse=True):
-        if c["score"] < cfg.min_score:
-            continue
-        cand = {"mbid": c["mbid"], "name": c["name"], "score": c["score"],
-                "seed_count": len(c["seeds"]), "sources": c["sources"]}
-        inserted = _upsert_artist(conn, cand)
-        artists += inserted
-        # Only derive albums for candidates that produced a live pending suggestion.
-        # A no-op upsert (rowcount 0) means the artist was already dismissed/wanted —
-        # don't surface their albums against a dismissal.
-        if inserted:
-            surfaced.append(cand)
+        affected |= _record_contributions(conn, seed_mbid, live, followed, cfg)
 
+    surfaced = _recompute_suggestions(conn, affected, cfg)
     albums = _derive_albums(conn, browser, surfaced, cfg)
 
     with conn.cursor() as cur:
@@ -191,4 +237,4 @@ def run_discovery(conn: psycopg.Connection, sources: list[SimilaritySource],
             cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() WHERE mbid = %s',
                         (seed_mbid,))
     conn.commit()
-    return DiscoveryResult(artists=artists, albums=albums, seeds=len(seeds))
+    return DiscoveryResult(artists=len(surfaced), albums=albums, seeds=len(seeds))