diff --git a/core/metadata/canonical_resolver.py b/core/metadata/canonical_resolver.py
index b671a759..a240870a 100644
--- a/core/metadata/canonical_resolver.py
+++ b/core/metadata/canonical_resolver.py
@@ -73,4 +73,90 @@ def resolve_canonical_for_album(
     }
 
 
-__all__ = ["resolve_canonical_for_album"]
+def default_fetch_tracklist(source: str, album_id: str) -> Optional[List[Dict[str, Any]]]:
+    """Production ``fetch_tracklist``: pull a release's tracklist from a metadata
+    source and normalise to ``{title, track_number, duration_ms}``. Duration is
+    best-effort (not every source exposes it); when absent the scorer just leans
+    on track-count + title. Returns None on any failure."""
+    try:
+        from core.metadata_service import get_album_tracks_for_source
+        data = get_album_tracks_for_source(source, album_id)
+    except Exception:
+        return None
+    items = data if isinstance(data, list) else (
+        (data.get('items') or data.get('tracks') or []) if isinstance(data, dict) else []
+    )
+    if isinstance(items, dict):  # {'tracks': {'items': [...]}}
+        items = items.get('items') or []
+    out: List[Dict[str, Any]] = []
+    for it in items:
+        get = it.get if isinstance(it, dict) else (lambda k, d=None: getattr(it, k, d))
+        dur = get('duration_ms')
+        if dur is None:
+            secs = get('duration')  # some sources give seconds
+            dur = int(secs * 1000) if isinstance(secs, (int, float)) and secs else None
+        out.append({
+            'title': get('name') or get('title') or '',
+            'track_number': get('track_number'),
+            'duration_ms': dur,
+        })
+    return out or None
+
+
+def resolve_and_store_canonical_for_album(
+    db,
+    album_id,
+    *,
+    fetch_tracklist: Optional[Callable[[str, str], Any]] = None,
+    source_priority: Optional[List[str]] = None,
+    min_score: float = 0.5,
+) -> Optional[Dict[str, Any]]:
+    """Gather an album's source IDs + its tracks' (duration, title) from the DB,
+    resolve the best-fit canonical release, and persist it. Returns the stored
+    ``{source, album_id, score}`` or None when unresolved.
+
+    Uses the SAME album/source-id loader the Reorganizer uses
+    (``load_album_and_tracks`` + ``_extract_source_ids``) so the canonical is
+    chosen over exactly the source IDs the reorganizer sees. Scores off the DB
+    track rows' ``duration`` (stored in ms) + ``title`` — the library's view of
+    the files — so no per-file disk reads are needed."""
+    from core.library_reorganize import _extract_source_ids, load_album_and_tracks
+
+    album_data, tracks = load_album_and_tracks(db, album_id)
+    if not album_data or not tracks:
+        return None
+    source_ids = {s: v for s, v in _extract_source_ids(album_data).items() if v}
+    if not source_ids:
+        return None
+
+    file_tracks = [
+        {'duration_ms': t.get('duration') or 0, 'title': t.get('title') or ''}
+        for t in tracks
+    ]
+
+    if fetch_tracklist is None:
+        fetch_tracklist = default_fetch_tracklist
+    if source_priority is None:
+        try:
+            from core.metadata_service import get_primary_source, get_source_priority
+            source_priority = get_source_priority(get_primary_source())
+        except Exception:
+            source_priority = list(source_ids.keys())
+
+    result = resolve_canonical_for_album(
+        album_source_ids=source_ids,
+        file_tracks=file_tracks,
+        fetch_tracklist=fetch_tracklist,
+        source_priority=source_priority,
+        min_score=min_score,
+    )
+    if result:
+        db.set_album_canonical(album_id, result['source'], result['album_id'], result['score'])
+    return result
+
+
+__all__ = [
+    "resolve_canonical_for_album",
+    "resolve_and_store_canonical_for_album",
+    "default_fetch_tracklist",
+]
diff --git a/tests/test_canonical_orchestration.py b/tests/test_canonical_orchestration.py
new file mode 100644
index 00000000..2f3ae8a0
--- /dev/null
+++ b/tests/test_canonical_orchestration.py
@@ -0,0 +1,97 @@
+"""End-to-end orchestration for canonical resolve+store (#765 Stage 2 trigger).
+
+Uses a real temp DB (album + tracks + source IDs) and an INJECTED fetcher, so
+the DB gathering + persistence are exercised for real without live APIs.
+"""
+
+from __future__ import annotations
+
+from core.metadata.canonical_resolver import (
+    default_fetch_tracklist,
+    resolve_and_store_canonical_for_album,
+)
+from database.music_database import MusicDatabase
+
+STD = [{"duration_ms": 180_000 + i * 10_000, "title": f"Song {i+1}", "track_number": i + 1} for i in range(11)]
+DLX = STD + [{"duration_ms": 320_000 + i * 10_000, "title": f"Bonus {i+1}", "track_number": 12 + i} for i in range(6)]
+
+
+def _seed(db, *, spotify=None, deezer=None, n_files=11):
+    """Insert an album (with given source IDs) + n_files tracks whose
+    durations/titles match the STANDARD release."""
+    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, spotify_album_id, deezer_id) "
+        "VALUES ('alb1', 'Evolve', 'art1', ?, ?)",
+        (spotify, deezer),
+    )
+    for i in range(n_files):
+        cur.execute(
+            "INSERT INTO tracks (id, album_id, artist_id, title, track_number, duration) "
+            "VALUES (?, 'alb1', 'art1', ?, ?, ?)",
+            (f"t{i}", f"Song {i+1}", i + 1, 180_000 + i * 10_000),
+        )
+    conn.commit()
+    conn.close()
+    return "alb1"
+
+
+def test_resolve_and_store_picks_best_fit_and_persists(tmp_path):
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    album_id = _seed(db, spotify="sp_deluxe", deezer="dz_std")  # 11 files
+
+    table = {("spotify", "sp_deluxe"): DLX, ("deezer", "dz_std"): STD}
+    out = resolve_and_store_canonical_for_album(
+        db, album_id,
+        fetch_tracklist=lambda s, a: table.get((s, a)),
+        source_priority=["spotify", "deezer"],
+    )
+    # Deezer's standard matches the 11 files better than Spotify's deluxe.
+    assert out["source"] == "deezer" and out["album_id"] == "dz_std"
+    # ...and it was persisted.
+    stored = db.get_album_canonical(album_id)
+    assert stored["source"] == "deezer" and stored["album_id"] == "dz_std"
+
+
+def test_resolve_returns_none_when_album_has_no_source_ids(tmp_path):
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    album_id = _seed(db, spotify=None, deezer=None)
+    out = resolve_and_store_canonical_for_album(
+        db, album_id, fetch_tracklist=lambda s, a: STD, source_priority=["spotify"],
+    )
+    assert out is None
+    assert db.get_album_canonical(album_id) is None
+
+
+def test_resolve_returns_none_for_missing_album(tmp_path):
+    db = MusicDatabase(str(tmp_path / "m.db"))
+    out = resolve_and_store_canonical_for_album(
+        db, "does-not-exist", fetch_tracklist=lambda s, a: STD, source_priority=["spotify"],
+    )
+    assert out is None
+
+
+# ── default_fetch_tracklist normalization (no DB / no live API) ────────────
+
+def test_default_fetcher_normalizes_dict_items(monkeypatch):
+    import core.metadata_service as ms
+    monkeypatch.setattr(
+        ms, "get_album_tracks_for_source",
+        lambda s, a: [{"name": "A", "track_number": 1, "duration_ms": 200000},
+                      {"title": "B", "track_number": 2, "duration": 210}],  # seconds
+        raising=False,
+    )
+    out = default_fetch_tracklist("spotify", "x")
+    assert out[0] == {"title": "A", "track_number": 1, "duration_ms": 200000}
+    assert out[1] == {"title": "B", "track_number": 2, "duration_ms": 210_000}  # sec->ms
+
+
+def test_default_fetcher_handles_failure(monkeypatch):
+    import core.metadata_service as ms
+    monkeypatch.setattr(
+        ms, "get_album_tracks_for_source",
+        lambda s, a: (_ for _ in ()).throw(RuntimeError("boom")), raising=False,
+    )
+    assert default_fetch_tracklist("spotify", "x") is None