Canonical: richer, judge-able findings (the why behind a pin)

Live-run feedback: "Best-fit release: deezer (665666731), score 1.0" is too thin
to trust/accept. Each finding now explains WHY:

- score_release_detail() exposes the per-signal breakdown (count/duration/title)
  instead of just the blended score.
- resolve_canonical_for_album returns an enriched result: the breakdown,
  file_track_count vs release_track_count, and a `candidates` list of every
  source it scored (so a finding can show what the winner beat).
- resolve_and_store adds album/artist/thumb context from the row it already
  loaded (no extra query). Storage still only reads source/album_id/score.
- The job builds a real description via _describe_pin(), e.g.:
    "Pin deezer release 665666731 (confidence 100%).
     Fit to your library: 11 files vs 11 tracks on this release — track count
     100%, durations 100%, titles 100%.
     Beat: spotify 65% (17 tk)."
  and a clearer title ("Pin deezer as canonical: <artist> — <album>").

Tests: resolver enrichment (breakdown + candidate comparison fields), and
_describe_pin (judge-able text incl. the beaten alternatives, and honest "n/a"
for a missing signal). 42 canonical tests pass.

Note: the description string carries the judge-able info regardless of UI; how
the findings tab renders the extra details keys (thumb image, candidates table)
is still UI-dependent and unverified.

core/metadata/canonical_resolver.py

@@ -18,8 +18,8 @@ from __future__ import annotations
 from typing import Any, Callable, Dict, List, Optional
 
 from core.metadata.canonical_version import (
-    pick_canonical_release,
     score_release_against_files,
+    score_release_detail,
 )
 
 # Source-selection modes (a per-job setting). See resolve_canonical_for_album.
@@ -59,15 +59,20 @@ def resolve_canonical_for_album(
         floor; never considers other sources).
       - ``best_fit``: whichever source's release best matches the files.
 
-    Returns ``{'source', 'album_id', 'score'}`` or ``None`` when there are no
+    Returns an enriched dict for the chosen release — ``source``, ``album_id``,
-    files, no resolvable candidates, or nothing clears ``min_score``."""
+    ``score``, the per-signal breakdown (``count_fit``/``duration_fit``/
+    ``title_fit``), ``file_track_count`` vs ``release_track_count``, and a
+    ``candidates`` list of everything it scored (so a finding can show WHY the
+    pick won and what it beat). ``None`` when there are no files, no resolvable
+    candidates, or nothing clears ``min_score``."""
     if not file_tracks:
         return None
     primary = primary_source or (source_priority[0] if source_priority else None)
+    scored: List[Dict[str, Any]] = []  # every source we actually scored
 
-    def _candidate(source: Optional[str]) -> Optional[Dict[str, Any]]:
+    def _score(source: Optional[str]) -> Optional[Dict[str, Any]]:
-        if not source:
+        if not source or any(e['source'] == source for e in scored):
-            return None
+            return next((e for e in scored if e['source'] == source), None)
         album_id = album_source_ids.get(source)
         if not album_id:
             return None
@@ -77,38 +82,53 @@ def resolve_canonical_for_album(
             tracks = None
         if not tracks:
             return None
-        return {'source': source, 'album_id': str(album_id), 'tracks': tracks}
+        entry = {
+            'source': source, 'album_id': str(album_id),
+            'track_count': len(tracks), 'score': round(score_release_against_files(file_tracks, tracks), 4),
+            '_tracks': tracks,
+        }
+        scored.append(entry)
+        return entry
+
+    winner: Optional[Dict[str, Any]] = None
 
     # Active-source modes: try the primary first.
     if mode in (MODE_ACTIVE_ONLY, MODE_ACTIVE_PREFERRED):
-        cand = _candidate(primary)
+        p = _score(primary)
-        if cand:
+        if p and p['score'] >= min_score:
-            score = score_release_against_files(file_tracks, cand['tracks'])
+            winner = p
-            if score >= min_score:
+        elif mode == MODE_ACTIVE_ONLY:
-                return {'source': cand['source'], 'album_id': cand['album_id'], 'score': round(score, 4)}
+            return None  # never consider other sources
-        if mode == MODE_ACTIVE_ONLY:
-            return None  # never fall back to other sources
 
-    # best_fit (and active_preferred's fallback): score the candidate sources
+    # best_fit, or active_preferred fallback: score the rest and pick the best.
-    # (skipping the primary we already tried in active_preferred) and pick best.
+    if winner is None:
-    candidates: List[Dict[str, Any]] = []
+        for source in source_priority:
-    for source in source_priority:
+            _score(source)
-        if mode == MODE_ACTIVE_PREFERRED and source == primary:
+        best = None
-            continue
+        for e in scored:  # source_priority order -> strictly-greater = priority tiebreak
-        cand = _candidate(source)
+            if best is None or e['score'] > best['score'] + 1e-9:
-        if cand:
+                best = e
-            candidates.append(cand)
+        if best and best['score'] >= min_score:
+            winner = best
 
-    if not candidates:
+    if winner is None:
         return None
 
-    best, score = pick_canonical_release(file_tracks, candidates, min_score=min_score)
+    detail = score_release_detail(file_tracks, winner['_tracks'])
-    if not best:
-        return None
     return {
-        'source': best['source'],
+        'source': winner['source'],
-        'album_id': best['album_id'],
+        'album_id': winner['album_id'],
-        'score': round(score, 4),
+        'score': winner['score'],
+        'file_track_count': detail['file_track_count'],
+        'release_track_count': detail['release_track_count'],
+        'count_fit': detail['count_fit'],
+        'duration_fit': detail['duration_fit'],
+        'title_fit': detail['title_fit'],
+        'candidates': [
+            {'source': e['source'], 'album_id': e['album_id'],
+             'track_count': e['track_count'], 'score': e['score']}
+            for e in scored
+        ],
     }
 
 
@@ -196,8 +216,15 @@ def resolve_and_store_canonical_for_album(
         mode=mode,
         primary_source=primary_source,
     )
-    if result and store:
+    if result:
-        db.set_album_canonical(album_id, result['source'], result['album_id'], result['score'])
+        # Album/artist/art context for richer findings (read from the row we
+        # already loaded — no extra query). Storage only uses source/id/score.
+        result['album_title'] = album_data.get('title') or ''
+        result['artist_name'] = album_data.get('artist_name') or ''
+        if album_data.get('thumb_url'):
+            result['album_thumb_url'] = album_data['thumb_url']
+        if store:
+            db.set_album_canonical(album_id, result['source'], result['album_id'], result['score'])
     return result
 
 

core/metadata/canonical_version.py

@@ -145,6 +145,34 @@ def score_release_against_files(
     return _combine([(count, _W_COUNT), (dur, _W_DURATION), (title, _W_TITLE)])
 
 
+def score_release_detail(
+    file_tracks: List[Dict[str, Any]],
+    release_tracks: List[Dict[str, Any]],
+    *,
+    duration_tolerance_ms: int = _DEFAULT_DURATION_TOLERANCE_MS,
+) -> Dict[str, Any]:
+    """Like ``score_release_against_files`` but returns the per-signal breakdown
+    so a UI can show WHY a release scored the way it did. ``duration_fit`` /
+    ``title_fit`` are ``None`` when that signal was absent."""
+    if not file_tracks or not release_tracks:
+        return {
+            'score': 0.0, 'count_fit': 0.0, 'duration_fit': None, 'title_fit': None,
+            'release_track_count': len(release_tracks), 'file_track_count': len(file_tracks),
+        }
+    count = _count_fit(len(file_tracks), len(release_tracks))
+    dur = _duration_fit(file_tracks, release_tracks, duration_tolerance_ms)
+    title = _title_fit(file_tracks, release_tracks)
+    score = _combine([(count, _W_COUNT), (dur, _W_DURATION), (title, _W_TITLE)])
+    return {
+        'score': round(score, 4),
+        'count_fit': round(count, 3),
+        'duration_fit': round(dur, 3) if dur is not None else None,
+        'title_fit': round(title, 3) if title is not None else None,
+        'release_track_count': len(release_tracks),
+        'file_track_count': len(file_tracks),
+    }
+
+
 def pick_canonical_release(
     file_tracks: List[Dict[str, Any]],
     candidates: List[Dict[str, Any]],

core/repair_jobs/canonical_version_resolve.py

@@ -22,6 +22,32 @@ from utils.logging_config import get_logger
 logger = get_logger("repair_job.canonical_version")
 
 
+def _pct(v) -> str:
+    return f"{round(v * 100)}%" if isinstance(v, (int, float)) else "n/a"
+
+
+def _describe_pin(resolved: dict) -> str:
+    """Human-readable, judge-able explanation of WHY this release was chosen."""
+    lines = [
+        f"Pin {resolved['source']} release {resolved['album_id']} "
+        f"(confidence {_pct(resolved.get('score'))}).",
+        f"Fit to your library: {resolved.get('file_track_count', '?')} files vs "
+        f"{resolved.get('release_track_count', '?')} tracks on this release — "
+        f"track count {_pct(resolved.get('count_fit'))}, "
+        f"durations {_pct(resolved.get('duration_fit'))}, "
+        f"titles {_pct(resolved.get('title_fit'))}.",
+    ]
+    others = [c for c in resolved.get('candidates', []) if c.get('source') != resolved.get('source')]
+    if others:
+        comp = ", ".join(
+            f"{c['source']} {_pct(c['score'])} ({c['track_count']} tk)" for c in others
+        )
+        lines.append(f"Beat: {comp}.")
+    elif len(resolved.get('candidates', [])) == 1:
+        lines.append("Only this source had a release linked for this album.")
+    return "\n".join(lines)
+
+
 @register_job
 class CanonicalVersionResolveJob(RepairJob):
     job_id = 'canonical_version_resolve'
@@ -133,6 +159,8 @@ class CanonicalVersionResolveJob(RepairJob):
             result.scanned += 1
             if resolved:
                 if dry_run and context.create_finding:
+                    artist = resolved.get('artist_name') or ''
+                    label = f"{artist} — {album_title}" if artist else (album_title or str(album_id))
                     inserted = context.create_finding(
                         job_id=self.job_id,
                         finding_type='canonical_version',
@@ -140,11 +168,8 @@ class CanonicalVersionResolveJob(RepairJob):
                         entity_type='album',
                         entity_id=str(album_id),
                         file_path=None,
-                        title=f'Would pin canonical: {album_title or album_id}',
+                        title=f'Pin {resolved["source"]} as canonical: {label}',
-                        description=(
+                        description=_describe_pin(resolved),
-                            f"Best-fit release: {resolved['source']} "
-                            f"({resolved['album_id']}), score {resolved['score']}"
-                        ),
                         details={'album_id': str(album_id), **resolved},
                     )
                     if inserted:

tests/test_canonical_resolver.py

@@ -92,6 +92,24 @@ def test_active_only_pins_primary_and_never_falls_back():
     assert out is None  # primary didn't fit, and active_only won't consider others
 
 
+def test_result_includes_breakdown_and_candidate_comparison():
+    files = list(STD)
+    table = {("spotify", "sp1"): DLX, ("deezer", "dz1"): STD}
+    out = resolve_canonical_for_album(
+        album_source_ids={"spotify": "sp1", "deezer": "dz1"},
+        file_tracks=files, fetch_tracklist=_fetcher(table),
+        source_priority=["spotify", "deezer", "itunes", "musicbrainz"],
+        mode="best_fit",
+    )
+    assert out["source"] == "deezer"
+    assert out["file_track_count"] == 11
+    assert out["release_track_count"] == 11
+    assert out["count_fit"] == 1.0 and out["duration_fit"] == 1.0 and out["title_fit"] == 1.0
+    by_src = {c["source"]: c for c in out["candidates"]}
+    assert by_src["deezer"]["track_count"] == 11 and by_src["deezer"]["score"] > 0.9
+    assert by_src["spotify"]["track_count"] == 17 and by_src["spotify"]["score"] < 0.8
+
+
 def test_active_only_pins_primary_when_it_fits():
     files = list(STD)
     table = {("spotify", "sp1"): STD, ("musicbrainz", "mb1"): STD}

tests/test_canonical_version_job.py

@@ -6,7 +6,10 @@ import types
 
 import core.repair_jobs.canonical_version_resolve as cvr
 from core.repair_jobs import get_all_jobs
-from core.repair_jobs.canonical_version_resolve import CanonicalVersionResolveJob
+from core.repair_jobs.canonical_version_resolve import (
+    CanonicalVersionResolveJob,
+    _describe_pin,
+)
 from database.music_database import MusicDatabase
 
 
@@ -56,6 +59,33 @@ def test_source_selection_defaults_to_active_preferred():
     assert CanonicalVersionResolveJob.default_settings["source_selection"] == "active_preferred"
 
 
+def test_describe_pin_is_judgeable():
+    desc = _describe_pin({
+        "source": "deezer", "album_id": "665666731", "score": 1.0,
+        "file_track_count": 11, "release_track_count": 11,
+        "count_fit": 1.0, "duration_fit": 1.0, "title_fit": 1.0,
+        "candidates": [
+            {"source": "deezer", "album_id": "665666731", "track_count": 11, "score": 1.0},
+            {"source": "spotify", "album_id": "sp", "track_count": 17, "score": 0.65},
+        ],
+    })
+    assert "deezer" in desc and "665666731" in desc
+    assert "11 files vs 11 tracks" in desc           # your library vs the release
+    assert "durations 100%" in desc and "titles 100%" in desc  # the WHY
+    assert "Beat:" in desc and "spotify 65% (17 tk)" in desc    # what it beat
+
+
+def test_describe_pin_single_source():
+    desc = _describe_pin({
+        "source": "spotify", "album_id": "x", "score": 0.9,
+        "file_track_count": 10, "release_track_count": 10,
+        "count_fit": 1.0, "duration_fit": None, "title_fit": 0.9,
+        "candidates": [{"source": "spotify", "album_id": "x", "track_count": 10, "score": 0.9}],
+    })
+    assert "Only this source" in desc
+    assert "durations n/a" in desc  # missing signal shown honestly
+
+
 def test_live_resolves_and_stores(tmp_path, monkeypatch):
     db = MusicDatabase(str(tmp_path / "m.db"))
     _seed_two_albums(db)