Canonical album version — Stage 4: Track Number Repair prefers canonical (read)

_resolve_album_tracklist gains a Fallback -1: if the album has a pinned
canonical (source, album_id), use it before the existing 6-level cascade — so
Track Number Repair resolves the SAME release the Reorganizer does (Stage 3) and
the two stop contradicting each other (#765, the Spotify-4 vs MusicBrainz-3
conflict).

Gated + additive: the entire existing cascade is untouched for albums without a
canonical, so this job's all-01-album rescue (which relies on the MusicBrainz/
AudioDB fallbacks for albums with no DB source ID) is fully preserved — that's
the regression we explicitly refused to take in a reactive fix.

New helper _lookup_canonical_from_db() mirrors _lookup_album_ids_from_db
(file-path -> track -> album), returns None when no DB / no match / columns
absent / unresolved.

Tests: tests/test_track_repair_canonical.py (4) — returns canonical when pinned,
None when unresolved / file untracked / no DB. Existing track_number_repair
tests still pass (no regression).

core/repair_jobs/track_number_repair.py

@@ -275,6 +275,22 @@ class TrackNumberRepairJob(RepairJob):
         primary_source = get_primary_source()
         source_priority = get_source_priority(primary_source)
 
+        # Fallback -1 (#765): a pinned canonical release wins over the whole
+        # cascade below — so Track Number Repair resolves the SAME release the
+        # Reorganizer does (Stage 3) and the two stop contradicting each other.
+        # Gated on the album carrying a canonical; everything below is untouched
+        # for albums without one (preserving the all-01-album rescue this job
+        # exists for — the regression we refused to take in a reactive fix).
+        canonical = _lookup_canonical_from_db(file_track_data, context)
+        if canonical:
+            c_source, c_id = canonical
+            if _is_valid_album_id(c_id):
+                tracks = _get_album_tracklist(c_source, c_id, cache)
+                if tracks:
+                    logger.info("[Repair] %s — resolved via canonical %s album ID: %s",
+                                folder_name, c_source, c_id)
+                    return tracks
+
         # Fallback 0: Check DB first. If any tracked file already has source IDs,
         # prefer the configured source order and use the first available album ID.
         source_album_ids = _lookup_album_ids_from_db(file_track_data, context)
@@ -806,6 +822,46 @@ def _update_db_file_path(db, old_path: str, new_path: str):
             conn.close()
 
 
+def _lookup_canonical_from_db(file_track_data: List[Tuple[str, str, Any]],
+                              context: JobContext) -> Optional[Tuple[str, str]]:
+    """Return the album's pinned canonical ``(source, album_id)`` or None.
+
+    #765: when the album this folder's files belong to has a canonical release
+    pinned (best-fit to the files), Track Number Repair uses it first so it
+    agrees with the Reorganizer. Resolves by matching a file path to its DB
+    track row. None when no DB, no match, columns absent, or unresolved."""
+    if not context.db:
+        return None
+    conn = None
+    try:
+        conn = context.db._get_connection()
+        cursor = conn.cursor()
+        cursor.execute("PRAGMA table_info(albums)")
+        cols = {row[1] for row in cursor.fetchall()}
+        if 'canonical_source' not in cols or 'canonical_album_id' not in cols:
+            return None
+        for fpath, _, _ in file_track_data:
+            cursor.execute(
+                """
+                SELECT al.canonical_source, al.canonical_album_id
+                FROM tracks t
+                JOIN albums al ON al.id = t.album_id
+                WHERE t.file_path = ?
+                LIMIT 1
+                """,
+                (fpath,),
+            )
+            row = cursor.fetchone()
+            if row and row[0] and row[1]:
+                return (str(row[0]), str(row[1]))
+    except Exception as e:
+        logger.debug("Error looking up canonical from DB: %s", e)
+    finally:
+        if conn:
+            conn.close()
+    return None
+
+
 def _lookup_album_ids_from_db(file_track_data: List[Tuple[str, str, Any]],
                               context: JobContext) -> Dict[str, Optional[str]]:
     """Look up album IDs from the database using file paths.

tests/test_track_repair_canonical.py

@@ -0,0 +1,52 @@
+"""Track Number Repair canonical lookup (#765 Stage 4, read side)."""
+
+from __future__ import annotations
+
+import types
+
+from core.repair_jobs.track_number_repair import _lookup_canonical_from_db
+from database.music_database import MusicDatabase
+
+
+def _ctx(db):
+    return types.SimpleNamespace(db=db)
+
+
+def _seed(db, *, with_canonical: bool, file_path: str = "/music/Evolve/01 - Believer.flac"):
+    conn = db._get_connection()
+    cur = conn.cursor()
+    cur.execute("INSERT INTO artists (id, name) VALUES ('art1', 'Imagine Dragons')")
+    cur.execute("INSERT INTO albums (id, title, artist_id) VALUES ('alb1', 'Evolve', 'art1')")
+    cur.execute(
+        "INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration, file_path) "
+        "VALUES ('t1', 'alb1', 'art1', 'Believer', 1, 204000, ?)",
+        (file_path,),
+    )
+    conn.commit()
+    conn.close()
+    if with_canonical:
+        db.set_album_canonical("alb1", "spotify", "sp_evolve", 0.96)
+
+
+def test_returns_canonical_when_pinned(tmp_path):
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    fp = "/music/Evolve/01 - Believer.flac"
+    _seed(db, with_canonical=True, file_path=fp)
+    assert _lookup_canonical_from_db([(fp, "01 - Believer.flac", 1)], _ctx(db)) == ("spotify", "sp_evolve")
+
+
+def test_none_when_unresolved(tmp_path):
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    fp = "/music/Evolve/01 - Believer.flac"
+    _seed(db, with_canonical=False, file_path=fp)
+    assert _lookup_canonical_from_db([(fp, "01 - Believer.flac", 1)], _ctx(db)) is None
+
+
+def test_none_when_file_not_tracked(tmp_path):
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    _seed(db, with_canonical=True)
+    assert _lookup_canonical_from_db([("/some/other/path.flac", "x.flac", 1)], _ctx(db)) is None
+
+
+def test_none_when_no_db():
+    assert _lookup_canonical_from_db([("/p.flac", "p.flac", 1)], types.SimpleNamespace(db=None)) is None