feat(discover): max-normalize similarity scores across sources

_record_contributions now normalizes each source's top-N to [0,1] by its own
top score before summing across sources. ListenBrainz co-occurrence counts
(thousands) no longer drown Last.fm's 0–1 match scores, so a small-scale
source contributes real ranking signal instead of only adding candidates.

Discovery score assertions updated to the normalized scale (single-candidate
fakes normalize to 1.0); added a cross-source test showing a small-scale
source's candidate ranks equal to a large-scale one. min_score is now on the
[0, n_sources] scale (default 0.0 → unaffected). worker 231 tests.

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

worker/lyra_worker/discovery.py

@@ -181,11 +181,18 @@ def _record_contributions(conn: psycopg.Connection, seed_mbid: str, live, follow
         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]:
+        top = sorted(similar, key=lambda a: a.score, reverse=True)[: cfg.similar_per_seed]
+        # Max-normalize each source to [0, 1] by its own top score before summing across
+        # sources, so sources on wildly different scales contribute comparably (Last.fm's
+        # `match` is 0–1; ListenBrainz co-occurrence counts run to the thousands — raw sums
+        # let ListenBrainz drown Last.fm's ranking signal).
+        max_score = max((sa.score for sa in top), default=0.0)
+        for sa in top:
             if sa.mbid in followed:
                 continue
+            norm = sa.score / max_score if max_score > 0 else 0.0
             c = per_cand.setdefault(sa.mbid, {"name": sa.name, "score": 0.0, "sources": set()})
-            c["score"] += sa.score
+            c["score"] += norm
             c["sources"].add(src.name)
             if sa.name and not c["name"]:
                 c["name"] = sa.name

worker/tests/test_discovery.py

@@ -21,12 +21,33 @@ def test_aggregates_scores_across_seeds(conn):
     result = run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())
     assert result.artists == 2
     rows = _artist_rows(conn)
-    # c1 recommended by both seeds: score 1.1, seedCount 2; c2 by one seed
-    assert rows[0][0] == "c1" and abs(rows[0][2] - 1.1) < 1e-6 and rows[0][3] == 2
+    # c1 recommended by both seeds; each seed max-normalizes its top score to 1.0, so
+    # c1 = 1.0 + 1.0 = 2.0, seedCount 2; c2 (0.4/0.6 within s1) by one seed
+    assert rows[0][0] == "c1" and abs(rows[0][2] - 2.0) < 1e-6 and rows[0][3] == 2
     assert rows[0][4] == ["fake"] and rows[0][5] == "pending"
     assert rows[1][0] == "c2" and rows[1][3] == 1
 
 
+def test_normalizes_across_sources_of_different_scales(conn):
+    # ListenBrainz emits co-occurrence counts in the thousands; Last.fm emits 0–1 match
+    # scores. Max-normalizing each source before summing keeps the small-scale source's
+    # candidates from being drowned.
+    insert_watched_artist(conn, mbid="s1", name="Seed")
+    lb = FakeSimilaritySource(similar={"s1": [
+        SimilarArtist("c_lb", "LB Only", 5000.0), SimilarArtist("c_shared", "Shared", 4000.0)]})
+    lb.name = "listenbrainz"
+    lfm = FakeSimilaritySource(similar={"s1": [
+        SimilarArtist("c_lfm", "LFM Only", 0.9), SimilarArtist("c_shared", "Shared", 0.5)]})
+    lfm.name = "lastfm"
+    run_discovery(conn, [lb, lfm], FakeMbBrowser(), DiscoveryConfig())
+
+    lb_only = _artist_score(conn, "c_lb")[0]    # 5000/5000 = 1.0
+    lfm_only = _artist_score(conn, "c_lfm")[0]  # 0.9/0.9  = 1.0
+    shared = _artist_score(conn, "c_shared")[0]  # 4000/5000 + 0.5/0.9 = 0.8 + 0.556
+    assert abs(lb_only - lfm_only) < 1e-6         # small-scale source not drowned — ranks equal
+    assert shared > lb_only                        # co-sourced candidate outranks a single-source one
+
+
 def test_skips_already_followed_artists(conn):
     insert_watched_artist(conn, mbid="s1", name="Seed")
     insert_watched_artist(conn, mbid="c1", name="Already Followed")
@@ -116,7 +137,7 @@ def test_score_sums_across_separate_chunks(conn):
     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)
+    assert _artist_score(conn, "c1") == (2.0, 2)  # each seed max-normalizes to 1.0
 
 
 def test_partial_sweep_preserves_other_seeds_contribution(conn):
@@ -131,7 +152,7 @@ def test_partial_sweep_preserves_other_seeds_contribution(conn):
         "s2": [SimilarArtist("c1", "Shared", 0.5)],
     })
     run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())  # both -> 1.1/2
-    assert _artist_score(conn, "c1") == (1.1, 2)
+    assert _artist_score(conn, "c1") == (2.0, 2)  # each seed max-normalizes to 1.0
 
     # s1 due again (older than interval), s2 fresh; s1 now scores c1 at 0.7.
     with conn.cursor() as cur:
@@ -140,7 +161,7 @@ def test_partial_sweep_preserves_other_seeds_contribution(conn):
     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)
+    assert _artist_score(conn, "c1") == (2.0, 2)  # 1.0 (new s1) + 1.0 (kept s2), max-normalized
 
 
 def test_seed_dropoff_lowers_seedcount(conn):
@@ -153,7 +174,7 @@ def test_seed_dropoff_lowers_seedcount(conn):
         "s2": [SimilarArtist("c1", "Shared", 0.5)],
     })
     run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())
-    assert _artist_score(conn, "c1") == (1.1, 2)
+    assert _artist_score(conn, "c1") == (2.0, 2)  # each seed max-normalizes to 1.0
 
     # s1 re-swept but no longer finds c1 -> its contribution is removed, c1 recomputed.
     with conn.cursor() as cur:
@@ -162,7 +183,7 @@ def test_seed_dropoff_lowers_seedcount(conn):
     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
+    assert _artist_score(conn, "c1") == (1.0, 1)  # only s2 remains (max-normalized to 1.0)
 
 
 def test_evicts_contributions_from_removed_seed(conn):
@@ -179,7 +200,7 @@ def test_evicts_contributions_from_removed_seed(conn):
         "s2": [SimilarArtist("c1", "Shared", 0.5)],
     })
     run_discovery(conn, [src], FakeMbBrowser(), DiscoveryConfig())
-    assert _artist_score(conn, "c1") == (1.1, 2)
+    assert _artist_score(conn, "c1") == (2.0, 2)  # each seed max-normalizes to 1.0
 
     # Unfollow s1 (no FK cascade on DiscoverySeedContribution).
     with conn.cursor() as cur:
@@ -193,7 +214,7 @@ def test_evicts_contributions_from_removed_seed(conn):
     src2 = FakeSimilaritySource(similar={"s2": [SimilarArtist("c1", "Shared", 0.5)]})
     run_discovery(conn, [src2], FakeMbBrowser(), DiscoveryConfig())
 
-    assert _artist_score(conn, "c1") == (0.5, 1)  # s1's stale 0.6 evicted, not summed
+    assert _artist_score(conn, "c1") == (1.0, 1)  # s1's stale contribution evicted; s2 alone -> 1.0
     with conn.cursor() as cur:
         cur.execute('SELECT count(*) FROM "DiscoverySeedContribution" WHERE "seedMbid" = %s', ("s1",))
         assert cur.fetchone()[0] == 0