fix(scan): one prompt per file for recorded-run links; writes survive a library move

The recorded run's stored prompt now links in one transaction with the file's
link rows, so a file linked to another prompt is a conflict instead of a
second link. The scan's stores run through survive_library_switch. A single
file asks history once (a one-file _BatchHistory), the scan passes
in_output=False and reads the file size only for files with prompt
metadata. A conflict row reads "Kept: already linked to a different
prompt". TESTING.md records the rules.

TESTING.md

@@ -5050,12 +5050,14 @@ workflows without a PromptManager node. It does no file or database I/O.
   pair, A1111, sampler-less). Only a graph it leaves ambiguous or unresolved
   reaches the inspector's resolve (`inspection_resolution` in
   `utils/prompt_inspector.py`, the inspection profile, then the recorded saver
-  anchor of #386; never the file-name saver anchor: a file only its saver's
-  file name resolves is queued for Needs review, never linked, the #390 ruling,
-  `name_anchor=False`), and only that positive's `storage_text` is stored (never
+  anchor of #386; `name_anchor=False`: the file-name anchor never decides a
+  link; the scan's legacy floor may still link the file, as before #247, and a
+  file nothing links is queued with the anchor's choice per #390), and only that positive's `storage_text` is stored (never
   prepend or append text, never `recorded_texts`). Then the #170 browser parser
   (`graph_prompt_text`), the legacy floor, then the recorded run's stored
-  prompt. History is asked once per file and only at that step; the scan's
+  prompt. The legacy answer is refused when a PromptManager-family node has a
+  prepend or append part and the answer is not that node's own `text` (the
+  parser joins the parts, so the stored text cannot be told from it). History is asked once per file and only at that step; the scan's
   batch lookup stays lazy and makes one `recorded_runs` pass for the stored
   prompt and the saver (`_BatchHistory`).
 - **Policies.** `SCAN` keeps the legacy floor for every undecided file.
@@ -5064,14 +5066,18 @@ workflows without a PromptManager node. It does no file or database I/O.
   malformed one still gets the legacy floor.
 - **Junk.** `junk_text` refuses `blank`, `number` (the #252 `_numeric` rule),
   `lora-only` (only `<lora:...>` tags and separators) and `model-file` (one
-  token ending in a model extension) at whichever step found the text; the
-  file reports `found` and nothing later is tried.
+  token ending in a model extension) at whichever step found the text and
+  nothing later is tried; the file reports `found`, except at the undecided
+  graph step, which keeps its own status (an ambiguous file is still queued).
 - **One prompt per file.** `database/scanned_prompts.py` stores a positive in
   one `BEGIN IMMEDIATE` transaction: a file already linked to another prompt
   (either spelling, `file_link_rows`/`link_conflict`, the pair ingester's rule)
   is `conflict` and nothing is written; a failed link rolls the new prompt back
-  (`found`); two writers of one text store one prompt. A rescan therefore never
-  adds a second link to a file an older scan linked to the legacy answer.
+  (`found`); two writers of one text store one prompt. A recorded run's stored
+  prompt links the same way (`link_stored_prompt_or_raise`). A rescan therefore
+  never adds a second link to a file an older scan linked to the legacy answer;
+  the row reads "Kept: already linked to a different prompt". Every scan write
+  runs through `survive_library_switch`, so a library move reruns it.
 - **Marks.** Every prompt ingestion creates, pair components included, carries
   category `scanned`, tag `auto-scanned` and the note `Auto-scanned from
   <file>` (`SCAN_MARKS`); a reused prompt keeps its fields. Run counts are

database/scanned_prompts.py

@@ -47,11 +47,41 @@ def store_scanned_prompt_or_raise(
 def store_scanned_prompt(
     db: Any, path: str, canonical: Dict[str, Any], text: str, marks: Marks
 ) -> Tuple[str, Optional[int]]:
-    """The scan's form: an error is logged and reported as ``("found", None)``."""
+    """The scan's form: an error is logged and reported as ``("found", None)``.
+
+    A write that failed because Move library switched runs again on the new
+    library (``survive_library_switch``); the rerun re-checks the link rows.
+    """
     try:
-        return store_scanned_prompt_or_raise(db, path, canonical, text, marks)
+        return db.survive_library_switch(
+            lambda: store_scanned_prompt_or_raise(db, path, canonical, text, marks)
+        )
     except Exception as error:
         _LOGGER.error(
             f"Failed to store the prompt of {os.path.basename(path)}: {error}"
         )
         return "found", None
+
+
+def link_stored_prompt_or_raise(
+    db: Any, path: str, canonical: Dict[str, Any], prompt_id: int
+) -> Optional[str]:
+    """Link ``path`` to the stored prompt ``prompt_id`` (recorded history, #342).
+
+    ``linked``, ``unchanged`` (already linked to it), ``conflict`` (linked to
+    another prompt; nothing written) or None when the prompt is gone. One
+    ``BEGIN IMMEDIATE`` transaction; database errors are raised.
+    """
+    with db._pair_transaction() as conn:
+        rows = file_link_rows(conn, path)
+        if len(rows) > 1 or (rows and rows[0]["prompt_id"] != prompt_id):
+            return "conflict"
+        if rows:
+            return "unchanged"
+        exists = conn.execute(
+            "SELECT 1 FROM prompts WHERE id = ?", (prompt_id,)
+        ).fetchone()
+        if exists is None:
+            return None
+        db._insert_image_link(conn, prompt_id, path, canonical)
+        return "linked"

py/api/admin.py

@@ -44,26 +44,26 @@ try:
     from ...database.run_attribution import (
         RecordedOutput,
         host_output_directory,
-        in_output_folder,
         named_prompt,
-        recorded_output_lookup,
         recorded_runs,
-        stored_run_prompt,
     )
-    from ...database.scanned_prompts import store_scanned_prompt
+    from ...database.scanned_prompts import (
+        link_stored_prompt_or_raise,
+        store_scanned_prompt,
+    )
     from ...utils.file_prompt import NEVER, PAIR, Policy, decide_file_prompt
 except ImportError:
     from database.pair_ingestion import SCAN_MARKS, ingest_resolved_pair
     from database.run_attribution import (
         RecordedOutput,
         host_output_directory,
-        in_output_folder,
         named_prompt,
-        recorded_output_lookup,
         recorded_runs,
-        stored_run_prompt,
     )
-    from database.scanned_prompts import store_scanned_prompt
+    from database.scanned_prompts import (
+        link_stored_prompt_or_raise,
+        store_scanned_prompt,
+    )
     from utils.file_prompt import NEVER, PAIR, Policy, decide_file_prompt
 
 
@@ -250,7 +250,7 @@ SCAN_DETAILS = {
     "skipped": "Skipped · already linked",
     "ambiguous": "Needs review: more than one prompt could match",
     "unresolved": "Needs review: the prompt could not be traced in the workflow",
-    "conflict": "Needs review: already linked to a different prompt",
+    "conflict": "Kept: already linked to a different prompt",
     "failed": "Error: the prompt could not be stored",
     "error": "Error: the file could not be processed (see the server log)",
     "found": "Metadata found, but no prompt text to store",
@@ -1709,16 +1709,14 @@ class AdminRoutesMixin:
         the batch's lazy recorded-history lookup.
         """
         path = str(media_file)
-        if history is not None:
-            lookup = functools.partial(history.recorded, path)
-        else:
-            lookup = functools.partial(recorded_output_lookup(self.db), path)
+        if history is None:  # one file: history is still asked once
+            history = _BatchHistory(self, [media_file])
         decision = decide_file_prompt(
             metadata,
             file_name=Path(path).name,
-            file_size=_file_size(media_file),
-            in_output=in_output_folder(path, host_output_directory()),
-            lookup=lookup,
+            file_size=functools.partial(_file_size, media_file),
+            in_output=False,  # only the monitor policy reads it
+            lookup=functools.partial(history.recorded, path),
             policy=Policy.SCAN,
         )
         self._note_review(media_file, decision, ambiguous, untraced, records)
@@ -1731,8 +1729,15 @@ class AdminRoutesMixin:
                     return linked
             return decision.unlinked_status
         if decision.kind == PAIR:
-            result = ingest_resolved_pair(
-                self.db, path, decision.canonical, decision.resolution, marks=SCAN_MARKS
+            # A rerun after a library move re-checks the link rows.
+            result = self.db.survive_library_switch(
+                lambda: ingest_resolved_pair(
+                    self.db,
+                    path,
+                    decision.canonical,
+                    decision.resolution,
+                    marks=SCAN_MARKS,
+                )
             )
             if not detailed:
                 return result["status"]
@@ -1765,31 +1770,25 @@ class AdminRoutesMixin:
         ``{"status", "title"}``), or None when history names no single stored
         prompt for the file (``database.run_attribution``). Never creates a
         prompt. Works without any prompt metadata in the file. ``history`` is
-        the batch's precomputed lookup; without it the file is looked up alone.
+        the batch's lookup (``_BatchHistory``). A file already linked to
+        another prompt is a ``conflict``: one prompt per file.
         """
-        if history is not None:
-            prompt_id = history.get(str(media_file))
-        else:
-            try:
-                prompt_id = stored_run_prompt(
-                    self.db, str(media_file), host_output_directory()
-                )
-            except Exception:
-                self.logger.warning(
-                    f"History attribution failed for {media_file.name}",
-                    exc_info=True,
-                )
-                return None
+        if history is None:
+            history = _BatchHistory(self, [media_file])
+        prompt_id = history.get(str(media_file))
         if prompt_id is None:
             return None
         try:
-            image_id = self.db.link_image_to_prompt(
-                prompt_id, str(media_file), metadata=canonical
+            status = self.db.survive_library_switch(
+                lambda: link_stored_prompt_or_raise(
+                    self.db, str(media_file), canonical, prompt_id
+                )
             )
         except Exception as e:
             self.logger.error(f"Failed to link {media_file.name} from history: {e}")
             return "found"
-        status = "linked" if image_id else "unchanged"
+        if status is None:
+            return None
         if not detailed or status not in SCAN_TITLED_DETAILS:
             return status
         prompt = self.db.get_prompt_by_id(prompt_id) or {}

tests/test_pair_scan.py

@@ -1,7 +1,9 @@
 """Metadata scanning enriches known pairs and conservatively leaves conflicts."""
 
 import json
+import sqlite3
 from pathlib import Path
+from unittest.mock import patch
 
 from tests.test_api_admin import AdminAPITestCase
 from utils.hashing import generate_prompt_hash
@@ -549,3 +551,42 @@ class TestScanDecision(AdminAPITestCase):
         self.assertEqual(status, "added")
         self.assertEqual((ambiguous, untraced), ([path], []))
         self.assertEqual(records["canonical"]["prompt"], g)
+
+    def test_an_ambiguous_file_with_a_junk_answer_is_still_queued(self):
+        from tests.test_file_prompt import _numeric_first_branches
+
+        path = self._path("numbers.png")
+        meta = {"prompt": json.dumps(_numeric_first_branches())}
+        with patch.object(
+            type(self.api), "_record_scanned_review", autospec=True
+        ) as record:
+            counts = self.api._ingest_scan_batch_sync([(path, meta)])
+        self.assertEqual(counts["ambiguous"], 1, counts)
+        record.assert_called_once()
+        self.assertEqual(self._links(path), [])
+
+    def test_a_conflict_row_says_the_link_was_kept(self):
+        from py.api.admin import _scan_outcome
+
+        _group, row, _line = _scan_outcome(self._path("c.png"), [], "conflict")
+        self.assertEqual(row["detail"], "Kept: already linked to a different prompt")
+
+    def test_a_library_move_reruns_the_scan_writes(self):
+        from py.api import admin
+
+        real = admin.ingest_resolved_pair
+        model = self.api.db.model
+
+        def moved_once(*args, **kwargs):
+            if not calls:
+                calls.append(1)
+                model._generation += 1
+                raise sqlite3.OperationalError("closed")
+            return real(*args, **kwargs)
+
+        calls = []
+        path = self._path("pair.png")
+        with patch.object(admin, "ingest_resolved_pair", side_effect=moved_once):
+            meta = {"prompt": json.dumps(sampled_graph("cat", "blur"))}
+            self.assertEqual(self.api._ingest_scanned_file(path, meta), "added")
+        self.assertEqual(calls, [1])

tests/test_run_attribution.py

@@ -492,6 +492,47 @@ class TestScanUsesHistory(RecordedRuns, AdminAPITestCase):
             db.get_prompt_by_hash(generate_prompt_hash(shapes.BRANCH_TEXTS[1]))
         )
 
+    def test_a_recorded_run_never_adds_a_second_link(self):
+        db = self.api.db
+        db.save_prompt(text=LLM_TEXT, prompt_hash=generate_prompt_hash(LLM_TEXT))
+        other = db.save_prompt(text="a dog", prompt_hash=generate_prompt_hash("a dog"))
+        path = self.write("bare_00003_.png")
+        db.link_image_to_prompt(other, str(path))
+        self.record_run(db, str(self.output_dir), [("", path.name)], [LLM_TEXT])
+        self.assertEqual(self.api._ingest_scanned_file(path, {}), "conflict")
+        rows = db.model.get_connection().execute(
+            "SELECT prompt_id FROM generated_images WHERE image_path=?", (str(path),)
+        )
+        self.assertEqual([row[0] for row in rows], [other])
+
+    def test_a_single_file_asks_history_once(self):
+        db = self.api.db
+        db.save_prompt(text=LLM_TEXT, prompt_hash=generate_prompt_hash(LLM_TEXT))
+        path = self.write("llm_00002_.png")
+        self.record_run(db, str(self.output_dir), [("", path.name)], [LLM_TEXT])
+        meta = {"prompt": json.dumps(llm_text_graph())}
+        statements = []
+        conn = db.model.get_connection()
+        conn.set_trace_callback(statements.append)
+        try:
+            self.assertEqual(self.api._ingest_scanned_file(path, meta), "linked")
+        finally:
+            conn.set_trace_callback(None)
+        lookups = [s for s in statements if "FROM execution_run_outputs" in s]
+        self.assertEqual(len(lookups), 1)
+
+    def test_a_file_the_recorded_saver_links_is_not_queued(self):
+        """A default-ambiguous file linked at the inspector's step is no
+        Needs review case."""
+        with mock.patch.object(
+            type(self.api), "_record_scanned_review", autospec=True
+        ) as record:
+            _paths, counts = self.scan_dataset(
+                ["IMG_GEN_00003_.png", "IMG_GEN_00004_.png"], ["13", "23"]
+            )
+        self.assertEqual(counts["added"], 2, counts)
+        record.assert_not_called()
+
 
 class TestHostHelpers(unittest.TestCase):
     """The host lookups degrade to "no proof", never to an error."""
@@ -893,6 +934,26 @@ class TestScanBatchLookup(RecordedRuns, AdminAPITestCase):
             self.assertIsNone(history.recorded(str(path)))
         self.assertIsNone(history.get(str(path)))
 
+    def test_a_failed_prompt_lookup_is_logged_and_names_nothing(self):
+        from py.api.admin import _BatchHistory
+
+        path = self.output_dir / "one.png"
+        path.write_bytes(b"png")
+        root = str(self.output_dir)
+        self.record_run(self.api.db, root, [("", path.name)], [LLM_TEXT])
+        history = _BatchHistory(self.api, [path])
+        with (
+            mock.patch(
+                "database.run_attribution.host_output_directory", return_value=root
+            ),
+            mock.patch(
+                "py.api.admin.named_prompt", side_effect=sqlite3.OperationalError("x")
+            ),
+            self.assertLogs("prompt_manager", "WARNING"),
+        ):
+            self.assertIsNone(history.get(str(path)))
+            self.assertEqual(history.recorded(str(path)).saver_id, "9")
+
 
 class TestLinkedOutputRoot(RecordedRuns, unittest.TestCase):
     """R2: ComfyUI's output folder reached through a symlink or junction."""

tests/test_scanned_prompts.py

@@ -166,6 +166,27 @@ class TestStoreScannedPrompt(StoreCase):
         self.assertEqual(len(self.prompts()), 1)
         self.assertEqual(len(self.links()), 1)
 
+    def test_a_library_move_reruns_the_store(self):
+        from database import scanned_prompts
+
+        real = scanned_prompts.store_scanned_prompt_or_raise
+        calls = []
+
+        def moved_once(*args, **kwargs):
+            if not calls:
+                calls.append(1)
+                self.db.model._generation += 1
+                raise sqlite3.OperationalError("closed")
+            return real(*args, **kwargs)
+
+        with mock.patch.object(
+            scanned_prompts, "store_scanned_prompt_or_raise", side_effect=moved_once
+        ):
+            status, _id = store_scanned_prompt(
+                self.db, self.path, META, "a cat", SCAN_MARKS
+            )
+        self.assertEqual((status, calls), ("added", [1]))
+
 
 class TestPairMarks(StoreCase):
     def test_marks_reach_only_created_components(self):