fix(history): scope transforms to captured source paths

tests/test_execution_context_capture.py

@@ -10,6 +10,8 @@ import unittest
 import weakref
 from unittest.mock import patch
 
+from test_execution_context_graph import transformed_input_graph
+
 from utils.execution_context_budget import ContextBudget
 from utils.execution_context_graph import SourceRef
 from utils.execution_contexts import PromptSnapshot, TextEvidence
@@ -216,6 +218,32 @@ class CaptureTests(unittest.TestCase):
         self.assertIsNone(result.bundle.contexts[0].positive.effective.value)
         self.assertNotEqual(result.state, "complete")
 
+    def test_upstream_transform_preserves_final_pm_runtime_snapshot(self):
+        for field in ("text", "prepend_text", "append_text"):
+            for unresolved in (False, True):
+                with self.subTest(field=field, unresolved=unresolved):
+                    cap = self.begin(g=transformed_input_graph(field, unresolved))
+                    self.host_context()
+                    capture.record_runtime_context(
+                        "PromptManagerPair",
+                        " actual raw ",
+                        " actual pre ",
+                        " actual post ",
+                        " actual injected effective ",
+                        role="positive",
+                    )
+                    side = self.seal(cap).bundle.contexts[0].positive
+                    for name, expected in (
+                        ("raw", " actual raw "),
+                        ("prepend", " actual pre "),
+                        ("append", " actual post "),
+                        ("effective", " actual injected effective "),
+                    ):
+                        evidence = getattr(side, name)
+                        self.assertEqual(evidence.value, expected)
+                        self.assertEqual(evidence.provenance, "recorded")
+                        self.assertEqual(evidence.source, "runtime_return")
+
     def test_unsupported_graph_has_no_supported_context(self):
         result = self.seal(self.begin(g={}))
         self.assertEqual(result.reason, "no_supported_context")

tests/test_execution_context_graph.py

@@ -21,6 +21,24 @@ def graph(kind="KSampler", source="PromptManagerPair"):
     }
 
 
+def transformed_input_graph(field: str, unresolved: bool = False) -> dict:
+    g = graph()
+    g["upstream"] = node("PromptManagerText", text="looks literal")
+    g["dynamic"] = node("RuntimeGenerator")
+    g["join"] = node(
+        "StringConcatenate",
+        string_a=["upstream", 0],
+        string_b=["dynamic", 0] if unresolved else " suffix",
+    )
+    if unresolved:
+        g["inner"] = node(
+            "StringConcatenate", string_a=["upstream", 0], string_b=" inner"
+        )
+        g["join"]["inputs"]["string_a"] = ["inner", 0]
+    g["p"]["inputs"][field] = ["join", 0]
+    return g
+
+
 class ProjectionTests(unittest.TestCase):
     def setUp(self):
         self.m = load_module("execution_context_graph")
@@ -219,6 +237,29 @@ class ProjectionTests(unittest.TestCase):
         self.assertEqual(role.sources[0].node_id, "p")
         self.assertEqual(role.reason, "source-transformed")
 
+    def test_transform_provenance_stops_at_selected_pm_capture_boundary(self):
+        for field in ("text", "prepend_text", "append_text"):
+            for unresolved in (False, True):
+                with self.subTest(field=field, unresolved=unresolved):
+                    g = transformed_input_graph(field, unresolved)
+                    g["outer"] = node(
+                        "StringConcatenate", string_a=["p", 2], string_b=" outer"
+                    )
+                    g["clip"] = node("CLIPTextEncode", text=["outer", 0])
+                    g["s"]["inputs"]["negative"] = ["clip", 0]
+                    result = self.m.project_contexts(g, None)
+                    positive, negative = (
+                        result.recipes[0].positive,
+                        result.recipes[0].negative,
+                    )
+                    self.assertEqual(
+                        [(ref.node_id, ref.output_slot) for ref in positive.sources],
+                        [("p", 0)],
+                    )
+                    self.assertNotEqual(positive.reason, "source-transformed")
+                    self.assertEqual(negative.reason, "source-transformed")
+                    self.assertEqual(negative.sources[0].node_id, "p")
+
     def test_malformed_inputs_do_not_stringify(self):
         class HostValue:
             def __str__(self):

tests/test_execution_context_host.py

@@ -11,6 +11,8 @@ import types
 import unittest
 from unittest.mock import patch
 
+from test_execution_context_graph import transformed_input_graph
+
 from utils.execution_context_budget import ContextBudget
 from utils.execution_context_capture import ContextCapture
 from utils.execution_history_events import RunStart
@@ -247,6 +249,54 @@ class HostTests(unittest.TestCase):
                     else:
                         self.assertEqual(result.raw.value, " raw ")
 
+    def test_upstream_transform_preserves_final_pm_cached_effective_only(self):
+        for field, snapshot_field in (
+            ("text", "raw"),
+            ("prepend_text", "prepend"),
+            ("append_text", "append"),
+        ):
+            for unresolved in (False, True):
+                with self.subTest(field=field, unresolved=unresolved):
+                    g = transformed_input_graph(field, unresolved)
+                    g["outer"] = {
+                        "class_type": "StringConcatenate",
+                        "inputs": {"string_a": ["p", 2], "string_b": " outer"},
+                    }
+                    g["clip"] = {
+                        "class_type": "CLIPTextEncode",
+                        "inputs": {"text": ["outer", 0]},
+                    }
+                    g["s"]["inputs"]["negative"] = ["clip", 0]
+                    self.cap = self.capture.begin(
+                        RunStart(SESSION, SESSION, 1, JOB, EPOCH, 0),
+                        3,
+                        g,
+                        None,
+                    )
+                    self.executor.caches.outputs.entries["p"] = Entry(
+                        None, [object(), object(), [" cached injected "], [""]]
+                    )
+                    self.observe("execution_start")
+                    self.observe("execution_cached", nodes=["upstream", "p"])
+                    result = self.seal()
+                    self.assertIsNone(result.negative.effective.value)
+                    side = result.positive
+                    self.assertEqual(side.effective.value, " cached injected ")
+                    self.assertEqual(side.effective.provenance, "recorded")
+                    self.assertEqual(side.effective.source, "host_cache_output")
+                    for name, expected in (
+                        ("raw", " cat "),
+                        ("prepend", "pre"),
+                        ("append", "post"),
+                    ):
+                        evidence = getattr(side, name)
+                        if name == snapshot_field:
+                            self.assertIsNone(evidence.value)
+                            self.assertEqual(evidence.provenance, "unknown")
+                        else:
+                            self.assertEqual(evidence.value, expected)
+                            self.assertEqual(evidence.provenance, "inferred")
+
     def test_provider_local_reset_eviction_and_unknown_modes_fail_closed(self):
         for mode in (
             "NullCache",

utils/execution_context_graph.py

@@ -163,7 +163,6 @@ class _Projector:
                 sources.extend(refs)
                 if len(sources) > 64:
                     raise _Unknown("source_reference_limit_reached")
-            self.transformed = self.transformed or bool(sources)
             text = delimiter.join(parts)
             return self.snapshot(
                 text, text, "", "", composed=True, raw_composed=True
@@ -331,6 +330,7 @@ class _Projector:
         sources = ()
         try:
             self.transformed = False
+            # Only the capture-source-to-consumer path determines transforms.
             sources = self.discover_sources(value, "conditioning")
             snapshot, _ = self.text(value, "conditioning")
             return GraphRole(