From 2fcbf5389d7273bbecf3051448d89ad46c33c1dc Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 17 Oct 2026 19:03:56 -0700 Subject: [PATCH] fix(history): scope transforms to captured source paths --- tests/test_execution_context_capture.py | 28 ++++++++++++++ tests/test_execution_context_graph.py | 41 ++++++++++++++++++++ tests/test_execution_context_host.py | 50 +++++++++++++++++++++++++ utils/execution_context_graph.py | 2 +- 4 files changed, 120 insertions(+), 1 deletion(-) diff --git a/tests/test_execution_context_capture.py b/tests/test_execution_context_capture.py index 01a3049..db0214d 100644 --- a/tests/test_execution_context_capture.py +++ b/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") diff --git a/tests/test_execution_context_graph.py b/tests/test_execution_context_graph.py index 6759ac2..e1bdc09 100644 --- a/tests/test_execution_context_graph.py +++ b/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): diff --git a/tests/test_execution_context_host.py b/tests/test_execution_context_host.py index e65aae1..7e61729 100644 --- a/tests/test_execution_context_host.py +++ b/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", diff --git a/utils/execution_context_graph.py b/utils/execution_context_graph.py index 985a145..bff0ea0 100644 --- a/utils/execution_context_graph.py +++ b/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(