#889 Phase 2: import seam — hint short-circuits identification, replaces on success

When a staged single-file candidate carries a re-identify hint, the worker builds
the identification straight from the user-chosen release (album_id/source) and
skips the guessing tiers — so the ambiguity that mis-filed the track is gone. No
hint → byte-identical to before (the lookup returns (None, None), fail-safe on any
DB error). A hinted import auto-processes (explicit user choice), still gated on the
global auto_process pref.

After the re-import lands, _finalize_rematch_hint consumes the hint and (if replace
was chosen) deletes the old row + file via delete_replaced_track — deferred to
success so a failed import never loses the original. Safe by construction: unlink
only when no surviving row references the file, and the modal never offers the
track's current release so old path != new path.

All hint logic lives in auto_import_worker.py + the pure rematch_hints helpers —
pipeline.py / side_effects.py untouched. 18 tests; full auto-import suite green.

core/auto_import_worker.py

@@ -660,8 +660,13 @@ class AutoImportWorker:
             auto_process = self._config_manager.get('auto_import.auto_process', True)
 
         try:
-            # Phase 3: Identify
+            # Phase 3: Identify.
-            identification = self._identify_folder(candidate)
+            # Re-identify (#889): if the user designated this exact file's release in
+            # the Re-identify modal, a hint short-circuits the guessing — we match
+            # straight against the chosen album. No hint → byte-identical to before.
+            rematch_hint, identification = self._resolve_rematch_hint(candidate)
+            if identification is None:
+                identification = self._identify_folder(candidate)
             if not identification:
                 self._record_result(candidate, 'needs_identification', 0.0,
                                     error_message='Could not identify album from tags, folder name, or fingerprint')
@@ -690,7 +695,10 @@ class AutoImportWorker:
             high_conf_matches = [m for m in match_result.get('matches', []) if m['confidence'] >= 0.8]
             has_strong_individual_matches = len(high_conf_matches) > 0
 
-            if (confidence >= threshold or has_strong_individual_matches) and auto_process:
+            # A re-identify is an explicit user choice — let it auto-process like a
+            # strong match (still gated on the global auto_process preference).
+            if (confidence >= threshold or has_strong_individual_matches
+                    or rematch_hint is not None) and auto_process:
                 # Phase 5: Auto-process — insert an in-progress row
                 # so the UI sees the import the moment it starts,
                 # then update it with the final status when done.
@@ -709,6 +717,11 @@ class AutoImportWorker:
                 confidence = max(confidence, effective_conf)
                 if success:
                     self._bump_stat('auto_processed')
+                    # Re-identify (#889): only NOW that the new home exists do we
+                    # consume the hint and (if replace was chosen) delete the old
+                    # row + file — so a failed import never loses the original.
+                    if rematch_hint is not None:
+                        self._finalize_rematch_hint(rematch_hint)
                 else:
                     self._bump_stat('failed')
 
@@ -1003,6 +1016,62 @@ class AutoImportWorker:
         except Exception:
             return False
 
+    # ── Re-identify hints (#889) ──
+
+    def _resolve_rematch_hint(self, candidate: 'FolderCandidate'):
+        """If this staged file carries a user-designated re-identify hint, return
+        ``(hint, identification)`` so matching skips the guessing tiers; otherwise
+        ``(None, None)`` and the caller falls back to normal identification.
+
+        Fail-safe: ANY error (no table, DB hiccup) returns ``(None, None)`` so a
+        re-identify problem can never break ordinary auto-import. Only single-file
+        candidates are eligible — a re-identify always stages exactly one track."""
+        try:
+            files = candidate.audio_files or []
+            if len(files) != 1:
+                return None, None
+            from core.imports.rematch_hints import (
+                build_identification_from_hint,
+                find_hint_for_file,
+                quick_file_signature,
+            )
+            file_path = files[0]
+            sig = quick_file_signature(file_path)
+            conn = self.database._get_connection()
+            try:
+                cursor = conn.cursor()
+                hint = find_hint_for_file(cursor, file_path, sig)
+            finally:
+                conn.close()
+            if hint is None:
+                return None, None
+            logger.info("[Auto-Import] Re-identify hint for %s → %s '%s' (%s)",
+                        candidate.name, hint.album_type or 'release',
+                        hint.album_name or '?', hint.source)
+            return hint, build_identification_from_hint(hint)
+        except Exception as e:
+            logger.debug("[Auto-Import] rematch-hint lookup skipped: %s", e)
+            return None, None
+
+    def _finalize_rematch_hint(self, hint) -> None:
+        """Post-success: delete the replaced library row + file (if the user chose
+        replace) and consume the hint so it's single-use. Best-effort — a cleanup
+        failure is logged, never raised, since the re-import already succeeded."""
+        try:
+            from core.imports.rematch_hints import consume_hint, delete_replaced_track
+            conn = self.database._get_connection()
+            try:
+                cursor = conn.cursor()
+                removed = delete_replaced_track(cursor, hint.replace_track_id)
+                consume_hint(cursor, hint.id)
+                conn.commit()
+            finally:
+                conn.close()
+            if removed:
+                logger.info("[Auto-Import] Re-identify replaced old track — removed %s", removed)
+        except Exception as e:
+            logger.warning("[Auto-Import] rematch-hint finalize failed (import still OK): %s", e)
+
     # ── Identification ──
 
     def _identify_folder(self, candidate: FolderCandidate) -> Optional[Dict]:

core/imports/rematch_hints.py

@@ -193,6 +193,65 @@ def list_pending_hints(cursor: Any) -> list:
     return [_row_to_hint(r) for r in cursor.fetchall()]
 
 
+def build_identification_from_hint(hint: RematchHint) -> dict:
+    """Turn a hint into the ``identification`` dict the auto-import matcher expects,
+    so a re-identify SKIPS the guessing tiers entirely and matches straight against
+    the user-chosen release. Mirrors the shape `_identify_folder` returns (album_id
+    / source / track_number drive the album fetch + file→track match)."""
+    return {
+        "album_id": hint.album_id or None,
+        "album_name": hint.album_name or hint.track_title or "",
+        "artist_name": hint.artist_name or "",
+        "artist_id": hint.artist_id or "",
+        "track_name": hint.track_title or "",
+        "track_id": hint.track_id or "",
+        "image_url": "",
+        "release_date": "",
+        "track_number": hint.track_number or 1,
+        "total_tracks": 1,
+        "source": hint.source,
+        "method": "rematch_hint",
+        "identification_confidence": 1.0,
+        "is_single": True,
+        "album_type": hint.album_type,
+    }
+
+
+def delete_replaced_track(cursor: Any, replace_track_id: Any, *, unlink=os.remove) -> Optional[str]:
+    """Remove the OLD library row a re-identify replaces, and its file.
+
+    Called only AFTER the re-import has landed the track at its new home, so the
+    original is never lost on failure. Safe by construction: the file is unlinked
+    only if it still exists and **no other track row references it** (guards against
+    yanking a file a different row legitimately points to). Returns the path it
+    removed, or ``None`` if there was nothing to do. ``unlink`` is injectable for
+    tests. Assumes the new home differs from the old — the Re-identify modal never
+    offers the track's current release as a target, so this holds."""
+    if not replace_track_id:
+        return None
+    cursor.execute("SELECT file_path FROM tracks WHERE id = ?", (replace_track_id,))
+    row = cursor.fetchone()
+    if row is None:
+        return None
+    old_path = (row["file_path"] if not isinstance(row, (tuple, list)) else row[0]) or ""
+
+    cursor.execute("DELETE FROM tracks WHERE id = ?", (replace_track_id,))
+
+    if not old_path:
+        return None
+    # Only unlink if no surviving row still points at this file.
+    cursor.execute("SELECT 1 FROM tracks WHERE file_path = ? LIMIT 1", (old_path,))
+    if cursor.fetchone() is not None:
+        return None
+    try:
+        if os.path.exists(old_path):
+            unlink(old_path)
+            return old_path
+    except OSError:
+        pass
+    return None
+
+
 def quick_file_signature(path: str, *, chunk: int = 65536) -> Optional[str]:
     """A cheap, rename-proof content fingerprint: size + first/last chunk, hashed.
 
@@ -222,5 +281,7 @@ __all__ = [
     "find_hint_for_file",
     "consume_hint",
     "list_pending_hints",
+    "build_identification_from_hint",
+    "delete_replaced_track",
     "quick_file_signature",
 ]

tests/imports/test_rematch_hints_seam.py

@@ -0,0 +1,168 @@
+"""#889 Phase 2: the import seam — a hint short-circuits identification, and the
+old library row is replaced only after the re-import succeeds.
+
+Two layers:
+  * pure helpers (build_identification_from_hint, delete_replaced_track) — exact
+    mapping + safe replacement against an in-memory DB, injectable unlink.
+  * the worker seam (_resolve_rematch_hint / _finalize_rematch_hint) — proves the
+    NO-HINT path is untouched, the hint path returns a ready identification, the
+    lookup is fail-safe, and finalize consumes + replaces.
+"""
+
+from __future__ import annotations
+
+import sqlite3
+import types
+
+import pytest
+
+from core.auto_import_worker import AutoImportWorker, FolderCandidate
+from core.imports.rematch_hints import (
+    RematchHint,
+    build_identification_from_hint,
+    consume_hint,
+    create_hint,
+    delete_replaced_track,
+    find_hint_for_file,
+)
+
+_SCHEMA = """
+CREATE TABLE rematch_hints (
+    id INTEGER PRIMARY KEY AUTOINCREMENT, staged_path TEXT NOT NULL, content_hash TEXT,
+    source TEXT NOT NULL, isrc TEXT, track_id TEXT, album_id TEXT, artist_id TEXT,
+    track_title TEXT, album_name TEXT, artist_name TEXT, album_type TEXT,
+    track_number INTEGER, disc_number INTEGER, replace_track_id INTEGER,
+    exempt_dedup INTEGER NOT NULL DEFAULT 1, status TEXT NOT NULL DEFAULT 'pending',
+    created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, consumed_at TIMESTAMP
+);
+CREATE TABLE tracks (
+    id INTEGER PRIMARY KEY, album_id INTEGER, artist_id INTEGER, title TEXT,
+    track_number INTEGER, file_path TEXT
+);
+"""
+
+
+@pytest.fixture
+def conn():
+    c = sqlite3.connect(":memory:")
+    c.row_factory = sqlite3.Row
+    c.executescript(_SCHEMA)
+    yield c
+    c.close()
+
+
+def _hint(**kw):
+    base = dict(staged_path="/staging/Song.flac", source="spotify", album_id="alb_album1",
+                artist_id="art_1", track_id="trk_1", track_title="Song", album_name="Album1",
+                artist_name="Artist", album_type="album", track_number=5, disc_number=1)
+    base.update(kw)
+    return RematchHint(**base)
+
+
+# ── pure: identification mapping ──────────────────────────────────────────────
+def test_build_identification_maps_hint_fields():
+    ident = build_identification_from_hint(_hint())
+    assert ident["album_id"] == "alb_album1"
+    assert ident["source"] == "spotify"
+    assert ident["album_name"] == "Album1"
+    assert ident["artist_id"] == "art_1"
+    assert ident["track_number"] == 5
+    assert ident["method"] == "rematch_hint"
+    assert ident["identification_confidence"] == 1.0
+    assert ident["is_single"] is True
+
+
+# ── pure: safe replacement ────────────────────────────────────────────────────
+def test_delete_replaced_track_removes_row_and_file(conn):
+    cur = conn.cursor()
+    cur.execute("INSERT INTO tracks (id, file_path) VALUES (7, '/lib/EP1/05 - Song.flac')")
+    removed = []
+    out = delete_replaced_track(cur, 7, unlink=lambda p: removed.append(p))
+    assert out == "/lib/EP1/05 - Song.flac"
+    assert removed == ["/lib/EP1/05 - Song.flac"]   # file removed (we faked existence below)
+    cur.execute("SELECT 1 FROM tracks WHERE id = 7")
+    assert cur.fetchone() is None                    # row gone
+
+
+def test_delete_replaced_track_keeps_file_if_another_row_points_at_it(conn):
+    cur = conn.cursor()
+    cur.execute("INSERT INTO tracks (id, file_path) VALUES (7, '/lib/shared.flac')")
+    cur.execute("INSERT INTO tracks (id, file_path) VALUES (8, '/lib/shared.flac')")
+    removed = []
+    out = delete_replaced_track(cur, 7, unlink=lambda p: removed.append(p))
+    assert out is None and removed == []             # row 8 still references it → no unlink
+    cur.execute("SELECT 1 FROM tracks WHERE id = 7")
+    assert cur.fetchone() is None                     # but row 7 still deleted
+
+
+def test_delete_replaced_track_noops_on_missing_id(conn):
+    cur = conn.cursor()
+    assert delete_replaced_track(cur, None) is None
+    assert delete_replaced_track(cur, 999) is None    # no such row
+
+
+# patch os.path.exists so the unlink branch is reachable without real files
+@pytest.fixture(autouse=True)
+def _exists(monkeypatch):
+    monkeypatch.setattr("core.imports.rematch_hints.os.path.exists", lambda p: True)
+
+
+# ── worker seam ───────────────────────────────────────────────────────────────
+def _worker(conn):
+    # Production hands out a FRESH connection per call (the worker closes it);
+    # here we share one in-memory DB, so proxy close() to a no-op.
+    w = AutoImportWorker.__new__(AutoImportWorker)
+    proxy = types.SimpleNamespace(cursor=conn.cursor, commit=conn.commit, close=lambda: None)
+    w.database = types.SimpleNamespace(_get_connection=lambda: proxy)
+    return w
+
+
+def test_resolve_returns_none_when_no_hint(conn):
+    w = _worker(conn)
+    cand = FolderCandidate(path="/staging", name="Song", audio_files=["/staging/Song.flac"])
+    assert w._resolve_rematch_hint(cand) == (None, None)   # untouched → normal identify
+
+
+def test_resolve_returns_identification_when_hinted(conn, monkeypatch):
+    # don't hash a real file
+    monkeypatch.setattr("core.imports.rematch_hints.quick_file_signature", lambda p: None)
+    create_hint(conn.cursor(), _hint(staged_path="/staging/Song.flac"))
+    conn.commit()
+    w = _worker(conn)
+    cand = FolderCandidate(path="/staging", name="Song", audio_files=["/staging/Song.flac"])
+    hint, ident = w._resolve_rematch_hint(cand)
+    assert hint is not None and hint.album_id == "alb_album1"
+    assert ident["album_id"] == "alb_album1" and ident["method"] == "rematch_hint"
+
+
+def test_resolve_ignores_multi_file_candidates(conn):
+    create_hint(conn.cursor(), _hint(staged_path="/staging/Song.flac"))
+    conn.commit()
+    w = _worker(conn)
+    cand = FolderCandidate(path="/staging", name="Album",
+                           audio_files=["/staging/Song.flac", "/staging/Other.flac"])
+    assert w._resolve_rematch_hint(cand) == (None, None)   # re-identify is single-track only
+
+
+def test_resolve_is_failsafe_on_db_error():
+    w = AutoImportWorker.__new__(AutoImportWorker)
+    def _boom():
+        raise RuntimeError("db down")
+    w.database = types.SimpleNamespace(_get_connection=_boom)
+    cand = FolderCandidate(path="/staging", name="Song", audio_files=["/staging/Song.flac"])
+    assert w._resolve_rematch_hint(cand) == (None, None)   # error never breaks auto-import
+
+
+def test_finalize_consumes_and_replaces(conn, monkeypatch):
+    monkeypatch.setattr("core.imports.rematch_hints.quick_file_signature", lambda p: None)
+    cur = conn.cursor()
+    cur.execute("INSERT INTO tracks (id, file_path) VALUES (42, '/lib/EP1/05 - Song.flac')")
+    hid = create_hint(cur, _hint(replace_track_id=42))
+    conn.commit()
+    w = _worker(conn)
+    hint = find_hint_for_file(conn.cursor(), "/staging/Song.flac")
+    w._finalize_rematch_hint(hint)
+    # old row deleted, hint consumed
+    cur.execute("SELECT 1 FROM tracks WHERE id = 42")
+    assert cur.fetchone() is None
+    assert find_hint_for_file(conn.cursor(), "/staging/Song.flac") is None   # consumed