Merge branch 'feat/247-foreign-workflows' into feat/247-foreign-workflows-pr2

TESTING.md

@@ -5066,12 +5066,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.
@@ -5080,14 +5082,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_file_prompt.py

@@ -82,6 +82,43 @@ def _gated(graph):
     return graph
 
 
+def _prompt_manager_branches(prepend="PRE", append="APP"):
+    """Two KSampler branches, each fed by a PromptManagerText string."""
+    graph = sampled_graph()
+    del graph["1"]
+    sampler = graph.pop("3")
+    for index, text in enumerate(("a cat", "a dog"), start=1):
+        base = index * 10
+        graph[str(base)] = {
+            "class_type": "PromptManagerText",
+            "inputs": {"text": text, "prepend_text": prepend, "append_text": append},
+        }
+        graph[str(base + 1)] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": [str(base), 0]},
+        }
+        graph[str(base + 2)] = {
+            **sampler,
+            "inputs": {**sampler["inputs"], "positive": [str(base + 1), 0]},
+        }
+    return graph
+
+
+def _numeric_first_branches():
+    """Three branches; the legacy parser's answer (the first) is a number."""
+    graph = sampled_graph("2048")
+    for index, text in ((5, "a cat"), (6, "a dog")):
+        graph[str(index * 10)] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": text},
+        }
+        graph[str(index * 10 + 1)] = {
+            **graph["3"],
+            "inputs": {**graph["3"]["inputs"], "positive": [str(index * 10), 0]},
+        }
+    return graph
+
+
 class TestStepsBeforeResolving(unittest.TestCase):
     def test_step_1_no_metadata_tries_the_recorded_run(self):
         for metadata in ({}, None):
@@ -114,6 +151,19 @@ class TestStepsBeforeResolving(unittest.TestCase):
         )
         self.assertNotIn("file_info", _decide(_graph(sampled_graph())).canonical)
 
+    def test_a_file_size_reader_is_called_only_for_prompt_metadata(self):
+        reads = []
+
+        def size():
+            reads.append(1)
+            return 9
+
+        for metadata in ({}, {"workflow": "{}"}):
+            _decide(metadata, file_size=size)
+        self.assertEqual(reads, [])
+        decision = _decide(_graph(sampled_graph()), file_size=size)
+        self.assertEqual((reads, decision.canonical["file_info"]), ([1], {"size": 9}))
+
 
 class TestResolvedSteps(unittest.TestCase):
     def test_step_6_a_blank_positive_counts_as_unresolved(self):
@@ -184,8 +234,10 @@ class TestUndecidedGraph(unittest.TestCase):
         self.assertEqual(decision.source, "legacy-parser")
 
     def test_step_8_1_the_file_name_anchor_alone_never_decides(self):
-        """#390 ruling: a file only its saver's file name resolves is queued
-        for Needs review by the scan, never linked; the monitor leaves it."""
+        """#390: 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; the monitor
+        stores nothing."""
         name = "s45_branch_2_00001_.png"
         scan = _decide(_graph(shapes.branch_bench()), lookup=_Lookup(), file_name=name)
         self.assertEqual((scan.kind, scan.source), ("positive", "legacy-parser"))
@@ -231,8 +283,29 @@ class TestUndecidedGraph(unittest.TestCase):
         )
         self.assertEqual(decision.kind, "never")
         self.assertEqual(decision.unlinked_status, "ambiguous")
+        self.assertFalse(decision.try_recorded_run)
         self.assertEqual(lookup.calls, 1)
 
+    def test_step_8_4_joined_prompt_manager_text_is_never_the_legacy_answer(self):
+        """The legacy parser returns prepend + text + append; the stored text
+        of a PromptManager node cannot be told from it."""
+        decision = _decide(_graph(_prompt_manager_branches()), lookup=_Lookup())
+        self.assertEqual(decision.kind, "never")
+        self.assertEqual(decision.unlinked_status, "ambiguous")
+        self.assertEqual(decision.review, "ambiguous")
+        self.assertTrue(decision.try_recorded_run)
+        # A linked part the parser does not follow: its answer is the node's
+        # own text, which is what the node stores (#170 parity).
+        linked = _prompt_manager_branches(prepend=["99", 0], append="")
+        decision = _decide(_graph(linked), lookup=_Lookup())
+        self.assertEqual((decision.kind, decision.text), ("positive", "a cat"))
+
+    def test_step_8_4_prompt_manager_text_without_parts_keeps_the_floor(self):
+        """Regression guard (passes on the base): no prepend or append."""
+        graph = _prompt_manager_branches(prepend="", append=" ")
+        decision = _decide(_graph(graph), lookup=_Lookup())
+        self.assertEqual((decision.kind, decision.text), ("positive", "a cat"))
+
     def test_step_8_4_an_ambiguous_file_gets_the_legacy_answer_in_the_scan(self):
         decision = _decide(_graph(shapes.shared_prefix()), lookup=_Lookup())
         self.assertEqual(decision.kind, "positive")
@@ -358,6 +431,7 @@ class TestJunkGate(unittest.TestCase):
             "nan",
             "a portrait saved as final.safetensors",
             "4k",
+            "C:\\My Models\\x.safetensors",
         ):
             with self.subTest(text=text):
                 self.assertIsNone(junk_text(text))
@@ -381,6 +455,16 @@ class TestJunkGate(unittest.TestCase):
             decision = _decide(_graph(_gated(sampled_graph(LORA))), lookup=_Lookup())
         self.assertEqual(decision.source, "junk:lora-only")
         self.assertEqual(decision.kind, "never")
+        self.assertEqual(decision.unlinked_status, "unresolved")
+
+    def test_step_13_junk_at_step_8_keeps_that_steps_status_and_review(self):
+        # An ambiguous file whose legacy answer is a number is still a
+        # Needs review case: its status stays "ambiguous".
+        decision = _decide(_graph(_numeric_first_branches()), lookup=_Lookup())
+        self.assertEqual((decision.kind, decision.source), ("never", "junk:number"))
+        self.assertEqual(decision.unlinked_status, "ambiguous")
+        self.assertEqual(decision.review, "ambiguous")
+        self.assertTrue(decision.try_recorded_run)
 
 
 class TestParseComfyuiPrompt(unittest.TestCase):

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

@@ -493,6 +493,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."""
@@ -1179,6 +1220,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):

utils/file_prompt.py

@@ -20,7 +20,7 @@ import json
 import re
 from dataclasses import dataclass, field
 from enum import Enum
-from typing import Any, Callable, Mapping, Optional
+from typing import Any, Callable, Mapping, Optional, Union
 
 from .generation_params import (
     PLACEHOLDERS,
@@ -95,7 +95,7 @@ def decide_file_prompt(
     metadata: Any,
     *,
     file_name: Optional[str],
-    file_size: Optional[int],
+    file_size: Union[None, int, Callable[[], Optional[int]]],
     in_output: bool,
     lookup: Optional[Callable[[], Any]],
     policy: Policy,
@@ -106,6 +106,8 @@ def decide_file_prompt(
     (the monitor). ``lookup`` returns the file's ``RecordedOutput`` or None;
     it is None when history must not be asked yet. It is called at most
     once, and only for a graph the default profile leaves undecided.
+    ``file_size`` may be a callable: it is read only for a file with prompt
+    metadata whose ``file_info`` has no size.
     """
     if not metadata:
         return Decision(NEVER, try_recorded_run=True)
@@ -127,7 +129,7 @@ def decide_file_prompt(
         lookup,
         policy,
     )
-    return _gated(_decide(found))
+    return _gated(_decide(found), found)
 
 
 def _decide(found: _File) -> Decision:
@@ -165,8 +167,12 @@ def _undecided_graph(found: _File) -> Decision:
         if found.lookup is None and found.in_output:
             return _decision(found, HISTORY, None, "none")
         if not inspected.malformed and inspected.resolution["state"] == "ambiguous":
-            return _never(found, _unlinked(found), try_recorded_run=True)
+            return _never(found, _unlinked(found))
     text = graph_prompt_text(found.metadata)
+    if text is not None and _joined_text(found.canonical["prompt"], text):
+        # The legacy parser joins prepend + text + append: the text a
+        # PromptManager node stores cannot be told from it (#343).
+        return _never(found, _unlinked(found), try_recorded_run=True)
     if text is not None:
         return _decision(found, POSITIVE, text, "legacy-parser")
     return _never(found, _unlinked(found), try_recorded_run=True)
@@ -187,8 +193,55 @@ def _sampler_less(found: _File, graph: Any) -> Decision:
     return _never(found, status, try_recorded_run=True)
 
 
-def _gated(decision: Decision) -> Decision:
-    """Junk text is never stored; the file reports ``found`` (step 13)."""
+_UNDECIDED_SOURCES = ("inspector", "recorded-run", "legacy-parser")
+
+
+def _joined_text(graph: Any, text: str) -> bool:
+    """Whether ``text`` may hold a PromptManager node's prepend or append.
+
+    True when a PromptManager-family node has a prepend or append part and
+    ``text`` is not the literal ``text`` of one such node (the legacy
+    parser's answer when it does not follow a linked part).
+    """
+    own = _prompt_manager_parts(graph)
+    return own is not None and text.strip() not in own
+
+
+def _prompt_manager_parts(graph: Any) -> Optional[set]:
+    """The literal texts of PromptManager-family nodes with a prepend or
+    append part, or None when no such node exists."""
+    if not isinstance(graph, dict):
+        return None
+    found = None
+    for node in graph.values():
+        if not isinstance(node, dict):
+            continue
+        name, inputs = node.get("class_type"), node.get("inputs")
+        if not isinstance(name, str) or "promptmanager" not in name.lower():
+            continue
+        if not isinstance(inputs, dict) or not _has_part(inputs):
+            continue
+        found = found if found is not None else set()
+        if isinstance(inputs.get("text"), str):
+            found.add(inputs["text"].strip())
+    return found
+
+
+def _has_part(inputs: dict) -> bool:
+    for key in ("prepend_text", "append_text"):
+        part = inputs.get(key)
+        if part is not None and not (isinstance(part, str) and not part.strip()):
+            return True
+    return False
+
+
+def _gated(decision: Decision, found: _File) -> Decision:
+    """Junk text is never stored (step 13).
+
+    A junk answer of the undecided-graph step keeps that step's unlinked
+    status, so an ambiguous file is still a Needs review case; any other
+    step reports ``found``.
+    """
     if decision.kind not in (PAIR, POSITIVE):
         return decision
     junk = junk_text(decision.text)
@@ -201,7 +254,9 @@ def _gated(decision: Decision) -> Decision:
         source=f"junk:{junk}",
         review=decision.review,
         try_recorded_run=True,
-        unlinked_status="found",
+        unlinked_status=(
+            _unlinked(found) if decision.source in _UNDECIDED_SOURCES else "found"
+        ),
     )
 
 
@@ -263,7 +318,7 @@ def _parsed(metadata: Mapping) -> dict:
     return canonical
 
 
-def _with_size(canonical: dict, file_size: Optional[int]) -> None:
+def _with_size(canonical: dict, file_size: Any) -> None:
     """Record the file's size in ``file_info`` unless the metadata carries one.
 
     Stored as ``generated_images.file_size``; the "new" scan scope skips a
@@ -271,8 +326,10 @@ def _with_size(canonical: dict, file_size: Optional[int]) -> None:
     """
     info = canonical.get("file_info")
     info = dict(info) if isinstance(info, dict) else {}
-    if info.get("size") is None and file_size is not None:
-        info["size"] = file_size
+    if info.get("size") is None:
+        size = file_size() if callable(file_size) else file_size
+        if size is not None:
+            info["size"] = size
     if info:
         canonical["file_info"] = info