fix(metadata): preserve bounded paired execution evidence (#193)

prompt_manager_pair.py

@@ -8,6 +8,7 @@ from typing import Any, Dict, Optional, Tuple
 try:
     from .prompt_manager import IO, InputTypeDict, PromptManagerFocused
     from .utils.hashing import generate_prompt_hash
+    from .utils.pair_execution import record_pair_execution
     from .utils.prompt_graph import intrinsic_pair_nodes, run_prompt_nodes
     from .utils.prompt_pair_resolver import resolve_text_input
     from .utils.usage_tracking import (
@@ -18,6 +19,7 @@ try:
 except ImportError:
     from prompt_manager import IO, InputTypeDict, PromptManagerFocused
     from utils.hashing import generate_prompt_hash
+    from utils.pair_execution import record_pair_execution
     from utils.prompt_graph import intrinsic_pair_nodes, run_prompt_nodes
     from utils.prompt_pair_resolver import resolve_text_input
     from utils.usage_tracking import (
@@ -41,6 +43,7 @@ class PromptManagerPair(PromptManagerFocused):
             ),
             "clip": inputs["required"]["clip"],
         }
+        inputs["hidden"]["extra_pnginfo"] = "EXTRA_PNGINFO"
         return inputs
 
     RETURN_TYPES = (IO.CONDITIONING, IO.CONDITIONING, IO.STRING, IO.STRING)
@@ -65,6 +68,7 @@ class PromptManagerPair(PromptManagerFocused):
         append_text: str = "",
         prompt: Optional[Dict[str, Any]] = None,
         unique_id: Optional[str] = None,
+        extra_pnginfo: Optional[Dict[str, Any]] = None,
     ) -> Tuple[Any, Any, str, str]:
         """Encode both sides once; database and tracking failures never block CLIP."""
         if clip is None:
@@ -84,6 +88,16 @@ class PromptManagerPair(PromptManagerFocused):
         negative_conditioning = clip.encode_from_tokens_scheduled(
             clip.tokenize(negative_text)
         )
+        record_pair_execution(
+            extra_pnginfo,
+            unique_id,
+            text=text,
+            negative_text=negative_text,
+            positive_text=positive,
+            negative_effective_text=negative_text,
+            prepend_text=prepend_text.strip(),
+            append_text=append_text.strip(),
+        )
         pair_id = prompt_id = None
         known = isinstance(prompt, dict) and unique_id is not None
         node_id = str(unique_id)

tests/test_pair_execution.py

@@ -0,0 +1,322 @@
+"""Embedded actual encoding evidence is used only for its traced paired node."""
+
+import json
+import logging
+import tempfile
+import unittest
+from pathlib import Path
+from unittest.mock import Mock
+
+from PIL import Image
+from PIL.PngImagePlugin import PngInfo
+
+from database.operations import PromptDatabase
+from prompt_manager_pair import PromptManagerPair
+from utils.metadata_extractor import ComfyUIMetadataExtractor
+from utils.prompt_pair_resolver import resolve_prompt_pair
+
+KEY = "comfyassets_prompt_pairs"
+
+
+def graph(positive="cat", negative="blur"):
+    return {
+        "1": {
+            "class_type": "PromptManagerPair",
+            "inputs": {"text": positive, "negative_text": negative},
+        },
+        "5": {
+            "class_type": "KSampler",
+            "inputs": {"positive": ["1", 0], "negative": ["1", 1]},
+        },
+    }
+
+
+class TestPairExecution(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(self.tmp.cleanup)
+        self.db = PromptDatabase(str(Path(self.tmp.name) / "test.db"))
+        self.addCleanup(self.db.close_all)
+        self.node = PromptManagerPair.__new__(PromptManagerPair)
+        self.node.db = self.db
+        self.node.logger = logging.getLogger("test.execution")
+        self.node.prompt_tracker = Mock()
+        self.node.comfyui_integration = Mock()
+        self.node._inject_lora_trigger_words = lambda text: text + " LORA"
+        self.extra = {"workflow": {"unchanged": True}, "sibling": {"keep": True}}
+
+    def execute(self, text="cat", negative="blur", **kwargs):
+        return self.node.encode_prompt(
+            Mock(),
+            text,
+            negative_text=negative,
+            unique_id="1",
+            extra_pnginfo=self.extra,
+            **kwargs,
+        )
+
+    def resolve(self, source=None, extra=None):
+        return resolve_prompt_pair(
+            {"prompt": source or graph(), **(self.extra if extra is None else extra)}
+        )
+
+    def test_schema_requests_extra_pnginfo_and_actual_positive_lora_is_preserved(self):
+        self.assertEqual(
+            PromptManagerPair.INPUT_TYPES()["hidden"]["extra_pnginfo"], "EXTRA_PNGINFO"
+        )
+        self.execute(prepend_text="trigger")
+        source = graph()
+        source["1"]["inputs"]["prepend_text"] = "trigger"
+        result = self.resolve(source)
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(result["positive"]["text"], "trigger cat LORA")
+        self.assertEqual(result["positive"]["storage_text"], "cat")
+        self.assertEqual(result["positive"]["provenance"]["text_source"], "execution")
+        self.assertEqual(result["negative"]["text"], "blur")
+        self.assertEqual(self.extra["workflow"], {"unchanged": True})
+        self.assertEqual(self.extra["sibling"], {"keep": True})
+
+    def test_one_unknown_linked_execution_is_recoverable_without_fifo_guess(self):
+        source = graph(positive=["2", 0])
+        source["2"] = {"class_type": "PromptSearchList", "inputs": {}}
+        self.execute(text="actual batch cat")
+        result = self.resolve(source)
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(result["positive"]["storage_text"], "actual batch cat")
+        self.assertEqual(result["positive"]["text"], "actual batch cat LORA")
+
+    def test_distinct_batch_records_are_unresolved_instead_of_using_last_execution(
+        self,
+    ):
+        source = graph(positive=["2", 0])
+        source["2"] = {"class_type": "PromptSearchList", "inputs": {}}
+        self.execute(text="first")
+        self.execute(text="second")
+        result = self.resolve(source)
+        self.assertEqual(result["state"], "unresolved")
+        self.assertIn("multiple", result["reason"].lower())
+
+    def test_empty_negative_runtime_record_remains_explicit(self):
+        self.execute(negative="")
+        result = self.resolve(graph(negative=""))
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(result["negative"]["text"], "")
+
+    def test_cached_graph_without_execution_evidence_is_explicitly_inferred(self):
+        result = self.resolve(extra={})
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(
+            result["positive"]["provenance"]["text_source"], "graph-inferred"
+        )
+
+    def test_stale_or_malformed_evidence_is_not_accepted(self):
+        self.execute(text="stale")
+        self.assertEqual(self.resolve()["state"], "unresolved")
+        for value in (
+            "not json",
+            [],
+            {"version": 900, "nodes": {}},
+            {"version": 1, "nodes": {"1": {"records": [{"bad": True}]}}},
+        ):
+            with self.subTest(value=value):
+                self.assertEqual(
+                    self.resolve(extra={KEY: value})["state"], "unresolved"
+                )
+
+    def test_unrelated_node_evidence_does_not_override_plain_encoders(self):
+        source = graph()
+        source["1"] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": "plain positive"},
+        }
+        source["2"] = {
+            "class_type": "CLIPTextEncode",
+            "inputs": {"text": "plain negative"},
+        }
+        source["5"]["inputs"]["negative"] = ["2", 0]
+        result = self.resolve(source, extra={KEY: "broken"})
+        self.assertEqual(result["state"], "resolved")
+        self.assertEqual(result["positive"]["text"], "plain positive")
+
+    def test_png_extractor_preserves_namespace_for_live_and_raw_scan_readers(self):
+        self.execute()
+        path = Path(self.tmp.name) / "image.png"
+        info = PngInfo()
+        info.add_text("prompt", json.dumps(graph()))
+        info.add_text(KEY, json.dumps(self.extra[KEY]))
+        Image.new("RGB", (2, 2)).save(path, pnginfo=info)
+        parsed = ComfyUIMetadataExtractor().extract_metadata(str(path))
+        self.assertEqual(resolve_prompt_pair(parsed)["positive"]["text"], "cat LORA")
+        with Image.open(path) as image:
+            raw = dict(image.text)
+        raw["prompt"] = json.loads(raw["prompt"])
+        self.assertEqual(resolve_prompt_pair(raw)["positive"]["text"], "cat LORA")
+
+    def test_recording_limits_fail_closed_and_repeat_identical_values_are_safe(self):
+        from utils.pair_execution import MAX_EXECUTION_RECORDS
+
+        self.execute()
+        self.execute()
+        self.assertEqual(self.resolve()["state"], "resolved")
+        for index in range(MAX_EXECUTION_RECORDS + 1):
+            self.execute(text=str(index))
+        self.assertEqual(self.resolve()["state"], "unresolved")
+
+    def test_live_metadata_lookup_uses_unique_execution_for_unknown_source(self):
+        from utils.image_monitor import ImageGenerationHandler
+
+        source = graph(positive=["2", 0])
+        source["2"] = {"class_type": "PromptSearchList", "inputs": {}}
+        self.execute(text="actual")
+        handler = ImageGenerationHandler.__new__(ImageGenerationHandler)
+        handler.db_manager = self.db
+        handler.logger = logging.getLogger("test.execution")
+        state, context = handler._find_prompt_from_metadata(
+            {"prompt": source, **self.extra}
+        )
+        self.assertEqual(state, "linked")
+        expected = self.db.list_prompt_pairs()["pairs"][0]["positive_prompt_id"]
+        self.assertEqual(context["id"], expected)
+
+    def test_submission_clears_old_execution_evidence_without_changing_workflow(self):
+        from utils.usage_tracking import handle_queued_prompt, QueuedRuns
+
+        self.execute()
+        request = {"prompt": graph(), "extra_data": {"extra_pnginfo": self.extra}}
+        handle_queued_prompt(request, QueuedRuns(), pair_queued=QueuedRuns())
+        self.assertNotIn(KEY, self.extra)
+        self.assertEqual(self.extra["workflow"], {"unchanged": True})
+        self.assertEqual(
+            self.resolve()["positive"]["provenance"]["text_source"], "graph-inferred"
+        )
+
+    def test_known_invalid_output_slot_or_cycles_cannot_be_overridden(self):
+        self.execute()
+        source = graph(positive=["2", 4])
+        source["2"] = {"class_type": "PrimitiveString", "inputs": {"value": "cat"}}
+        self.assertEqual(self.resolve(source)["state"], "unresolved")
+        source["2"] = {
+            "class_type": "StringConcatenate",
+            "inputs": {"string_a": ["2", 0], "string_b": "cat"},
+        }
+        source["1"]["inputs"]["text"] = ["2", 0]
+        self.assertEqual(self.resolve(source)["state"], "unresolved")
+
+    def test_live_image_association_keeps_actual_encoding_and_repeated_scan_is_stable(
+        self,
+    ):
+        from utils.image_monitor import ImageGenerationHandler
+        from database.pair_ingestion import ingest_resolved_pair
+
+        self.execute()
+        metadata = {"prompt": graph(), **self.extra}
+        handler = ImageGenerationHandler.__new__(ImageGenerationHandler)
+        handler.db_manager = self.db
+        handler.logger = logging.getLogger("test.execution")
+        _, context = handler._find_prompt_from_metadata(metadata)
+        path = str(Path(self.tmp.name) / "actual.png")
+        handler.link_image_to_prompt(path, context, metadata)
+        pair_context = self.db.get_image_prompt_info(path)["pair_context"]
+        self.assertEqual(pair_context["positive_text"], "cat LORA")
+        repeated = ingest_resolved_pair(
+            self.db, path, metadata, resolve_prompt_pair(metadata)
+        )
+        self.assertEqual(repeated["status"], "unchanged")
+
+    def test_unknown_batch_with_multiple_executions_is_not_exactly_linked(self):
+        from utils.image_monitor import ImageGenerationHandler
+
+        source = graph(positive=["2", 0])
+        source["2"] = {"class_type": "PromptSearchList", "inputs": {}}
+        self.execute(text="first")
+        self.execute(text="second")
+        handler = ImageGenerationHandler.__new__(ImageGenerationHandler)
+        handler.db_manager = self.db
+        handler.logger = logging.getLogger("test.execution")
+        state, context = handler._find_prompt_from_metadata(
+            {"prompt": source, **self.extra}
+        )
+        self.assertEqual(state, "unknown")
+        self.assertIsNone(context)
+
+    def test_execution_evidence_bounds_and_malformed_structures_fail_closed(self):
+        from utils import pair_execution as evidence
+        from unittest.mock import patch
+
+        kwargs = dict(
+            text="cat",
+            negative_text="blur",
+            positive_text="cat LORA",
+            negative_effective_text="blur",
+        )
+        evidence.record_pair_execution(None, "1", **kwargs)
+        envelope = {}
+        evidence.record_pair_execution(envelope, None, **kwargs)
+        self.assertEqual(envelope, {})
+        evidence.record_pair_execution(envelope, "1", **kwargs)
+        self.assertIsNone(evidence.read_pair_execution(envelope, "untraced"))
+        malformed = (
+            {"version": 1, "nodes": []},
+            {"version": 1, "truncated": True, "nodes": {}},
+            {"version": 1, "nodes": {"1": []}},
+            {"version": 1, "nodes": {"1": {"records": []}}},
+            {"version": 1, "nodes": {"1": {"records": "bad"}}},
+            {
+                "version": 1,
+                "nodes": {
+                    "1": {
+                        "records": [
+                            {
+                                **kwargs,
+                                "prepend_text": "",
+                                "append_text": "",
+                                "negative_effective_text": "changed",
+                            }
+                        ]
+                    }
+                },
+            },
+        )
+        for value in malformed:
+            with self.subTest(value=value):
+                with self.assertRaises(evidence.ExecutionEvidenceError):
+                    evidence.read_pair_execution({KEY: value}, "1")
+        oversized = {}
+        evidence.record_pair_execution(
+            oversized,
+            "1",
+            **{**kwargs, "text": "x" * (evidence.MAX_EXECUTION_TEXT + 1)},
+        )
+        with self.assertRaises(evidence.ExecutionEvidenceError):
+            evidence.read_pair_execution(oversized, "1")
+        with patch.object(evidence, "MAX_EXECUTION_JSON", 10):
+            with self.assertRaises(evidence.ExecutionEvidenceError):
+                evidence.read_pair_execution({KEY: json.dumps(envelope[KEY])}, "1")
+            limited = {}
+            evidence.record_pair_execution(limited, "1", **kwargs)
+            with self.assertRaises(evidence.ExecutionEvidenceError):
+                evidence.read_pair_execution(limited, "1")
+        with patch.object(evidence, "MAX_EXECUTION_NODES", 1):
+            evidence.record_pair_execution(envelope, "2", **kwargs)
+            evidence.record_pair_execution(envelope, "3", **kwargs)
+            with self.assertRaises(evidence.ExecutionEvidenceError):
+                evidence.read_pair_execution(envelope, "1")
+
+    def test_corrupted_existing_namespace_cannot_break_valid_encoding(self):
+        from utils.pair_execution import (
+            record_pair_execution,
+            read_pair_execution,
+            ExecutionEvidenceError,
+        )
+
+        kwargs = dict(
+            text="cat",
+            negative_text="blur",
+            positive_text="cat LORA",
+            negative_effective_text="blur",
+        )
+        for nodes in ([], {"1": {"records": "bad"}}, {"other": {"bad": object()}}):
+            extra = {KEY: {"version": 1, "nodes": nodes}}
+            record_pair_execution(extra, "1", **kwargs)
+            with self.assertRaises(ExecutionEvidenceError):
+                read_pair_execution(extra, "1")

utils/image_monitor.py

@@ -386,6 +386,29 @@ class ImageGenerationHandler(FileSystemEventHandler):
             return "unknown", None
 
         prompt_text = None
+        resolution = resolve_prompt_pair(metadata)
+        graph = metadata.get("prompt")
+        if resolution["state"] == "resolved" and isinstance(graph, dict):
+            positive = resolution["positive"]
+            path = positive["provenance"].get("path", [])
+            paired_source = any(
+                isinstance(graph.get(item.get("node_id")), dict)
+                and graph[item["node_id"]].get("class_type") == "PromptManagerPair"
+                for item in path
+            )
+            if paired_source:
+                try:
+                    existing = self.db_manager.get_prompt_by_hash(
+                        generate_prompt_hash(positive["storage_text"])
+                    )
+                    if existing:
+                        return "linked", {
+                            "id": existing["id"],
+                            "text": existing["text"],
+                            "from_metadata": True,
+                        }
+                except Exception as error:
+                    self.logger.warning(f"Paired metadata lookup failed: {error}")
 
         # The executed graph: use the PromptManager node feeding the positive input
         prompt_data = metadata.get("prompt")

utils/metadata_extractor.py

@@ -116,6 +116,7 @@ class ComfyUIMetadataExtractor:
 
                     # Extract other common metadata fields
                     metadata_fields = [
+                        "comfyassets_prompt_pairs",
                         "parameters",
                         "model",
                         "sampler",

utils/pair_execution.py

@@ -0,0 +1,145 @@
+"""Bounded per-node execution evidence embedded in ComfyUI PNG metadata."""
+
+import json
+from typing import Any, Dict, Optional
+
+EXECUTION_KEY = "comfyassets_prompt_pairs"
+MAX_EXECUTION_RECORDS = 16
+MAX_EXECUTION_NODES = 64
+MAX_EXECUTION_TEXT = 100_000
+MAX_EXECUTION_JSON = 500_000
+_FIELDS = (
+    "text",
+    "negative_text",
+    "positive_text",
+    "negative_effective_text",
+    "prepend_text",
+    "append_text",
+)
+
+
+class ExecutionEvidenceError(ValueError):
+    """Execution metadata is malformed, stale, truncated or ambiguous."""
+
+
+def record_pair_execution(
+    extra_pnginfo: Any,
+    node_id: Any,
+    *,
+    text: str,
+    negative_text: str,
+    positive_text: str,
+    negative_effective_text: str,
+    prepend_text: str = "",
+    append_text: str = "",
+) -> None:
+    """Append evidence without changing workflow or sibling extension metadata."""
+    if not isinstance(extra_pnginfo, dict) or node_id is None:
+        return
+    envelope = extra_pnginfo.get(EXECUTION_KEY)
+    if not isinstance(envelope, dict) or envelope.get("version") != 1:
+        envelope = {"version": 1, "nodes": {}}
+        extra_pnginfo[EXECUTION_KEY] = envelope
+    if envelope.get("truncated"):
+        return
+    nodes = envelope.get("nodes")
+    if not isinstance(nodes, dict):
+        envelope["truncated"] = True
+        return
+    if str(node_id) not in nodes and len(nodes) >= MAX_EXECUTION_NODES:
+        envelope["truncated"] = True
+        return
+    node = nodes.setdefault(str(node_id), {"records": []})
+    if not isinstance(node, dict) or not isinstance(node.get("records"), list):
+        envelope["truncated"] = True
+        return
+    record = {
+        "text": text,
+        "negative_text": negative_text,
+        "positive_text": positive_text,
+        "negative_effective_text": negative_effective_text,
+        "prepend_text": prepend_text,
+        "append_text": append_text,
+    }
+    if any(
+        not isinstance(value, str) or len(value) > MAX_EXECUTION_TEXT
+        for value in record.values()
+    ):
+        node["truncated"] = True
+        return
+    if record not in node["records"]:
+        if len(node["records"]) >= MAX_EXECUTION_RECORDS:
+            node["truncated"] = True
+            return
+        node["records"].append(record)
+    try:
+        if len(json.dumps(envelope, ensure_ascii=False)) > MAX_EXECUTION_JSON:
+            extra_pnginfo[EXECUTION_KEY] = {
+                "version": 1,
+                "truncated": True,
+                "nodes": {},
+            }
+    except (TypeError, ValueError, RecursionError):
+        extra_pnginfo[EXECUTION_KEY] = {"version": 1, "truncated": True, "nodes": {}}
+
+
+def read_pair_execution(metadata: Any, node_id: str) -> Optional[Dict[str, str]]:
+    """Return one distinct record for a traced node; missing evidence stays absent."""
+    if not isinstance(metadata, dict) or EXECUTION_KEY not in metadata:
+        return None
+    envelope = metadata[EXECUTION_KEY]
+    if isinstance(envelope, str):
+        if len(envelope) > MAX_EXECUTION_JSON:
+            raise ExecutionEvidenceError("Execution evidence exceeds the size limit")
+        try:
+            envelope = json.loads(envelope)
+        except (ValueError, TypeError, RecursionError) as error:
+            raise ExecutionEvidenceError("Malformed execution evidence JSON") from error
+    if not isinstance(envelope, dict) or envelope.get("version") != 1:
+        raise ExecutionEvidenceError("Unsupported execution evidence format")
+    nodes = envelope.get("nodes")
+    if (
+        envelope.get("truncated")
+        or not isinstance(nodes, dict)
+        or len(nodes) > MAX_EXECUTION_NODES
+    ):
+        raise ExecutionEvidenceError("Truncated or malformed execution evidence")
+    if node_id not in nodes:
+        return None
+    node = nodes[node_id]
+    if not isinstance(node, dict) or node.get("truncated"):
+        raise ExecutionEvidenceError("Truncated or malformed node execution evidence")
+    records = node.get("records")
+    if (
+        not isinstance(records, list)
+        or not records
+        or len(records) > MAX_EXECUTION_RECORDS
+    ):
+        raise ExecutionEvidenceError("Malformed node execution records")
+    unique = {}
+    for record in records:
+        if not isinstance(record, dict) or any(
+            not isinstance(record.get(key), str)
+            or len(record[key]) > MAX_EXECUTION_TEXT
+            for key in _FIELDS
+        ):
+            raise ExecutionEvidenceError("Malformed execution text values")
+        clean = {key: record[key] for key in _FIELDS}
+        if clean["negative_effective_text"] != clean["negative_text"]:
+            raise ExecutionEvidenceError("Negative execution text must be unchanged")
+        unique[json.dumps(clean, sort_keys=True)] = clean
+    if len(unique) != 1:
+        raise ExecutionEvidenceError(
+            "Multiple distinct executions have no image anchor"
+        )
+    return next(iter(unique.values()))
+
+
+def clear_pair_execution(request: Any) -> None:
+    """Discard client-carried execution records before a new validated run."""
+    if not isinstance(request, dict):
+        return
+    extra = request.get("extra_data")
+    png = extra.get("extra_pnginfo") if isinstance(extra, dict) else None
+    if isinstance(png, dict):
+        png.pop(EXECUTION_KEY, None)

utils/prompt_pair_resolver.py

@@ -10,8 +10,9 @@ ambiguous sampler selection. Unknown conditioning semantics are never flattened.
 
 Future nodes can register declarative ``OutputAdapter`` entries by
 ``(class_type, output_slot)``. This does not prescribe a future paired-node schema.
-Effective text is reconstructed from recorded graph inputs; runtime-only trigger
-injection cannot be recovered unless it was recorded. The resolver has no
+Effective text uses a unique bounded execution record for traced paired nodes
+when available; cached graph-only results remain explicitly inferred.
+The resolver has no
 ComfyUI, database, configuration or filesystem dependencies.
 """
 
@@ -19,6 +20,8 @@ import re
 from dataclasses import dataclass
 from typing import Any, Dict, Mapping, Optional, Tuple
 
+from .pair_execution import ExecutionEvidenceError, read_pair_execution
+
 
 @dataclass(frozen=True)
 class OutputAdapter:
@@ -101,6 +104,7 @@ class _GraphResolver:
         max_visits: int,
         max_depth: int,
         max_text_length: int,
+        metadata: Optional[dict] = None,
     ) -> None:
         self.graph = graph
         self.adapters = {**_DEFAULT_ADAPTERS, **adapters}
@@ -109,6 +113,8 @@ class _GraphResolver:
         self.max_text_length = max_text_length
         self.memo: dict = {}
         self.active: set = set()
+        self.metadata = metadata or {}
+        self.text_sources: dict = {}
 
     def node(self, node_id: str) -> Tuple[str, dict]:
         node = self.graph.get(node_id)
@@ -151,26 +157,43 @@ class _GraphResolver:
                 adapter = self.adapters[(name, slot)]
                 if not isinstance(adapter, OutputAdapter) or adapter.kind != kind:
                     raise _Unresolved("Output slot has an incompatible type")
-                text, path, storage_text = self.text(
-                    inputs.get(adapter.input_name), "string", depth + 1
+                evidence = (
+                    self.execution_record(node_id, inputs)
+                    if name == "PromptManagerPair"
+                    else None
                 )
-                if adapter.compose or adapter.storage_input:
-                    storage_text = text
-                if adapter.compose:
-                    before, before_path, _ = self.text(
-                        inputs.get("prepend_text", ""), "string", depth + 1
-                    )
-                    after, after_path, _ = self.text(
-                        inputs.get("append_text", ""), "string", depth + 1
-                    )
-                    text = self.bounded(
-                        " ".join(
-                            part
-                            for part in (before.strip(), text, after.strip())
-                            if part
+                if evidence is not None:
+                    text = evidence[
+                        (
+                            "positive_text"
+                            if adapter.input_name == "text"
+                            else "negative_effective_text"
                         )
+                    ]
+                    storage_text = evidence[adapter.input_name]
+                    path = []
+                    self.text_sources[(node_id, slot)] = "execution"
+                else:
+                    text, path, storage_text = self.text(
+                        inputs.get(adapter.input_name), "string", depth + 1
                     )
-                    path = before_path + path + after_path
+                    if adapter.compose or adapter.storage_input:
+                        storage_text = text
+                    if adapter.compose:
+                        before, before_path, _ = self.text(
+                            inputs.get("prepend_text", ""), "string", depth + 1
+                        )
+                        after, after_path, _ = self.text(
+                            inputs.get("append_text", ""), "string", depth + 1
+                        )
+                        text = self.bounded(
+                            " ".join(
+                                part
+                                for part in (before.strip(), text, after.strip())
+                                if part
+                            )
+                        )
+                        path = before_path + path + after_path
             elif (
                 name in ("StringConcatenate", "Text Concatenate")
                 and slot == 0
@@ -216,6 +239,44 @@ class _GraphResolver:
         finally:
             self.active.discard(key)
 
+    def execution_record(self, node_id: str, inputs: dict) -> Optional[dict]:
+        try:
+            record = read_pair_execution(self.metadata, node_id)
+        except ExecutionEvidenceError as error:
+            raise _Unresolved(str(error)) from error
+        if record is None:
+            return None
+        for name in ("text", "negative_text", "prepend_text", "append_text"):
+            try:
+                expected = self.text(inputs.get(name, ""), "string")[0]
+            except _Unresolved as error:
+                if str(error) == "Unsupported node semantics or output slot":
+                    value = inputs.get(name)
+                    if _link(value):
+                        source_name, _ = self.node(str(value[0]))
+                        known_slots = {
+                            slot
+                            for (kind, slot) in self.adapters
+                            if kind == source_name
+                        }
+                        if known_slots and value[1] not in known_slots:
+                            raise
+                        if (
+                            source_name
+                            in ("StringConcatenate", "Text Concatenate", "Reroute")
+                            and value[1] != 0
+                        ):
+                            raise
+                    continue
+                raise
+            if name in ("prepend_text", "append_text"):
+                expected = expected.strip()
+            if expected != record[name]:
+                raise _Unresolved(
+                    "Execution evidence does not match recorded node inputs"
+                )
+        return record
+
     def side(self, value: Any, sampler_id: str) -> dict:
         text, path, storage_text = self.text(value, "conditioning")
         origin = path[0]
@@ -227,6 +288,9 @@ class _GraphResolver:
                 "sampler_id": sampler_id,
                 **origin,
                 "path": path,
+                "text_source": self.text_sources.get(
+                    (origin["node_id"], origin["output_slot"]), "graph-inferred"
+                ),
             },
         }
 
@@ -310,7 +374,11 @@ def _parameters(value: Any, max_text_length: int) -> dict:
     positive = positive.strip()
     if not positive:
         return _result("unresolved", reason="Positive prompt is empty")
-    provenance = {"format": "a1111", "field": "parameters"}
+    provenance = {
+        "format": "a1111",
+        "field": "parameters",
+        "text_source": "embedded-parameters",
+    }
     positive_side = {
         "text": positive,
         "storage_text": positive,
@@ -359,7 +427,12 @@ def resolve_prompt_pair(
     if len(normalized) != len(graph):
         return _result("unresolved", reason="Graph node IDs are not unique")
     resolver = _GraphResolver(
-        normalized, adapters or {}, max_visits, max_depth, max_text_length
+        normalized,
+        adapters or {},
+        max_visits,
+        max_depth,
+        max_text_length,
+        metadata if "prompt" in metadata else None,
     )
     if output_node_id is not None:
         try:

utils/usage_tracking.py

@@ -31,11 +31,13 @@ try:
     from .logging_config import get_logger
     from .prompt_graph import intrinsic_pair_nodes, resolve_text, run_prompt_nodes
     from .prompt_pair_resolver import resolve_text_input
+    from .pair_execution import clear_pair_execution
 except ImportError:
     from utils.hashing import generate_prompt_hash
     from utils.logging_config import get_logger
     from utils.prompt_graph import intrinsic_pair_nodes, resolve_text, run_prompt_nodes
     from utils.prompt_pair_resolver import resolve_text_input
+    from utils.pair_execution import clear_pair_execution
 
 logger = get_logger("prompt_manager.usage_tracking")
 
@@ -217,6 +219,7 @@ def handle_queued_prompt(
     the request has not been validated yet.
     """
     try:
+        clear_pair_execution(json_data)
         graph = json_data.get("prompt") if isinstance(json_data, dict) else None
         runs = queued_prompt_runs(graph)
         pair_runs = queued_pair_runs(graph)
@@ -262,6 +265,7 @@ def count_queued_prompt(
 ) -> Any:
     """Legacy on-prompt handler body: count at submit time (no prompt_queue)."""
     try:
+        clear_pair_execution(json_data)
         graph = json_data.get("prompt") if isinstance(json_data, dict) else None
         runs = queued_prompt_runs(graph)
         pair_runs = queued_pair_runs(graph)