Canonical album version — Stage 1: schema + pure scorer (dormant)

First stage of the canonical-album-version fix (#765 + #767-Bug2). Pins ONE
canonical (source, album_id) per album, chosen by best-fit to the user's actual
files, so the Reorganizer, Track Number Repair, and tagging stop re-resolving
independently and contradicting each other.

Ships DORMANT — nothing reads or writes the new data yet, so zero behavior
change. Later stages populate (Stage 2) and consume (Stages 3-4) it.

- core/metadata/canonical_version.py — pure scorer (the testable heart):
  score_release_against_files() rates a candidate release by track-count fit +
  duration alignment (greedy nearest within ±3s) + title overlap, dropping and
  renormalizing missing signals so it never crashes on sparse metadata.
  pick_canonical_release() takes candidates in source-priority order, picks the
  best fit, breaks ties toward the earlier (higher-priority) candidate so the
  choice is DETERMINISTIC — that determinism is what makes every tool agree
  (#765), while count/duration fit picks the right EDITION (#767-Bug2). A
  confidence floor (default 0.5) means a low-confidence guess is never pinned.

- database/music_database.py — additive, nullable columns on albums
  (canonical_source / canonical_album_id / canonical_score /
  canonical_resolved_at), guarded by the existing PRAGMA-table_info pattern.
  NULL = unresolved = every consumer falls back to today's behavior.

Tests: tests/test_canonical_version.py (11) — edition discrimination (11 files
-> standard, 17 -> deluxe), deterministic priority tiebreak, duration
disambiguation on count ties, graceful degradation (no durations / counts only /
fuzzy titles), confidence floor, empty-input safety. tests/test_canonical_
columns_migration.py (4) — fresh DB has the columns, they're nullable w/ NULL
default, migration is idempotent, and it ALTERs them onto an old albums table.
60 DB/schema regression tests still pass.

core/metadata/canonical_version.py

@@ -0,0 +1,182 @@
+"""Pick the canonical album release by best-fit to the user's actual files.
+
+Issue #765 / #767-Bug2: SoulSync never pins ONE canonical album version per
+album, so the Library Reorganizer, Track Number Repair, and tagging each
+re-resolve independently and can land on different releases (standard vs
+deluxe; Spotify vs MusicBrainz track numbering) and contradict each other.
+
+This module is the pure, testable heart of the fix: given the metadata of the
+files actually on disk and a set of candidate releases, score each release by
+how well it FITS those files and pick the best. "Best-fit to the files" means:
+
+  - track-count fit   — a 17-track deluxe is a poor fit for 11 files on disk
+  - duration alignment — each file should line up with a release track by length
+  - title overlap      — a tiebreaker / sanity check
+
+What this does and does NOT solve:
+  - It DOES pick the right EDITION (standard vs deluxe) — the discriminating
+    signal is track count + durations.
+  - It does NOT (and cannot) decide which of two listings of the SAME album is
+    "more correct" when they differ only in track numbering (same files match
+    both equally). Instead ``pick_canonical_release`` is DETERMINISTIC and
+    breaks ties toward the earlier candidate — so the caller passes candidates
+    in source-priority order and every tool that reads the pinned result agrees
+    on the same release. Agreement is what resolves #765, not picking a
+    "winner" of the numbering disagreement.
+
+Pure, no I/O. Callers fetch candidate tracklists and read on-disk file metadata;
+this module only scores.
+"""
+
+from __future__ import annotations
+
+import re
+from difflib import SequenceMatcher
+from typing import Any, Dict, List, Optional, Tuple
+
+# Weights for the three fit signals. Count + duration dominate because "matches
+# my files" is fundamentally about having the right NUMBER of the right-LENGTH
+# tracks; title is a tiebreaker. Missing signals are dropped and the present
+# ones renormalized (see _combine).
+_W_COUNT = 0.4
+_W_DURATION = 0.4
+_W_TITLE = 0.2
+
+_DEFAULT_DURATION_TOLERANCE_MS = 3000  # ±3s — covers encode/version length jitter
+_DEFAULT_MIN_SCORE = 0.5               # never pin below this — leave unresolved
+_TITLE_FUZZY_THRESHOLD = 0.85
+
+
+def _norm_title(text: str) -> str:
+    """Lowercase, drop bracketed qualifiers ((feat. …), [Remastered]), strip
+    punctuation, collapse whitespace."""
+    if not text:
+        return ""
+    t = str(text).lower()
+    t = re.sub(r"[\(\[].*?[\)\]]", "", t)
+    t = re.sub(r"[^a-z0-9 ]", " ", t)
+    return " ".join(t.split())
+
+
+def _count_fit(n_files: int, n_release: int) -> float:
+    """1.0 when track counts match; decays with the relative difference."""
+    if n_files <= 0 or n_release <= 0:
+        return 0.0
+    return 1.0 - min(1.0, abs(n_files - n_release) / max(n_files, n_release))
+
+
+def _duration_fit(
+    file_tracks: List[Dict[str, Any]],
+    release_tracks: List[Dict[str, Any]],
+    tolerance_ms: int,
+) -> Optional[float]:
+    """Fraction of tracks that line up by duration (greedy nearest match within
+    tolerance), over the larger of the two track counts — so missing or extra
+    tracks are penalised. Returns ``None`` when neither side has durations."""
+    f_durs = [int(f["duration_ms"]) for f in file_tracks if f.get("duration_ms")]
+    r_durs = [int(r["duration_ms"]) for r in release_tracks if r.get("duration_ms")]
+    if not f_durs or not r_durs:
+        return None
+    used = [False] * len(r_durs)
+    matched = 0
+    for fd in f_durs:
+        best_j, best_diff = -1, tolerance_ms + 1
+        for j, rd in enumerate(r_durs):
+            if used[j]:
+                continue
+            diff = abs(fd - rd)
+            if diff <= tolerance_ms and diff < best_diff:
+                best_diff, best_j = diff, j
+        if best_j >= 0:
+            used[best_j] = True
+            matched += 1
+    denom = max(len(file_tracks), len(release_tracks))
+    return matched / denom if denom else 0.0
+
+
+def _title_fit(
+    file_tracks: List[Dict[str, Any]],
+    release_tracks: List[Dict[str, Any]],
+) -> Optional[float]:
+    """Fraction of files whose title matches some release title (exact-normalised
+    or fuzzy), over the larger track count. ``None`` when titles are absent."""
+    f_titles = [_norm_title(f.get("title", "")) for f in file_tracks]
+    f_titles = [t for t in f_titles if t]
+    r_titles = [_norm_title(r.get("title", "")) for r in release_tracks]
+    r_titles = [t for t in r_titles if t]
+    if not f_titles or not r_titles:
+        return None
+    r_set = set(r_titles)
+    matched = 0
+    for ft in f_titles:
+        if ft in r_set or any(
+            SequenceMatcher(None, ft, rt).ratio() >= _TITLE_FUZZY_THRESHOLD
+            for rt in r_titles
+        ):
+            matched += 1
+    denom = max(len(file_tracks), len(release_tracks))
+    return matched / denom if denom else 0.0
+
+
+def _combine(parts: List[Tuple[Optional[float], float]]) -> float:
+    """Weighted mean over present (non-None) components, renormalising weights."""
+    present = [(v, w) for v, w in parts if v is not None]
+    total_w = sum(w for _, w in present)
+    if total_w <= 0:
+        return 0.0
+    return sum(v * w for v, w in present) / total_w
+
+
+def score_release_against_files(
+    file_tracks: List[Dict[str, Any]],
+    release_tracks: List[Dict[str, Any]],
+    *,
+    duration_tolerance_ms: int = _DEFAULT_DURATION_TOLERANCE_MS,
+) -> float:
+    """Score 0.0–1.0 of how well ``release_tracks`` fits the on-disk
+    ``file_tracks``. Each track dict may carry ``duration_ms`` and ``title``;
+    missing signals are dropped and the rest renormalised so the function never
+    crashes on sparse metadata (it just leans on what's available)."""
+    if not file_tracks or not release_tracks:
+        return 0.0
+    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)
+    return _combine([(count, _W_COUNT), (dur, _W_DURATION), (title, _W_TITLE)])
+
+
+def pick_canonical_release(
+    file_tracks: List[Dict[str, Any]],
+    candidates: List[Dict[str, Any]],
+    *,
+    min_score: float = _DEFAULT_MIN_SCORE,
+    duration_tolerance_ms: int = _DEFAULT_DURATION_TOLERANCE_MS,
+) -> Tuple[Optional[Dict[str, Any]], float]:
+    """Choose the best-fit candidate release for the on-disk files.
+
+    ``candidates`` is a list of dicts each with a ``'tracks'`` list (plus any
+    caller fields like ``source``/``album_id``, returned untouched). **Pass
+    candidates in source-priority order** — ties break toward the EARLIER one,
+    so the choice is deterministic and priority-respecting (this is what makes
+    every tool agree, #765).
+
+    Returns ``(best_candidate, score)``, or ``(None, best_score)`` when nothing
+    clears ``min_score`` — so a low-confidence guess is never pinned (the caller
+    leaves the album unresolved and falls back to today's behaviour)."""
+    best: Optional[Dict[str, Any]] = None
+    best_score = 0.0
+    for cand in candidates:
+        score = score_release_against_files(
+            file_tracks, cand.get("tracks") or [],
+            duration_tolerance_ms=duration_tolerance_ms,
+        )
+        # Strictly-greater so equal scores keep the earlier (higher-priority)
+        # candidate — deterministic tiebreak.
+        if score > best_score + 1e-9:
+            best, best_score = cand, score
+    if best is None or best_score < min_score:
+        return None, best_score
+    return best, best_score
+
+
+__all__ = ["score_release_against_files", "pick_canonical_release"]

database/music_database.py

@@ -956,6 +956,21 @@ class MusicDatabase:
             if album_cols and 'api_track_count' not in album_cols:
                 cursor.execute("ALTER TABLE albums ADD COLUMN api_track_count INTEGER DEFAULT NULL")
                 logger.info("Repaired missing api_track_count column on albums table")
+
+            # Canonical album version (#765 / #767-Bug2). Additive + nullable:
+            # a NULL canonical means "unresolved" and every tool falls back to
+            # today's behavior, so this is safe to ship dormant. Columns are
+            # populated/consumed in later stages.
+            _canonical_cols = {
+                'canonical_source': 'TEXT DEFAULT NULL',
+                'canonical_album_id': 'TEXT DEFAULT NULL',
+                'canonical_score': 'REAL DEFAULT NULL',
+                'canonical_resolved_at': 'TIMESTAMP DEFAULT NULL',
+            }
+            for _col, _typedef in _canonical_cols.items():
+                if album_cols and _col not in album_cols:
+                    cursor.execute(f"ALTER TABLE albums ADD COLUMN {_col} {_typedef}")
+                    logger.info("Added %s column to albums table (canonical version)", _col)
         except Exception as e:
             logger.error("Error repairing core media schema columns: %s", e)
 

tests/test_canonical_columns_migration.py

@@ -0,0 +1,69 @@
+"""Migration test for the canonical-album-version columns (#765 Stage 1).
+
+Additive + nullable, so it must: appear on a fresh DB, be idempotent (re-running
+the migration is a no-op, not an error), and ALTER them onto an older albums
+table that lacks them. NULL = unresolved = tools fall back to today's behavior.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+
+from database.music_database import MusicDatabase
+
+_CANONICAL_COLS = {
+    'canonical_source', 'canonical_album_id', 'canonical_score', 'canonical_resolved_at',
+}
+
+
+def _album_cols(cur):
+    cur.execute("PRAGMA table_info(albums)")
+    return {c[1] for c in cur.fetchall()}
+
+
+def test_fresh_db_has_canonical_columns(tmp_path):
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    cur = db._get_connection().cursor()
+    assert _CANONICAL_COLS <= _album_cols(cur)
+
+
+def test_canonical_columns_default_null(tmp_path):
+    # Unresolved by default -> every consumer falls back. Verify each canonical
+    # column declares DEFAULT NULL and is nullable (notnull flag == 0).
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    cur = db._get_connection().cursor()
+    cur.execute("PRAGMA table_info(albums)")
+    info = {c[1]: c for c in cur.fetchall()}  # name -> (cid, name, type, notnull, dflt, pk)
+    for col in _CANONICAL_COLS:
+        assert col in info, f"{col} missing"
+        assert info[col][3] == 0, f"{col} must be nullable"
+        dflt = info[col][4]
+        assert dflt is None or str(dflt).upper() == 'NULL', f"{col} default should be NULL"
+
+
+def test_migration_is_idempotent(tmp_path):
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    cur = db._get_connection().cursor()
+    before = _album_cols(cur)
+    # Re-running must not raise (the PRAGMA guard skips existing columns).
+    db._ensure_core_media_schema_columns(cur)
+    db._ensure_core_media_schema_columns(cur)
+    assert _album_cols(cur) == before
+    assert _CANONICAL_COLS <= _album_cols(cur)
+
+
+def test_migration_adds_columns_to_old_albums_table(tmp_path):
+    # Simulate an upgraded DB whose albums table predates these columns.
+    path = str(tmp_path / "old.db")
+    conn = sqlite3.connect(path)
+    conn.execute("CREATE TABLE albums (id INTEGER PRIMARY KEY, title TEXT)")
+    conn.commit()
+    cur = conn.cursor()
+    assert not (_CANONICAL_COLS & _album_cols(cur))  # none present yet
+
+    # Run the real migration against this old cursor.
+    db = MusicDatabase(str(tmp_path / "scratch.db"))
+    db._ensure_core_media_schema_columns(cur)
+    conn.commit()
+
+    assert _CANONICAL_COLS <= _album_cols(cur)

tests/test_canonical_version.py

@@ -0,0 +1,143 @@
+"""Extreme battery for canonical-album-version scoring (#765 / #767-Bug2).
+
+The scorer must: pick the right EDITION by best-fit to the files on disk
+(standard when you have the standard, deluxe when you have the deluxe), break
+ties deterministically toward the higher-priority candidate (so every tool
+agrees), degrade gracefully when durations/titles are missing, and never pin a
+low-confidence guess.
+"""
+
+from __future__ import annotations
+
+from core.metadata.canonical_version import (
+    pick_canonical_release,
+    score_release_against_files,
+)
+
+
+# Helpers — build track lists ----------------------------------------------
+
+def _tracks(n, base_ms=180_000, step_ms=10_000, titles=None):
+    """n tracks with distinct, deterministic durations + optional titles."""
+    out = []
+    for i in range(n):
+        t = {"duration_ms": base_ms + i * step_ms, "track_number": i + 1}
+        if titles:
+            t["title"] = titles[i]
+        out.append(t)
+    return out
+
+
+STANDARD_TITLES = [f"Song {i+1}" for i in range(11)]
+DELUXE_TITLES = STANDARD_TITLES + [f"Bonus {i+1}" for i in range(6)]
+
+
+# ── edition discrimination ────────────────────────────────────────────────
+
+def test_eleven_files_prefer_standard_over_deluxe():
+    files = _tracks(11, titles=STANDARD_TITLES)
+    standard = _tracks(11, titles=STANDARD_TITLES)
+    deluxe = _tracks(17, titles=DELUXE_TITLES)
+    s_std = score_release_against_files(files, standard)
+    s_dlx = score_release_against_files(files, deluxe)
+    assert s_std > s_dlx
+    best, score = pick_canonical_release(
+        files,
+        [{"source": "standard", "tracks": standard}, {"source": "deluxe", "tracks": deluxe}],
+    )
+    assert best["source"] == "standard" and score > 0.9
+
+
+def test_seventeen_files_prefer_deluxe():
+    files = _tracks(17, titles=DELUXE_TITLES)
+    standard = _tracks(11, titles=STANDARD_TITLES)
+    deluxe = _tracks(17, titles=DELUXE_TITLES)
+    best, _ = pick_canonical_release(
+        files,
+        # deluxe deliberately listed SECOND to prove count/fit beats order here
+        [{"source": "standard", "tracks": standard}, {"source": "deluxe", "tracks": deluxe}],
+    )
+    assert best["source"] == "deluxe"
+
+
+def test_exact_count_and_durations_scores_near_one():
+    files = _tracks(11, titles=STANDARD_TITLES)
+    assert score_release_against_files(files, _tracks(11, titles=STANDARD_TITLES)) > 0.99
+
+
+# ── deterministic tiebreak (the #765 resolution) ──────────────────────────
+
+def test_identical_releases_break_tie_to_first_candidate():
+    # Same album from two sources (same files match both equally) — must pick
+    # the FIRST (higher-priority) deterministically so both tools agree.
+    files = _tracks(11, titles=STANDARD_TITLES)
+    a = {"source": "spotify", "tracks": _tracks(11, titles=STANDARD_TITLES)}
+    b = {"source": "musicbrainz", "tracks": _tracks(11, titles=STANDARD_TITLES)}
+    best, _ = pick_canonical_release(files, [a, b])
+    assert best["source"] == "spotify"
+    # ...and stable when the order flips (priority is the caller's order).
+    best2, _ = pick_canonical_release(files, [b, a])
+    assert best2["source"] == "musicbrainz"
+
+
+# ── duration disambiguation when counts tie ───────────────────────────────
+
+def test_duration_breaks_tie_when_counts_equal():
+    # Two 11-track candidates; the files' durations match candidate A's lengths,
+    # not B's (e.g. album cuts vs radio edits). A must win on duration fit.
+    files = _tracks(11, base_ms=200_000, step_ms=5_000)
+    cand_a = {"source": "album", "tracks": _tracks(11, base_ms=200_000, step_ms=5_000)}
+    cand_b = {"source": "edits", "tracks": _tracks(11, base_ms=140_000, step_ms=5_000)}
+    best, _ = pick_canonical_release(files, [cand_b, cand_a])  # B listed first
+    assert best["source"] == "album"  # duration fit overrides order
+
+
+# ── graceful degradation ──────────────────────────────────────────────────
+
+def test_no_durations_falls_back_to_count_and_title():
+    files = [{"title": t} for t in STANDARD_TITLES]            # no durations
+    standard = [{"title": t} for t in STANDARD_TITLES]
+    deluxe = [{"title": t} for t in DELUXE_TITLES]
+    best, score = pick_canonical_release(
+        files,
+        [{"source": "standard", "tracks": standard}, {"source": "deluxe", "tracks": deluxe}],
+    )
+    assert best["source"] == "standard" and score > 0.5
+
+
+def test_only_counts_available_still_scores():
+    files = [{} for _ in range(11)]
+    assert score_release_against_files(files, [{} for _ in range(11)]) > 0.99
+    assert score_release_against_files(files, [{} for _ in range(17)]) < 0.8
+
+
+def test_fuzzy_titles_still_match():
+    files = _tracks(3, titles=["Believer", "Whatever It Takes", "Thunder"])
+    rel = _tracks(3, titles=["Believer (Remastered)", "Whatever It Takes", "Thunder!"])
+    assert score_release_against_files(files, rel) > 0.9
+
+
+# ── confidence floor / guards ─────────────────────────────────────────────
+
+def test_below_floor_returns_none():
+    files = _tracks(11, titles=STANDARD_TITLES)
+    # A wildly wrong candidate (3 unrelated tracks) must not be pinned.
+    bad = {"source": "wrong", "tracks": _tracks(3, base_ms=60_000, titles=["X", "Y", "Z"])}
+    best, score = pick_canonical_release(files, [bad])
+    assert best is None
+    assert score < 0.5
+
+
+def test_empty_inputs_are_safe():
+    assert score_release_against_files([], _tracks(11)) == 0.0
+    assert score_release_against_files(_tracks(11), []) == 0.0
+    best, score = pick_canonical_release(_tracks(11), [])
+    assert best is None and score == 0.0
+
+
+def test_min_score_is_tunable():
+    files = _tracks(11, titles=STANDARD_TITLES)
+    near = {"source": "near", "tracks": _tracks(10, titles=STANDARD_TITLES[:10])}
+    # default floor accepts a 10/11 fit, a strict floor rejects it
+    assert pick_canonical_release(files, [near])[0] is not None
+    assert pick_canonical_release(files, [near], min_score=0.99)[0] is None