From de052a4ef7cf6761bd309564c5d0d533b8a38e07 Mon Sep 17 00:00:00 2001 From: Jonathan Date: Mon, 13 Jul 2026 15:18:30 +0200 Subject: [PATCH] feat(worker): per-seed discovery contributions so score sums across chunks/sweeps Co-Authored-By: Claude Opus 4.8 (1M context) --- worker/lyra_worker/discovery.py | 114 ++++++++++++++++++++++---------- worker/tests/test_discovery.py | 69 +++++++++++++++++++ 2 files changed, 149 insertions(+), 34 deletions(-) diff --git a/worker/lyra_worker/discovery.py b/worker/lyra_worker/discovery.py index e978799..62d4a4b 100644 --- a/worker/lyra_worker/discovery.py +++ b/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)) diff --git a/worker/tests/test_discovery.py b/worker/tests/test_discovery.py index ec6110e..5425d84 100644 --- a/worker/tests/test_discovery.py +++ b/worker/tests/test_discovery.py @@ -96,6 +96,75 @@ def test_config_from_config_parses_and_defaults(): d.albums_per_artist, d.min_score) == (False, 168, 5, 20, 1, 0.0) +def _artist_score(conn, mbid): + with conn.cursor() as cur: + cur.execute('SELECT score, "seedCount" FROM "DiscoverySuggestion" WHERE "artistMbid" = %s', (mbid,)) + return cur.fetchone() + + +def test_score_sums_across_separate_chunks(conn): + # Two seeds, both eligible, forced into separate one-seed chunks by chunk_size=1. + insert_watched_artist(conn, mbid="s1", name="Seed One", + last_discovered_sql="now() - interval '200 hours'") + insert_watched_artist(conn, mbid="s2", name="Seed Two", + last_discovered_sql="now() - interval '190 hours'") + src = FakeSimilaritySource(similar={ + "s1": [SimilarArtist("c1", "Shared", 0.6)], + "s2": [SimilarArtist("c1", "Shared", 0.5)], + }) + cfg = DiscoveryConfig(chunk_size=1) + run_discovery(conn, [src], FakeMbBrowser(), cfg) # processes s1 (older) + run_discovery(conn, [src], FakeMbBrowser(), cfg) # processes s2 + # replace-not-accumulate would leave 0.5/seedCount1; contributions give the sum. + assert _artist_score(conn, "c1") == (1.1, 2) + + +def test_partial_sweep_preserves_other_seeds_contribution(conn): + # Full sweep: c1 gets s1(0.6)+s2(0.5)=1.1. Then only s1 is due again with a new + # score; s2's contribution must survive untouched. + insert_watched_artist(conn, mbid="s1", name="Seed One", + last_discovered_sql="now() - interval '200 hours'") + insert_watched_artist(conn, mbid="s2", name="Seed Two", + last_discovered_sql="now() - interval '200 hours'") + src = FakeSimilaritySource(similar={ + "s1": [SimilarArtist("c1", "Shared", 0.6)], + "s2": [SimilarArtist("c1", "Shared", 0.5)], + }) + run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig()) # both -> 1.1/2 + assert _artist_score(conn, "c1") == (1.1, 2) + + # s1 due again (older than interval), s2 fresh; s1 now scores c1 at 0.7. + with conn.cursor() as cur: + cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() - interval \'200 hours\' WHERE mbid = %s', ("s1",)) + cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() WHERE mbid = %s', ("s2",)) + conn.commit() + src2 = FakeSimilaritySource(similar={"s1": [SimilarArtist("c1", "Shared", 0.7)]}) + run_discovery(conn, [src2], FakeMbBrowser(), DiscoveryConfig()) + assert _artist_score(conn, "c1") == (1.2, 2) # 0.7 (new s1) + 0.5 (kept s2) + + +def test_seed_dropoff_lowers_seedcount(conn): + insert_watched_artist(conn, mbid="s1", name="Seed One", + last_discovered_sql="now() - interval '200 hours'") + insert_watched_artist(conn, mbid="s2", name="Seed Two", + last_discovered_sql="now() - interval '200 hours'") + src = FakeSimilaritySource(similar={ + "s1": [SimilarArtist("c1", "Shared", 0.6)], + "s2": [SimilarArtist("c1", "Shared", 0.5)], + }) + run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig()) + assert _artist_score(conn, "c1") == (1.1, 2) + + # s1 re-swept but no longer finds c1 -> its contribution is removed, c1 recomputed. + with conn.cursor() as cur: + cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() - interval \'200 hours\' WHERE mbid = %s', ("s1",)) + cur.execute('UPDATE "WatchedArtist" SET "lastDiscoveredAt" = now() WHERE mbid = %s', ("s2",)) + conn.commit() + src2 = FakeSimilaritySource(similar={"s1": [SimilarArtist("c9", "Other", 0.9)]}) + run_discovery(conn, [src2], FakeMbBrowser(), DiscoveryConfig()) + assert _artist_score(conn, "c1") == (0.5, 1) # only s2 remains + + def test_run_discovery_processes_at_most_chunk_size_seeds(conn): from lyra_worker.discovery import DiscoveryConfig, run_discovery from lyra_worker.adapters.fakes import FakeMbBrowser, FakeSimilaritySource