Merge remote-tracking branch 'origin/main' into feat/386-recorded-attribution

The history route inspects a case-normalized target on Windows, while
history matches file names exactly, so the occurrence's reference names
the file.

TESTING.md

@@ -889,6 +889,25 @@ unattributed stage, when it is a plain first stage: not a detailer, no prompt ov
 set, and no listed sampler in its own latent ancestry (else it stays unattributed) (`tests/test_resolver_saver_anchor.py`, shapes in
 `tests/saver_anchor_graphs.py` from the maintainer's `gpu_unlock_bench`, dataset and
 ANIMA runs; browser parity through the `anchor-*` cases, which carry `__file_name`).
+Recorded run history (#386): both inspect routes (Images and a history occurrence)
+ask `database.run_attribution.recorded_output` inside the file's stat window. It
+answers only for a file that exactly one recorded run reported and that was written
+during that run (the #342 rules): the one output node that run reported it from (none
+when several did) and, per sampler anchor, the effective texts PromptManager recorded
+at execution (a role only inferred from the graph is none). Only when the file alone
+leaves the prompt open (`ambiguous` or `unresolved`) the inspector resolves again
+anchored on that node (`recorded_saver_id`, walked like a file-name match; a file
+name naming a different saver contradicts history and anchors nothing), and keeps
+that result only when history's saver chose its samplers (`recorded_anchor`) and it
+names the positive (resolved, or a guider with no negative); its `graph-inferred`
+sides read `recorded-run` ("Recorded run" badge, ranked below every "Workflow
+inferred" source). Still open at the one sampler that saver leads to, with that
+sampler's positive text recorded (a language model's output feeding PromptManager),
+the recorded texts show as `execution`. Outputs and texts are read in one
+transaction. A result the file resolves alone never changes; a lookup failure is
+logged and judges the file alone; local Inspect and ingestion are unchanged
+(`tests/test_recorded_saver_anchor.py`, `tests/test_run_attribution.py`,
+`tests/test_api_history_outputs.py`).
 
 `tools/inspection_corpus.py --root DIR --out JSONL [--workers N] [--baseline JSONL]
 [--summary TXT] [--code TREE]` measures a media folder read-only against the #240

database/run_attribution.py

@@ -15,6 +15,10 @@ only when all of this holds:
 
 Anything else returns None: the caller leaves the file unlinked. Read-only.
 
+``recorded_output`` applies the first two rules and returns, for the
+inspector (#386), the one output node (save node) that run reported the file
+from and the texts PromptManager recorded at execution per sampler.
+
 Lookups are batched (``stored_run_prompts``): one parameterized query per
 chunk of at most ``LOOKUP_CHUNK`` distinct file names, matched in memory. The
 history schema is left exactly as it is (no extra index), so older builds can
@@ -26,6 +30,7 @@ import os
 import sqlite3
 import threading
 import weakref
+from dataclasses import dataclass
 from typing import Any, Callable, Dict, Iterable, List, Optional
 
 try:
@@ -45,7 +50,8 @@ LOOKUP_CHUNK = 500
 
 _RUNS_FOR_NAMES = (
     "SELECT DISTINCT o.filename, o.subfolder, o.root_key, o.path_style, o.run_id, "
-    "r.observed_dequeue_at, COALESCE(r.terminal_event_at, r.observed_callback_at) "
+    "r.observed_dequeue_at, COALESCE(r.terminal_event_at, r.observed_callback_at), "
+    "o.output_node_id "
     "FROM execution_run_outputs o JOIN execution_runs r ON r.run_id = o.run_id "
     "WHERE o.folder_type = 'output' AND o.filename IN ({marks})"
 )
@@ -138,9 +144,10 @@ def stored_run_prompt(
 
 
 def _runs_by_location(db: Any, names: List[str]) -> Optional[Dict[tuple, dict]]:
-    """``{(filename, subfolder, root_key, style): {run_id: (start, end)}}``.
+    """``{(filename, subfolder, root_key, style): {run_id: (start, end, nodes)}}``.
 
-    One query per ``LOOKUP_CHUNK`` names; None without history tables.
+    ``nodes`` is the set of output node ids that reported the location in that
+    run. One query per ``LOOKUP_CHUNK`` names; None without history tables.
     """
     found: Dict[tuple, dict] = {}
     conn = db.model.get_connection()
@@ -151,12 +158,49 @@ def _runs_by_location(db: Any, names: List[str]) -> Optional[Dict[tuple, dict]]:
             rows = conn.execute(query, chunk).fetchall()  # nosec B608
         except sqlite3.Error:
             return None  # no history tables (a library without history)
-        for name, subfolder, root_key, style, run_id, began, ended in rows:
+        for name, subfolder, root_key, style, run_id, began, ended, node in rows:
             runs = found.setdefault((name, subfolder, root_key, style), {})
-            runs[run_id] = (began, ended)
+            runs.setdefault(run_id, (began, ended, set()))[2].add(node)
     return found
 
 
+def _recorded_runs(
+    db: Any,
+    image_paths: Iterable[str],
+    output_root: Optional[str],
+    path_style: Optional[str],
+) -> Dict[str, tuple]:
+    """``{path: (run_id, output node ids)}`` for each file history attributes.
+
+    A file is attributed when exactly one recorded run reported its location
+    and it was last written while that run executed.
+    """
+    paths = list(dict.fromkeys(path for path in image_paths if path))
+    if not paths or not output_root:
+        return {}
+    style = path_style or ("windows" if os.name == "nt" else "posix")
+    located = {}
+    for path in paths:
+        location = _location(path, output_root, style)
+        if location is not None:
+            located[path] = location
+    if not located:
+        return {}
+    names = sorted({filename for _key, _sub, filename in located.values()})
+    runs = _runs_by_location(db, names)
+    if not runs:
+        return {}
+    attributed = {}
+    for path, (root_key, subfolder, filename) in located.items():
+        reported = runs.get((filename, subfolder, root_key, style), {})
+        if len(reported) != 1:
+            continue
+        ((run_id, (began, ended, nodes)),) = reported.items()
+        if _written_during(path, began, ended):
+            attributed[path] = (run_id, nodes)
+    return attributed
+
+
 def _named_prompt(db: Any, run_id: str) -> Optional[int]:
     contexts = db.get_execution_contexts(run_id).get("contexts") or []
     named = {
@@ -183,35 +227,115 @@ def stored_run_prompts(
     """
     paths = list(dict.fromkeys(path for path in image_paths if path))
     result: Dict[str, Optional[int]] = dict.fromkeys(paths)
-    if not paths or not output_root:
-        return result
-    style = path_style or ("windows" if os.name == "nt" else "posix")
-    located = {}
-    for path in paths:
-        location = _location(path, output_root, style)
-        if location is not None:
-            located[path] = location
-    if not located:
-        return result
-    names = sorted({filename for _key, _sub, filename in located.values()})
-    runs = _runs_by_location(db, names)
-    if not runs:
-        return result
     prompts: Dict[str, Optional[int]] = {}
-    for path, (root_key, subfolder, filename) in located.items():
-        reported = runs.get((filename, subfolder, root_key, style), {})
-        if len(reported) != 1:
-            continue
-        ((run_id, (began, ended)),) = reported.items()
-        if not _written_during(path, began, ended):
-            continue
+    for path, (run_id, _nodes) in _recorded_runs(
+        db, paths, output_root, path_style
+    ).items():
         if run_id not in prompts:
             prompts[run_id] = _named_prompt(db, run_id)
         result[path] = prompts[run_id]
     return result
 
 
+@dataclass(frozen=True)
+class RecordedOutput:
+    """What recorded history knows about the run that wrote one file (#386).
+
+    ``saver_id`` is the one output node the run reported the file from (None
+    when several did). ``texts`` maps each sampler anchor of the run to the
+    effective texts PromptManager recorded at execution for its roles
+    (``{"positive": text or None, "negative": text or None}``); a role whose
+    text was only inferred from the graph is None.
+    """
+
+    saver_id: Optional[str]
+    texts: Dict[str, Dict[str, Optional[str]]]
+
+
+def _recorded_texts(db: Any, run_id: str) -> Dict[str, Dict[str, Optional[str]]]:
+    texts = {}
+    for context in db.get_execution_contexts(run_id).get("contexts") or []:
+        evidence = context.get("evidence") or {}
+        roles = {}
+        for role in ("positive", "negative"):
+            values = context.get(role) or {}
+            proof = (evidence.get(role) or {}).get("effective") or {}
+            value = values.get("effective")
+            recorded = proof.get("provenance") == "recorded"
+            roles[role] = value if recorded and isinstance(value, str) else None
+        texts[context["anchor_id"]] = roles
+    return texts
+
+
+def recorded_output(
+    db: Any,
+    image_path: str,
+    output_root: Optional[str],
+    path_style: Optional[str] = None,
+) -> Optional[RecordedOutput]:
+    """The recorded run facts for ``image_path``, else None (#386).
+
+    The file must be attributed to one run (``_recorded_runs``: one run
+    reported its location and it was written during that run). Read-only; it
+    never links anything.
+    """
+    if not image_path:
+        return None
+    conn = db.model.get_connection()
+    own_transaction = not conn.in_transaction
+    if own_transaction:  # outputs and texts from one state of history
+        conn.execute("BEGIN")
+    try:
+        found = _recorded_runs(db, [image_path], output_root, path_style)
+        if image_path not in found:
+            return None
+        run_id, nodes = found[image_path]
+        saver = next(iter(nodes)) if len(nodes) == 1 else None
+        return RecordedOutput(saver, _recorded_texts(db, run_id))
+    finally:
+        if own_transaction:
+            conn.rollback()
+
+
 _LOGGER = logging.getLogger("prompt_manager.run_attribution")
+
+
+def recorded_output_lookup(
+    db: Any, *, close: bool = False, reference: Optional[dict] = None
+) -> Callable[[Any], Any]:
+    """``lookup(path) -> RecordedOutput or None`` for the inspect routes (#386).
+
+    Uses ComfyUI's output folder. Any failure (no history, a locked library)
+    is logged and names no node, so the file is judged from its own metadata.
+    ``close`` closes the calling thread's connection afterwards, for routes
+    that close theirs after reading. ``reference`` (a history occurrence's
+    output reference) spells the file as history recorded it: the route's
+    target is case-normalized on Windows, and history matches names exactly.
+    """
+
+    def lookup(path: Any) -> Optional[RecordedOutput]:
+        root = host_output_directory()
+        if reference is not None:
+            if reference.get("folder_type") != "output" or not root:
+                return None
+            path = os.path.join(
+                root, reference.get("subfolder") or "", reference.get("filename") or ""
+            )
+        try:
+            return recorded_output(db, str(path), root)
+        except (sqlite3.Error, OSError, ValueError):
+            _LOGGER.warning("Recorded run lookup failed", exc_info=True)
+            return None
+        except Exception:
+            _LOGGER.exception("Recorded run lookup failed unexpectedly")
+            return None
+        finally:
+            if close:
+                db.model.close()
+
+    return lookup
+
+
 _SEALED_LISTENERS: List[weakref.WeakMethod] = []
 _SEALED_LOCK = threading.Lock()
 

py/api/history.py

@@ -31,6 +31,7 @@ from .inspection_service import CurrentFileChanged, inspect_current_file
 try:
     from ...database.execution_history import HistoryEpochChanged
     from ...database.history_gallery import read_history_gallery
+    from ...database.run_attribution import recorded_output_lookup
     from ...database.history_reads import (
         HistoryCaptureChanged,
         OutputPosition,
@@ -42,6 +43,7 @@ try:
 except ImportError:
     from database.execution_history import HistoryEpochChanged
     from database.history_gallery import read_history_gallery
+    from database.run_attribution import recorded_output_lookup
     from database.history_reads import (
         HistoryCaptureChanged,
         OutputPosition,
@@ -728,6 +730,9 @@ class HistoryRoutesMixin:
                             "library",
                             {"status": "unavailable" if saved_unavailable else "none"},
                         ),
+                        recorded=recorded_output_lookup(
+                            db, close=True, reference=receipt["reference"]
+                        ),
                     )
                     _reauthorize_current_target(resolved, scope)
                     identity = {

py/api/images.py

@@ -61,6 +61,7 @@ try:
     from ...database.image_favorites import MAX_KEY_LENGTH
     from ...database.library import display_title
     from ...database.models import normalize_image_path
+    from ...database.run_attribution import recorded_output_lookup
     from ...utils.card_seed import card_seed, header_params, parsed_positive
     from ...utils.parallel import map_parallel
     from ...utils.json_strict import dumps_strict
@@ -81,6 +82,7 @@ except ImportError:
     from database.image_favorites import MAX_KEY_LENGTH
     from database.library import display_title
     from database.models import normalize_image_path
+    from database.run_attribution import recorded_output_lookup
     from utils.card_seed import card_seed, header_params, parsed_positive
     from utils.parallel import map_parallel
     from utils.json_strict import dumps_strict
@@ -3690,6 +3692,7 @@ class ImageRoutesMixin:
                         image_tags=self._inspection_image_tags(
                             roots[index], Path(relative).parts
                         ),
+                        recorded=recorded_output_lookup(self.db),
                     )
                 except CurrentFileChanged:
                     return None

py/api/inspection_service.py

@@ -3,7 +3,7 @@
 import os
 import stat
 from pathlib import Path
-from typing import Callable
+from typing import Any, Callable
 
 try:
     from ...utils.prompt_inspector import (
@@ -100,18 +100,23 @@ def inspect_current_file(
     saved: dict | None,
     library: dict,
     image_tags: Callable[[int, int], dict] | None = None,
+    recorded: Callable[[Path], Any] | None = None,
 ) -> dict:
     """Current-file evidence read between two equal stats of *path*.
 
     *image_tags*, when given, is called once with that stable
     ``(st_size, st_mtime_ns)`` and its facet is returned as ``image_tags``
     (spec amendment S1: tags of a byte record verified at exactly that stat).
+    *recorded*, when given, is asked within the same stat window for what
+    recorded run history knows about *path* (``RecordedOutput`` or None: the
+    save node that wrote it and the texts PromptManager recorded, #386).
     """
     before = _current_file_facts(path)
     try:
         inspection = read_inspection_file(path)
     except MetadataReadError:
         inspection = None
+    run = recorded(path) if recorded is not None and inspection is not None else None
     after = _current_file_facts(path)
     fields = ("st_dev", "st_ino", "st_mode", "st_size", "st_mtime_ns", "st_ctime_ns")
     if any(getattr(before, key) != getattr(after, key) for key in fields):
@@ -129,9 +134,15 @@ def inspect_current_file(
         profile = (
             PREVIEW_INSPECTION_PROFILE if _in_host_temp(path) else INSPECTION_PROFILE
         )
-        # The file name anchors on the saver that wrote it (#381).
+        # The file name anchors on the saver that wrote it (#381); recorded run
+        # history names that saver when the file alone cannot (#386).
         evidence = inspect_prompt_evidence(
-            inspection.metadata, role_saved, profile=profile, file_name=path.name
+            inspection.metadata,
+            role_saved,
+            profile=profile,
+            file_name=path.name,
+            recorded_saver_id=run.saver_id if run is not None else None,
+            recorded_texts=run.texts if run is not None else None,
         )
     return {
         "file": {

tests/js/workspace-file-inspector.test.js

@@ -185,6 +185,26 @@ test('legacy fill applies to local records', async () => {
     assert.notEqual(zeroed.prompts.positive.text, '');
 });
 
+test('a branch named by recorded run history shows the Recorded run badge', async () => {
+    // #386: the server labels sides whose save node recorded history named; Execution recorded stays stronger.
+    const base = await localRecord(new File([fs.readFileSync(new URL('f09.png', PARITY))], 'f09.png'));
+    const recorded = {...base, prompts:{positive:{...base.prompts.positive, source:'recorded-run'}, negative:{...base.prompts.negative, source:'recorded-run'}}};
+    let view = renderFull(recorded);
+    try {
+        assert.deepEqual(view.texts(n => n.className === 'chip blue'), ['Recorded run']);
+        assert.ok(view.texts(n => n.tagName === 'p').includes('Prompt evidence: recorded run'));
+    } finally { view.done(); }
+    view = renderFull({...recorded, prompts:{...recorded.prompts, negative:{...recorded.prompts.negative, source:'execution'}}});
+    try {
+        assert.deepEqual(view.texts(n => n.className === 'chip blue'), ['Execution recorded']);
+    } finally { view.done(); }
+    // A recorded branch whose negative left out a run-time part is still inferred from the workflow.
+    view = renderFull({...recorded, prompts:{...recorded.prompts, negative:{...recorded.prompts.negative, source:'runtime-parts-omitted'}}});
+    try {
+        assert.deepEqual(view.texts(n => n.className === 'chip blue'), ['Workflow inferred']);
+    } finally { view.done(); }
+});
+
 // #249 PR C Task C3 (D88): one chip row, the file's current image tags first, then D21's linked tags not already shown.
 function tagged(imageTags, library = {status:'linked', tags:['Minimal', 'warm_light', 'editorial'], tags_scope:'positive'}) {
     return {...record(), imageTags, library};

tests/test_api_history_outputs.py

@@ -21,6 +21,8 @@ from tests.test_api_images import ImageAPITestCase, make_png
 from tests.test_api_prompt_inspector import graph, png
 from tests.test_execution_context_database import bundle as context_bundle
 from tests.test_execution_history_database import JOB, SESSION
+from tests.saver_anchor_graphs import BRANCH_TEXTS, shared_prefix
+from tests.test_recorded_saver_anchor import LLM_TEXT, text_generate_run
 from utils.execution_contexts import ContextBundle
 from utils.execution_history_events import make_finish, make_start
 from utils.execution_outputs import OutputBundle, OutputReference, snapshot_output_roots
@@ -1505,3 +1507,88 @@ class HistoryOutputAPITests(ImageAPITestCase):
         body = await response.json()
         self.assertEqual(response.status, 200, body)
         self.assertEqual(body["prompts"]["positive"]["text"], "Readable portrait")
+
+    async def recorded_shared_prefix_run(self):
+        """#386: the dataset shape, branch 3's saver wrote the file in a run."""
+        path = png(self.output_dir / "IMG_GEN_00371_.png", prompt=shared_prefix())
+        # make_start/make_finish observe the run at 0 ms: the file was written then.
+        os.utime(path, ns=(1_000_000, 1_000_000))
+        start = self.capture(
+            [self.ref(node="33", filename=path.name), self.ref("43", 1, "other.png")]
+        )
+        return start
+
+    async def test_both_inspect_routes_name_the_branch_recorded_history_saved(self):
+        start = await self.recorded_shared_prefix_run()
+        _, body = await self.outputs(start)
+        descriptor = next(
+            item["current_candidate"]["media"]
+            for item in body["items"]
+            if item["occurrence"]["output_node_id"] == "33"
+        )
+        response = await self.current(descriptor["inspection_url"])
+        recorded = await response.json()
+        self.assertEqual(response.status, 200, recorded)
+        response = await self.client.get(
+            "/prompt_manager/images/inspect/IMG_GEN_00371_.png", params={"root": "0"}
+        )
+        images = await response.json()
+        self.assertEqual(response.status, 200, images)
+        for result in (recorded, images):
+            self.assertEqual(result["resolution"]["state"], "resolved")
+            self.assertEqual(
+                result["prompts"]["positive"],
+                {
+                    "status": "resolved",
+                    "text": BRANCH_TEXTS[2],
+                    "source": "recorded-run",
+                },
+            )
+
+    async def test_a_file_rewritten_after_its_run_keeps_several_prompts(self):
+        await self.recorded_shared_prefix_run()
+        os.utime(self.output_dir / "IMG_GEN_00371_.png")  # written now: not that run
+        response = await self.client.get(
+            "/prompt_manager/images/inspect/IMG_GEN_00371_.png", params={"root": "0"}
+        )
+        body = await response.json()
+        self.assertEqual(response.status, 200, body)
+        self.assertEqual(body["resolution"]["state"], "ambiguous")
+        self.assertEqual(body["prompts"]["positive"]["status"], "ambiguous")
+
+    async def test_a_failing_history_lookup_judges_the_file_alone(self):
+        await self.recorded_shared_prefix_run()
+        with (
+            patch(
+                "database.run_attribution.recorded_output",
+                side_effect=sqlite3.OperationalError("locked"),
+            ),
+            self.assertLogs("prompt_manager.run_attribution", "WARNING"),
+        ):
+            response = await self.client.get(
+                "/prompt_manager/images/inspect/IMG_GEN_00371_.png",
+                params={"root": "0"},
+            )
+            body = await response.json()
+        self.assertEqual(response.status, 200, body)
+        self.assertEqual(body["resolution"]["state"], "ambiguous")
+
+    async def test_a_language_model_text_shows_what_promptmanager_recorded(self):
+        path = png(self.output_dir / "moss_00026_.png", prompt=text_generate_run())
+        os.utime(path, ns=(1_000_000, 1_000_000))
+        start = self.capture([self.ref(node="9", filename=path.name)])
+        self.api.db.write_execution_contexts(
+            start.run_id,
+            start.storage_epoch,
+            context_bundle("", LLM_TEXT, anchor="7"),
+        )
+        response = await self.client.get(
+            "/prompt_manager/images/inspect/moss_00026_.png", params={"root": "0"}
+        )
+        body = await response.json()
+        self.assertEqual(response.status, 200, body)
+        self.assertEqual(
+            body["prompts"]["positive"],
+            {"status": "resolved", "text": LLM_TEXT, "source": "execution"},
+        )
+        self.assertEqual(body["prompts"]["negative"]["text"], "")

tests/test_recorded_saver_anchor.py

@@ -0,0 +1,379 @@
+"""Recorded run history names the saver that wrote a file (#386, #240).
+
+When a file's graph alone cannot name its prompt (sampler branches that differ
+only in their prompt and share one saver prefix), the inspector asks recorded
+history which save node wrote the file and walks from that node, as the #381
+file-name anchor does. A result the file alone resolves is never changed; a
+recorded saver that names nothing certain leaves the file "several possible
+prompts". The resolved sides are labelled as coming from the recorded run.
+"""
+
+import json
+import unittest
+
+from utils import prompt_pair_resolver as resolver
+from utils.prompt_inspector import inspect_prompt_evidence
+
+try:
+    from tests import saver_anchor_graphs as shapes
+except ImportError:  # discovered with ``-s tests``
+    import saver_anchor_graphs as shapes
+
+
+def _resolve(graph, **options):
+    return resolver.resolve_prompt_pair(
+        {"prompt": graph}, profile=resolver.INSPECTION_PROFILE, **options
+    )
+
+
+def _positive(result):
+    side = result["positive"]
+    return side["text"] if isinstance(side, dict) else None
+
+
+def _evidence(graph, **options):
+    return inspect_prompt_evidence({"prompt": json.dumps(graph)}, **options)
+
+
+# ``branch_bench``'s savers are node ``N3`` of branch N (10s based).
+def _saver(index):
+    return str(index * 10 + 3)
+
+
+def guider_dataset():
+    """The dataset run's shape: SamplerCustomAdvanced branches on BasicGuider
+    (no negative), every branch saved with one prefix."""
+    graph = {
+        "1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "m"}},
+        "2": {"class_type": "EmptyLatentImage", "inputs": {"width": 64}},
+        "4": {"class_type": "RandomNoise", "inputs": {"noise_seed": 1}},
+        "5": {"class_type": "KSamplerSelect", "inputs": {"sampler_name": "euler"}},
+        "6": {
+            "class_type": "BasicScheduler",
+            "inputs": {"model": ["1", 0], "steps": 8, "scheduler": "simple"},
+        },
+    }
+    for index, text in enumerate(shapes.BRANCH_TEXTS, start=1):
+        base = index * 10
+        graph[str(base)] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": text, "clip": ["1", 1]},
+        }
+        graph[str(base + 4)] = {
+            "class_type": "BasicGuider",
+            "inputs": {"model": ["1", 0], "conditioning": [str(base), 0]},
+        }
+        graph[str(base + 1)] = {
+            "class_type": "SamplerCustomAdvanced",
+            "inputs": {
+                "noise": ["4", 0],
+                "guider": [str(base + 4), 0],
+                "sampler": ["5", 0],
+                "sigmas": ["6", 0],
+                "latent_image": ["2", 0],
+            },
+        }
+        graph[str(base + 2)] = {
+            "class_type": "VAEDecode",
+            "inputs": {"samples": [str(base + 1), 0], "vae": ["1", 2]},
+        }
+        graph[str(base + 3)] = {
+            "class_type": "SaveImage",
+            "inputs": {
+                "filename_prefix": "dataset/IMG_GEN",
+                "images": [str(base + 2), 0],
+            },
+        }
+    return graph
+
+
+class TestRecordedSaverInResolver(unittest.TestCase):
+    def test_the_recorded_saver_picks_its_branch_of_a_shared_prefix(self):
+        for index, text in enumerate(shapes.BRANCH_TEXTS, start=1):
+            with self.subTest(branch=index):
+                result = _resolve(
+                    shapes.shared_prefix(),
+                    file_name="IMG_GEN_00371_.png",
+                    recorded_saver_id=_saver(index),
+                )
+                self.assertEqual(result["state"], "resolved")
+                self.assertEqual(_positive(result), text)
+                self.assertEqual(result["sampler_id"], str(index * 10 + 1))
+
+    def test_without_a_recorded_saver_the_shared_prefix_stays_several(self):
+        result = _resolve(shapes.shared_prefix(), file_name="IMG_GEN_00371_.png")
+        self.assertEqual(result["state"], "ambiguous")
+
+    def test_a_recorded_saver_the_graph_lacks_anchors_nothing(self):
+        for node in ("missing", "3", "2"):
+            with self.subTest(node=node):
+                result = _resolve(shapes.shared_prefix(), recorded_saver_id=node)
+                self.assertEqual(result["state"], "ambiguous")
+
+    def test_a_recorded_saver_of_an_unknown_class_anchors(self):
+        graph = shapes.branch_bench(
+            prefixes=["dataset/IMG_GEN"] * len(shapes.BRANCH_TEXTS),
+            saver="VendorSaveImage",
+        )
+        result = _resolve(graph, recorded_saver_id=_saver(2))
+        self.assertEqual(_positive(result), shapes.BRANCH_TEXTS[1])
+
+    def test_the_default_profile_ignores_a_recorded_saver(self):
+        result = resolver.resolve_prompt_pair(
+            {"prompt": shapes.shared_prefix()}, recorded_saver_id=_saver(1)
+        )
+        self.assertEqual(result["state"], "ambiguous")
+
+
+class TestRecordedSaverInInspector(unittest.TestCase):
+    def test_the_inspector_shows_the_recorded_branch_labelled_as_recorded(self):
+        evidence = _evidence(
+            shapes.shared_prefix(),
+            file_name="IMG_GEN_00371_.png",
+            recorded_saver_id=_saver(3),
+        )
+        self.assertEqual(evidence["resolution"]["state"], "resolved")
+        positive = evidence["prompts"]["positive"]
+        self.assertEqual(positive["text"], shapes.BRANCH_TEXTS[2])
+        self.assertEqual(positive["source"], "recorded-run")
+        self.assertEqual(evidence["prompts"]["negative"]["source"], "recorded-run")
+
+    def test_a_guider_branch_without_a_negative_shows_its_positive(self):
+        self.assertEqual(
+            _evidence(guider_dataset(), file_name="IMG_GEN_00371_.png")["resolution"][
+                "state"
+            ],
+            "ambiguous",
+        )
+        evidence = _evidence(
+            guider_dataset(), file_name="IMG_GEN_00371_.png", recorded_saver_id="23"
+        )
+        self.assertEqual(
+            evidence["prompts"]["positive"],
+            {
+                "status": "resolved",
+                "text": shapes.BRANCH_TEXTS[1],
+                "source": "recorded-run",
+            },
+        )
+        self.assertEqual(evidence["prompts"]["negative"]["status"], "absent")
+
+    def test_a_file_name_naming_another_saver_contradicts_history(self):
+        # The bench file name names branch 2's saver; history says branch 4's.
+        graph = shapes.branch_bench()
+        for index in (1, 3, 4):
+            graph[_saver(index)]["inputs"]["filename_prefix"] = "bench/run/shared"
+        result = _resolve(
+            graph, file_name="s45_branch_2_00001_.png", recorded_saver_id=_saver(4)
+        )
+        self.assertIs(result["recorded_anchor"], False)
+        evidence = _evidence(
+            shapes.shared_prefix(),
+            file_name="IMG_GEN_00001_.png",
+            recorded_saver_id=_saver(4),
+        )
+        self.assertEqual(
+            evidence["prompts"]["positive"]["text"], shapes.BRANCH_TEXTS[3]
+        )
+
+    def test_a_file_that_resolves_alone_keeps_its_own_result(self):
+        # The file name names branch 2; a recorded saver never overrides it.
+        evidence = _evidence(
+            shapes.branch_bench(),
+            file_name="s45_branch_2_00001_.png",
+            recorded_saver_id=_saver(4),
+        )
+        positive = evidence["prompts"]["positive"]
+        self.assertEqual(positive["text"], shapes.BRANCH_TEXTS[1])
+        self.assertEqual(positive["source"], "graph-inferred")
+
+    def test_a_recorded_saver_that_names_nothing_keeps_several_prompts(self):
+        evidence = _evidence(shapes.shared_prefix(), recorded_saver_id="missing")
+        self.assertEqual(evidence["resolution"]["state"], "ambiguous")
+        self.assertEqual(evidence["prompts"]["positive"]["status"], "ambiguous")
+
+    def test_no_recorded_saver_is_unchanged(self):
+        self.assertEqual(
+            _evidence(shapes.shared_prefix(), recorded_saver_id=None),
+            _evidence(shapes.shared_prefix()),
+        )
+
+
+LLM_TEXT = "a red vintage car parked by the sea, golden hour"
+
+
+def text_generate_run():
+    """The maintainer's moss run: a language model writes PromptManager's text."""
+    return {
+        "1": {"class_type": "CheckpointLoaderSimple", "inputs": {"ckpt_name": "m"}},
+        "2": {"class_type": "EmptyLatentImage", "inputs": {"width": 64}},
+        "17": {"class_type": "PrimitiveStringMultiline", "inputs": {"value": "a car"}},
+        "14": {
+            "class_type": "PrimitiveStringMultiline",
+            "inputs": {"value": "You are a prompt engineer."},
+        },
+        "11": {
+            "class_type": "TextGenerate",
+            "inputs": {
+                "prompt": ["17", 0],
+                "system_prompt": ["14", 0],
+                "clip": ["1", 1],
+                "max_length": 4096,
+                "sampling_mode.seed": 0,
+            },
+        },
+        "36": {
+            "class_type": "PromptManagerPair",
+            "inputs": {
+                "text": ["11", 0],
+                "negative_text": "",
+                "prepend_text": "",
+                "append_text": "",
+                "clip": ["1", 1],
+            },
+        },
+        "7": {
+            "class_type": "KSampler",
+            "inputs": {
+                "seed": 1,
+                "steps": 20,
+                "cfg": 1.0,
+                "sampler_name": "euler",
+                "scheduler": "simple",
+                "denoise": 1.0,
+                "model": ["1", 0],
+                "positive": ["36", 0],
+                "negative": ["36", 1],
+                "latent_image": ["2", 0],
+            },
+        },
+        "8": {
+            "class_type": "VAEDecode",
+            "inputs": {"samples": ["7", 0], "vae": ["1", 2]},
+        },
+        "9": {
+            "class_type": "SaveImage",
+            "inputs": {"filename_prefix": "moss", "images": ["8", 0]},
+        },
+    }
+
+
+def two_generated_branches():
+    """``text_generate_run`` plus a second generated branch saved by an unknown
+    saver and also previewed."""
+    graph = text_generate_run()
+    graph["31"] = {**graph["11"], "inputs": {**graph["11"]["inputs"]}}
+    graph["37"] = {
+        "class_type": "PromptManagerPair",
+        "inputs": {**graph["36"]["inputs"], "text": ["31", 0]},
+    }
+    graph["20"] = {
+        "class_type": "KSampler",
+        "inputs": {
+            **graph["7"]["inputs"],
+            "positive": ["37", 0],
+            "negative": ["37", 1],
+        },
+    }
+    graph["21"] = {
+        "class_type": "VAEDecode",
+        "inputs": {"samples": ["20", 0], "vae": ["1", 2]},
+    }
+    graph["22"] = {
+        "class_type": "Image Save",
+        "inputs": {"filename_prefix": "other", "images": ["21", 0]},
+    }
+    graph["23"] = {"class_type": "PreviewImage", "inputs": {"images": ["21", 0]}}
+    return graph
+
+
+class TestRecordedTexts(unittest.TestCase):
+    """The texts PromptManager recorded at execution, per sampler of the run."""
+
+    def test_a_run_time_text_shows_what_the_run_recorded(self):
+        alone = _evidence(text_generate_run(), file_name="moss_00026_.png")
+        self.assertEqual(alone["resolution"]["reason_code"], "runtime-value")
+        evidence = _evidence(
+            text_generate_run(),
+            file_name="moss_00026_.png",
+            recorded_saver_id="9",
+            recorded_texts={"7": {"positive": LLM_TEXT, "negative": ""}},
+        )
+        self.assertEqual(
+            evidence["prompts"],
+            {
+                "positive": {
+                    "status": "resolved",
+                    "text": LLM_TEXT,
+                    "source": "execution",
+                },
+                "negative": {"status": "resolved", "text": "", "source": "execution"},
+            },
+        )
+        self.assertEqual(evidence["resolution"]["state"], "resolved")
+        self.assertEqual(evidence["resolution"]["sampler_id"], "7")
+
+    def test_an_unrecorded_negative_stays_open(self):
+        evidence = _evidence(
+            text_generate_run(),
+            recorded_saver_id="9",
+            recorded_texts={"7": {"positive": LLM_TEXT, "negative": None}},
+        )
+        self.assertEqual(evidence["prompts"]["positive"]["text"], LLM_TEXT)
+        self.assertEqual(evidence["prompts"]["negative"]["status"], "unresolved")
+
+    def test_texts_that_do_not_name_the_open_sampler_change_nothing(self):
+        alone = _evidence(text_generate_run())
+        for texts in (
+            {"8": {"positive": LLM_TEXT, "negative": ""}},
+            {"7": {"positive": None, "negative": ""}},
+            {"7": {"positive": "  ", "negative": ""}},
+            {"7": "not roles"},
+            {},
+        ):
+            with self.subTest(texts=texts):
+                self.assertEqual(
+                    _evidence(
+                        text_generate_run(), recorded_saver_id="9", recorded_texts=texts
+                    ),
+                    alone,
+                )
+
+    def test_texts_need_the_saver_history_recorded_to_lead_to_the_sampler(self):
+        texts = {"7": {"positive": LLM_TEXT, "negative": ""}}
+        alone = _evidence(text_generate_run())
+        for saver in (None, "missing", "2"):
+            with self.subTest(saver=saver):
+                self.assertEqual(
+                    _evidence(
+                        text_generate_run(),
+                        recorded_saver_id=saver,
+                        recorded_texts=texts,
+                    ),
+                    alone,
+                )
+
+    def test_texts_follow_the_recorded_saver_not_another_sampler(self):
+        # Review: the texts are those of the sampler history's saver leads to.
+        graph = two_generated_branches()
+        texts = {
+            "7": {"positive": "STAGE X TEXT", "negative": ""},
+            "20": {"positive": "STAGE Y TEXT", "negative": ""},
+        }
+        for saver, text, sampler in (
+            ("22", "STAGE Y TEXT", "20"),
+            ("9", "STAGE X TEXT", "7"),
+        ):
+            with self.subTest(saver=saver):
+                evidence = _evidence(
+                    graph,
+                    file_name="renamed_00001_.png",
+                    recorded_saver_id=saver,
+                    recorded_texts=texts,
+                )
+                self.assertEqual(evidence["prompts"]["positive"]["text"], text)
+                self.assertEqual(evidence["resolution"]["sampler_id"], sampler)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_run_attribution.py

@@ -9,13 +9,22 @@ import unittest
 from pathlib import Path
 from unittest import mock
 
-from database.run_attribution import stored_run_prompt
+from database.run_attribution import (
+    RecordedOutput,
+    recorded_output,
+    stored_run_prompt,
+)
 from tests import test_execution_history_database as history_tests
 from tests.background_stops import close_handler
 from tests.test_api_admin import AdminAPITestCase
-from tests.test_execution_context_database import snapshot
+from tests.test_execution_context_database import known, snapshot
 from tests.test_prompt_attribution import issue_graph, llm_text_graph, make_tracker
-from utils.execution_contexts import ContextBundle, RecipeContext
+from utils.execution_contexts import (
+    ContextBundle,
+    PromptSnapshot,
+    RecipeContext,
+    TextEvidence,
+)
 from utils.execution_outputs import (
     OutputBundle,
     OutputReference,
@@ -38,28 +47,34 @@ class RecordedRuns:
     start = history_tests.HistoryDatabaseTests.start
     finish = history_tests.HistoryDatabaseTests.finish
 
-    def record_run(self, db, output_root, files, positives, item=0, window=None):
+    def record_run(
+        self, db, output_root, files, positives, item=0, window=None, nodes=None
+    ):
         started, finished = window or (now_ms() - 5000, now_ms() + 5000)
         start = self.start(db, item, started)
         db.write_execution_start(start)
         finish = self.finish(start, time=finished)
         db.write_execution_finish(finish)
-        db.write_execution_contexts(
-            start.run_id,
-            start.storage_epoch,
-            ContextBundle(
-                tuple(
-                    RecipeContext(str(4 + i), snapshot(text), snapshot(""), "inferred")
-                    for i, text in enumerate(positives)
+        if positives:  # a run with no PromptManager node records no contexts
+            db.write_execution_contexts(
+                start.run_id,
+                start.storage_epoch,
+                ContextBundle(
+                    tuple(
+                        RecipeContext(
+                            str(4 + i), snapshot(text), snapshot(""), "inferred"
+                        )
+                        for i, text in enumerate(positives)
+                    ),
+                    "complete",
+                    "captured",
                 ),
-                "complete",
-                "captured",
-            ),
-        )
+            )
         root = snapshot_output_roots(STYLE, output_root, output_root).output_key
+        nodes = nodes or ["9"] * len(files)
         references = tuple(
-            OutputReference("9", i, "output", sub, name, STYLE, root)
-            for i, (sub, name) in enumerate(files)
+            OutputReference(node, i, "output", sub, name, STYLE, root)
+            for i, (node, (sub, name)) in enumerate(zip(nodes, files))
         )
         db.write_execution_outputs_for_finish(
             finish, OutputBundle(references, "complete", "reported")
@@ -171,6 +186,186 @@ class TestStoredRunPrompt(RecordedRuns, unittest.TestCase):
         self.assertEqual(linked, [prompt_id])
 
 
+class TestRecordedOutputNode(RecordedRuns, unittest.TestCase):
+    """#386: the save node history says wrote a file, under the same rules."""
+
+    close_databases = history_tests.HistoryDatabaseTests.close_databases
+    database = history_tests.HistoryDatabaseTests.database
+    setUp = TestStoredRunPrompt.setUp
+
+    def node(self, path=None):
+        found = recorded_output(self.db, path or self.image, self.output)
+        return found.saver_id if found is not None else None
+
+    def test_one_run_one_saver_names_the_saver(self):
+        other = os.path.join(self.output, "sub", "img_00002_.png")
+        Path(other).write_bytes(b"png")
+        self.record_run(
+            self.db,
+            self.output,
+            [("sub", "img_00001_.png"), ("sub", "img_00002_.png")],
+            [],
+            nodes=["13", "23"],
+        )
+        self.assertEqual(self.node(), "13")
+        self.assertEqual(self.node(other), "23")
+
+    def test_a_run_without_prompt_manager_contexts_still_names_the_saver(self):
+        self.record_run(self.db, self.output, [("sub", "img_00001_.png")], [])
+        self.assertEqual(self.node(), "9")
+
+    def test_two_runs_reporting_the_file_name_nothing(self):
+        self.record_run(
+            self.db, self.output, [("sub", "img_00001_.png")], [], nodes=["13"]
+        )
+        self.record_run(
+            self.db, self.output, [("sub", "img_00001_.png")], [], item=1, nodes=["23"]
+        )
+        self.assertIsNone(self.node())
+
+    def test_one_run_reporting_the_file_from_two_savers_names_nothing(self):
+        self.record_run(
+            self.db,
+            self.output,
+            [("sub", "img_00001_.png"), ("sub", "img_00001_.png")],
+            [],
+            nodes=["13", "23"],
+        )
+        self.assertIsNone(self.node())
+
+    def test_a_file_written_outside_the_run_names_nothing(self):
+        self.record_run(
+            self.db,
+            self.output,
+            [("sub", "img_00001_.png")],
+            [],
+            window=(1_000_000, 2_000_000),
+        )
+        self.assertIsNone(self.node())
+
+    def test_unrecorded_outside_or_unknown_roots_name_nothing(self):
+        self.record_run(self.db, self.output, [("sub", "img_00001_.png")], [])
+        other = os.path.join(self.output, "sub", "other.png")
+        Path(other).write_bytes(b"png")
+        self.assertIsNone(self.node(other))
+        self.assertIsNone(recorded_output(self.db, self.image, None))
+        self.assertIsNone(recorded_output(self.db, "", self.output))
+
+    def test_the_run_texts_recorded_at_execution_come_with_the_saver(self):
+        self.record_run(
+            self.db, self.output, [("sub", "img_00001_.png")], ["an llm prompt"]
+        )
+        self.assertEqual(
+            recorded_output(self.db, self.image, self.output),
+            RecordedOutput("9", {"4": {"positive": "an llm prompt", "negative": ""}}),
+        )
+
+    def test_texts_only_inferred_from_the_graph_are_not_recorded(self):
+        start = self.start(self.db, 0, now_ms() - 5000)
+        self.db.write_execution_start(start)
+        finish = self.finish(start, time=now_ms() + 5000)
+        self.db.write_execution_finish(finish)
+        inferred = TextEvidence("graph text", "inferred", "graph_literal", None)
+        self.db.write_execution_contexts(
+            start.run_id,
+            start.storage_epoch,
+            ContextBundle(
+                (
+                    RecipeContext(
+                        "7",
+                        PromptSnapshot(inferred, inferred, known(""), known("")),
+                        snapshot("kept"),
+                        "inferred",
+                    ),
+                ),
+                "complete",
+                "captured",
+            ),
+        )
+        root = snapshot_output_roots(STYLE, self.output, self.output).output_key
+        self.db.write_execution_outputs_for_finish(
+            finish,
+            OutputBundle(
+                (
+                    OutputReference(
+                        "9", 0, "output", "sub", "img_00001_.png", STYLE, root
+                    ),
+                ),
+                "complete",
+                "reported",
+            ),
+        )
+        self.assertEqual(
+            recorded_output(self.db, self.image, self.output).texts,
+            {"7": {"positive": None, "negative": "kept"}},
+        )
+
+    def test_several_savers_name_no_saver_but_keep_the_texts(self):
+        self.record_run(
+            self.db,
+            self.output,
+            [("sub", "img_00001_.png"), ("sub", "img_00001_.png")],
+            ["text"],
+            nodes=["13", "23"],
+        )
+        found = recorded_output(self.db, self.image, self.output)
+        self.assertIsNone(found.saver_id)
+        self.assertEqual(found.texts, {"4": {"positive": "text", "negative": ""}})
+
+    def test_the_route_lookup_logs_a_failure_and_names_nothing(self):
+        from database.run_attribution import recorded_output_lookup
+
+        broken = mock.Mock()
+        broken.model.get_connection.return_value.execute.side_effect = (
+            sqlite3.OperationalError("locked")
+        )
+        broken.get_execution_contexts.side_effect = sqlite3.OperationalError("x")
+        with mock.patch(
+            "database.run_attribution.host_output_directory", return_value=self.output
+        ):
+            self.assertIsNone(recorded_output_lookup(broken)(self.image))
+            with (
+                mock.patch(
+                    "database.run_attribution._recorded_runs",
+                    side_effect=sqlite3.OperationalError("locked"),
+                ),
+                self.assertLogs("prompt_manager.run_attribution", "WARNING"),
+            ):
+                self.assertIsNone(
+                    recorded_output_lookup(broken, close=True)(self.image)
+                )
+        broken.model.close.assert_called_once_with()
+
+    def test_a_history_reference_spells_the_file_as_history_recorded_it(self):
+        # On Windows the history route's target is normcased (lower case);
+        # history matches names exactly, so the route passes its reference.
+        from database.run_attribution import recorded_output_lookup
+
+        upper = os.path.join(self.output, "sub", "IMG_00001_.png")
+        Path(upper).write_bytes(b"png")
+        self.record_run(self.db, self.output, [("sub", "IMG_00001_.png")], [])
+        target = os.path.join(self.output, "sub", "img_00001_.png")
+        reference = {
+            "folder_type": "output",
+            "subfolder": "sub",
+            "filename": "IMG_00001_.png",
+        }
+        with mock.patch(
+            "database.run_attribution.host_output_directory", return_value=self.output
+        ):
+            found = recorded_output_lookup(self.db, reference=reference)(target)
+            self.assertEqual(found.saver_id, "9")
+            temp = {**reference, "folder_type": "temp"}
+            self.assertIsNone(recorded_output_lookup(self.db, reference=temp)(target))
+
+    def test_a_library_without_history_tables_names_nothing(self):
+        broken = mock.Mock()
+        broken.model.get_connection.return_value.execute.side_effect = (
+            sqlite3.OperationalError("no such table")
+        )
+        self.assertIsNone(recorded_output(broken, self.image, self.output))
+
+
 class TestScanUsesHistory(RecordedRuns, AdminAPITestCase):
     """The output scan applies the same rules as the monitor."""
 

utils/prompt_inspector.py

@@ -10,6 +10,7 @@ import threading
 import zlib
 from dataclasses import dataclass, field as dataclass_field
 from pathlib import Path
+from collections.abc import Mapping
 from typing import Any
 
 from PIL import Image
@@ -461,6 +462,14 @@ def read_inspection_metadata(path: Path) -> dict[str, str]:
     return read_inspection_file(path).metadata
 
 
+# A side whose branch recorded run history named (#386): the save node that
+# wrote the file picked the sampler; the text is still that branch's graph walk.
+RECORDED_RUN_SOURCE = "recorded-run"
+RECORDED_TEXT_REASON = "PromptManager recorded this text when the run executed"
+# Outcomes the file alone leaves open; only these ask recorded history.
+_UNDECIDED = frozenset({"ambiguous", "unresolved"})
+
+
 def _side(status: str, text: str | None = None, source: str | None = None) -> dict:
     return {"status": status, "text": text, "source": source}
 
@@ -529,15 +538,60 @@ def _parsed_workflow(value: Any) -> dict | None:
     return value if isinstance(value, dict) else None
 
 
+def _recorded_execution(resolved: dict, texts: Mapping) -> dict | None:
+    """Evidence from the texts recorded at one sampler of the file's run (#386).
+
+    Only for an outcome that names one sampler (a single stage stopped at a
+    run-time value, such as a language model's output feeding PromptManager)
+    and only when that sampler's positive text was recorded and is not empty.
+    """
+    sampler = resolved.get("sampler_id")
+    roles = texts.get(sampler) if isinstance(sampler, str) else None
+    if not isinstance(roles, Mapping):
+        return None
+    positive = roles.get("positive")
+    if not isinstance(positive, str) or not positive.strip():
+        return None
+    negative = roles.get("negative")
+    prompts = {
+        "positive": _side("resolved", positive, "execution"),
+        "negative": (
+            _side("resolved", negative, "execution")
+            if isinstance(negative, str)
+            else _side("unresolved")
+        ),
+    }
+    resolution = {
+        "state": "resolved",
+        "sampler_id": sampler,
+        "reason": RECORDED_TEXT_REASON,
+        "reason_code": None,
+        "node": None,
+    }
+    return {"prompts": prompts, "resolution": resolution}
+
+
 def inspect_prompt_evidence(
     metadata: dict,
     saved: dict | None = None,
     *,
     profile: ResolverProfile = INSPECTION_PROFILE,
     file_name: str | None = None,
+    recorded_saver_id: str | None = None,
+    recorded_texts: Mapping[str, Mapping[str, str | None]] | None = None,
 ) -> dict:
     """Keep uncertainty visible; a saved association never replaces file evidence.
 
+    ``recorded_saver_id`` is the save node recorded run history says wrote the
+    file (#386). It is asked only when the file alone leaves the prompt open
+    (ambiguous or unresolved), and the anchored result is kept only when it
+    names the positive (resolved, or a guider with no negative); its
+    graph-inferred sides are then labelled ``recorded-run``. ``recorded_texts``
+    (sampler id to the effective texts PromptManager recorded at execution in
+    that run) is used last: when the prompt is still open at one named sampler
+    whose positive text was recorded, the recorded texts are shown as
+    ``execution``.
+
     Inspection routes resolve with ``INSPECTION_PROFILE``: bounded work, not
     graph size. Ingestion (scan, live monitor, admin import) calls the resolver
     directly with its default profile. ``resolution`` carries the closed
@@ -599,6 +653,24 @@ def inspect_prompt_evidence(
         # Only to recognize overwritten PromptManager text (R1); never a role.
         canonical["workflow"] = workflow
     resolved = resolve_prompt_pair(canonical, profile=profile, file_name=file_name)
+    recorded = False
+    if resolved["state"] in _UNDECIDED and recorded_saver_id is not None:
+        anchored = resolve_prompt_pair(
+            canonical,
+            profile=profile,
+            file_name=file_name,
+            recorded_saver_id=recorded_saver_id,
+        )
+        # Only a result whose samplers history's saver chose is about this file.
+        if anchored.get("recorded_anchor") is True:
+            if anchored["state"] not in _UNDECIDED and isinstance(
+                anchored["positive"], dict
+            ):
+                resolved, recorded = anchored, True
+            elif recorded_texts:
+                executed = _recorded_execution(anchored, recorded_texts)
+                if executed is not None:
+                    return executed
     state = resolved["state"]
     result = {}
     for role in ("positive", "negative"):
@@ -608,9 +680,10 @@ def inspect_prompt_evidence(
             and isinstance(side, dict)
             and (role == "negative" or side["text"].strip())
         ):
-            result[role] = _side(
-                "resolved", side["text"], side["provenance"]["text_source"]
-            )
+            source = side["provenance"]["text_source"]
+            if recorded and source == "graph-inferred":
+                source = RECORDED_RUN_SOURCE
+            result[role] = _side("resolved", side["text"], source)
         elif state == "resolved" and isinstance(side, dict):
             # A resolved pair withholds only an empty positive.
             result[role] = {**_side("unresolved"), "message": _EMPTY_POSITIVE}

utils/prompt_pair_resolver.py

@@ -1601,6 +1601,11 @@ class _GraphResolver:
         several, an unreadable walk or no candidate sampler is ``None``: the
         file is then judged unanchored, never guessed (#381).
         """
+        match = self.saver_match(file_name)
+        return self.saver_walk(match) if match is not None else None
+
+    def saver_match(self, file_name: Any) -> Optional[str]:
+        """The one node whose ``filename_prefix`` names ``file_name``, or None."""
         stem = _file_stem(file_name)
         if stem is None:
             return None
@@ -1611,15 +1616,31 @@ class _GraphResolver:
                 for node_id, node in self.graph.items()
                 if self.writes(node, stem)
             ]
-            if len(matches) != 1:
-                return None
-            _, inputs = self.node(matches[0])
+        except (_Unresolved, _Diverging):
+            return None
+        return matches[0] if len(matches) == 1 else None
+
+    def recorded_saver_stages(self, node_id: str) -> Optional[list]:
+        """The candidate samplers of the save node recorded history names (#386).
+
+        The node is walked exactly like a file-name match. A node the graph
+        lacks, a failing walk or no candidate sampler is ``None``: unanchored.
+        """
+        if not isinstance(self.graph.get(node_id), dict):
+            return None
+        self.budget.reset(self.per_candidate_visits)
+        return self.saver_walk(node_id)
+
+    def saver_walk(self, saver_id: str) -> Optional[list]:
+        """The listed samplers in one saver's linked inputs' ancestry, or None."""
+        try:
+            _, inputs = self.node(saver_id)
             values = [
                 value
                 for name, value in inputs.items()
                 if name != "filename_prefix" and _link(value)
             ]
-            found = self.stages_from(matches[0], values)
+            found = self.stages_from(saver_id, values)
         except (_Unresolved, _Diverging):
             return None
         candidates = [
@@ -2256,6 +2277,7 @@ def resolve_prompt_pair(
     max_depth: Optional[int] = None,
     max_text_length: Optional[int] = None,
     file_name: Optional[str] = None,
+    recorded_saver_id: Optional[str] = None,
 ) -> Dict[str, Any]:
     """Resolve parsed metadata; a present API graph is authoritative over text.
 
@@ -2268,9 +2290,14 @@ def resolve_prompt_pair(
     ``profile`` selects the bounds; explicit ``max_*`` keywords override it.
     ``file_name`` (inspection only, unanchored) anchors on the one saver that
     wrote that file (``_GraphResolver.saver_stages``) when exactly one matches.
+    ``recorded_saver_id`` (inspection only, unanchored) is the save node that
+    recorded history says wrote the file (#386); it is tried before the file
+    name, and a node that yields no candidate sampler anchors nothing.
     """
     profile = _profile_with(profile, max_visits, max_depth, max_text_length)
-    result = _resolve(metadata, output_node_id, adapters, profile, file_name)
+    result = _resolve(
+        metadata, output_node_id, adapters, profile, file_name, recorded_saver_id
+    )
     return _legacy_result(result) if profile.legacy else result
 
 
@@ -2324,6 +2351,7 @@ def _resolve(
     adapters: Optional[Mapping[Tuple[str, int], OutputAdapter]],
     profile: ResolverProfile,
     file_name: Optional[str] = None,
+    recorded_saver_id: Optional[str] = None,
 ) -> Dict[str, Any]:
     if not isinstance(metadata, dict) or not metadata:
         return _result("absent", reason="No parsed prompt metadata")
@@ -2379,11 +2407,12 @@ def _resolve(
                 "unresolved", reason=str(error), reason_code=error.code, node=error.node
             )
     else:
-        anchored = (
-            resolver.saver_stages(file_name)
-            if semantics is not None and file_name is not None
-            else None
-        )
+        anchored, by_record = None, False
+        if semantics is not None and recorded_saver_id is not None:
+            anchored = _recorded_anchor(resolver, str(recorded_saver_id), file_name)
+            by_record = anchored is not None
+        elif semantics is not None and file_name is not None:
+            anchored = resolver.saver_stages(file_name)
         sampler_ids = anchored or [
             node_id
             for node_id, node in normalized.items()
@@ -2393,6 +2422,34 @@ def _resolve(
         ]
         if semantics is not None and anchored is None:
             sampler_ids = resolver.final_stages(sampler_ids)
+        if semantics is not None and recorded_saver_id is not None:
+            # The caller can tell whether history's saver chose the samplers.
+            return {
+                **_judge(resolver, sampler_ids, normalized, profile),
+                "recorded_anchor": by_record,
+            }
+    return _judge(resolver, sampler_ids, normalized, profile)
+
+
+def _recorded_anchor(
+    resolver: "_GraphResolver", saver_id: str, file_name: Optional[str]
+) -> Optional[list]:
+    """The samplers of the saver history recorded, or None (#386).
+
+    A file name that names a different saver contradicts history: no anchor.
+    """
+    named = resolver.saver_match(file_name) if file_name is not None else None
+    if named is not None and named != saver_id:
+        return None
+    return resolver.recorded_saver_stages(saver_id)
+
+
+def _judge(
+    resolver: "_GraphResolver",
+    sampler_ids: list,
+    normalized: dict,
+    profile: ResolverProfile,
+) -> Dict[str, Any]:
     if not sampler_ids:
         return _no_sampler(normalized, profile)
     if len(sampler_ids) > 1:

web/js/workspace/file-inspector.mjs

@@ -158,9 +158,11 @@ function generationFacts(record, listClass, noteClass) {
 }
 // D21: the panel badge names the strongest resolved prompt source; any other source shows no badge.
 // A prompt that left out a run-time PromptManager part (D125) or a later stage's added conditioning (D126) is still
-// inferred from the workflow.
+// inferred from the workflow. "Recorded run" (#386): recorded run history named the save node that wrote the file; it
+// ranks below any workflow-inferred side, so a branch with an omitted or added part still reads "Workflow inferred".
 const EVIDENCE_BADGES = [['execution', 'Execution recorded'], ['graph-inferred', 'Workflow inferred'], ['runtime-parts-omitted', 'Workflow inferred'],
-    ['added-conditioning', 'Workflow inferred'], ['runtime-parts-omitted-and-added-conditioning', 'Workflow inferred'], ['embedded-parameters', 'Embedded parameters']];
+    ['added-conditioning', 'Workflow inferred'], ['runtime-parts-omitted-and-added-conditioning', 'Workflow inferred'], ['recorded-run', 'Recorded run'],
+    ['embedded-parameters', 'Embedded parameters']];
 function evidenceBadge(prompts) {
     const sources = new Set(Object.values(prompts).filter(role => role?.status === 'resolved').map(role => role.source));
     return EVIDENCE_BADGES.find(([source]) => sources.has(source))?.[1] ?? null;