diff --git a/tests/test_execution_context_graph.py b/tests/test_execution_context_graph.py index 8f6310a..e9dbdf4 100644 --- a/tests/test_execution_context_graph.py +++ b/tests/test_execution_context_graph.py @@ -117,6 +117,64 @@ class ProjectionTests(unittest.TestCase): g["s" * 129] = g.pop("s") self.assertFalse(self.m.project_contexts(g, None).recipes) + def test_dynamic_pm_inputs_preserve_observable_sources(self): + for kind, positive_slot, negative_slot in ( + ("PromptManagerText", 0, 0), + ("PromptManagerPair", 2, 3), + ("PromptManager", 1, 1), + ("PromptManagerFocused", 1, 1), + ("PromptManagerSplit", 1, 1), + ("PromptManagerPair", 0, 1), + ): + with self.subTest(kind=kind, slots=(positive_slot, negative_slot)): + g = graph(source=kind) + g["generator"] = node("UnknownGenerator") + g["p"]["inputs"]["text"] = ["generator", 0] + g["p"]["inputs"]["negative_text"] = ["generator", 0] + g["pos"] = node("CLIPTextEncode", text=["p", positive_slot]) + g["neg"] = node("CLIPTextEncode", text=["p", negative_slot]) + g["s"]["inputs"] = {"positive": ["pos", 0], "negative": ["neg", 0]} + if kind == "PromptManagerPair" and positive_slot == 0: + g["s"]["inputs"] = {"positive": ["p", 1], "negative": ["p", 0]} + positive_slot, negative_slot = 1, 0 + result = self.m.project_contexts(g, None) + recipe = result.recipes[0] + for role, slot in ( + (recipe.positive, positive_slot), + (recipe.negative, negative_slot), + ): + self.assertEqual( + tuple(ref.output_slot for ref in role.sources), (slot,) + ) + self.assertIsNone(role.inferred) + self.assertTrue(role.reason) + self.assertTrue(result.sources) + + def test_graph_evidence_distinguishes_literals_from_semantics(self): + result = self.m.project_contexts(graph(), None) + evidence = result.recipes[0].positive.inferred + self.assertEqual(evidence.raw.provenance, "inferred") + self.assertEqual(evidence.raw.source, "graph_literal") + self.assertEqual(evidence.effective.source, "graph_semantics") + + def test_derived_raw_and_composition_preserve_field_sources(self): + g = graph(source="PromptManagerText") + g["join"] = node( + "Text Concatenate", + text_a=" a ", + text_b=" b ", + delimiter="-", + clean_whitespace="true", + ) + g["p"]["inputs"]["text"] = ["join", 0] + g["clip"] = node("CLIPTextEncode", text=["p", 0]) + g["s"]["inputs"] = {"positive": ["clip", 0], "negative": ["clip", 0]} + role = self.m.project_contexts(g, None).recipes[0].positive + self.assertIsNotNone(role.inferred) + self.assertEqual(role.inferred.raw.value, "a-b") + self.assertEqual(role.inferred.raw.source, "graph_semantics") + self.assertEqual(role.inferred.prepend.source, "graph_literal") + def test_selected_custom_outputs_and_multistage_ancestry(self): g = graph() g["earlier"] = node("SamplerCustom", positive=["p", 0], negative=["p", 1]) @@ -157,7 +215,7 @@ class ProjectionTests(unittest.TestCase): role = self.m.project_contexts(g, None).recipes[0].positive self.assertIsNotNone(role.inferred) self.assertEqual(role.inferred.effective.value, "pre cat post!") - self.assertEqual(role.inferred.effective.provenance, "graph-inferred") + self.assertEqual(role.inferred.effective.provenance, "inferred") self.assertEqual(role.sources[0].node_id, "p") self.assertEqual(role.reason, "source-transformed") diff --git a/tests/test_execution_contexts.py b/tests/test_execution_contexts.py index 5f48626..ea55022 100644 --- a/tests/test_execution_contexts.py +++ b/tests/test_execution_contexts.py @@ -29,8 +29,17 @@ class ExactContextsTests(unittest.TestCase): def setUp(self): self.m = load_module("execution_contexts") - def evidence(self, value, provenance="graph-inferred"): - return self.m.TextEvidence(value, provenance, "node", None) + def evidence(self, value, provenance="inferred"): + return self.m.TextEvidence( + value, + "unknown" if value is None else provenance, + ( + "unknown" + if value is None + else "runtime_return" if provenance == "recorded" else "graph_literal" + ), + "unavailable" if value is None else None, + ) def snapshot(self, value): return self.m.PromptSnapshot( @@ -52,12 +61,12 @@ class ExactContextsTests(unittest.TestCase): ) a = self.snapshot("cat") b = m.PromptSnapshot( - *(self.evidence(v, "execution") for v in ("cat", "cat", "", "")) + *(self.evidence(v, "recorded") for v in ("cat", "cat", "", "")) ) self.assertEqual(m.snapshot_digest(a), m.snapshot_digest(b)) self.assertNotEqual( - m.context_digest(m.RecipeContext("1", a, None, "declared")), - m.context_digest(m.RecipeContext("1", None, a, "declared")), + m.context_digest(m.RecipeContext("1", a, None, "inferred")), + m.context_digest(m.RecipeContext("1", None, a, "inferred")), ) def test_invalid_text_and_bounds(self): @@ -71,8 +80,8 @@ class ExactContextsTests(unittest.TestCase): with self.assertRaises(ValueError): m.validate_bundle( m.ContextBundle( - (m.RecipeContext(identifier, None, None, "declared"),), - "unknown", + (m.RecipeContext(identifier, None, None, "inferred"),), + "unavailable", "missing", ) ) @@ -80,7 +89,7 @@ class ExactContextsTests(unittest.TestCase): m.validate_bundle( m.ContextBundle( tuple( - m.RecipeContext(str(i), None, None, "declared") + m.RecipeContext(str(i), None, None, "inferred") for i in range(17) ), "partial", @@ -90,7 +99,7 @@ class ExactContextsTests(unittest.TestCase): with self.assertRaises(ValueError): m.validate_bundle( m.ContextBundle( - (m.RecipeContext("1", self.snapshot("\ud800"), None, "declared"),), + (m.RecipeContext("1", self.snapshot("\ud800"), None, "inferred"),), "complete", "", ) @@ -99,7 +108,7 @@ class ExactContextsTests(unittest.TestCase): def test_bundle_preserves_unknown_and_empty(self): m = self.m bundle = m.ContextBundle( - (m.RecipeContext("1", self.snapshot(""), self.snapshot(None), "declared"),), + (m.RecipeContext("1", self.snapshot(""), self.snapshot(None), "inferred"),), "partial", "unknown", ) @@ -109,17 +118,77 @@ class ExactContextsTests(unittest.TestCase): m.ContextBundle(bundle.contexts, "partial", "", format_version=2) ) + def test_evidence_contract_rejects_incompatible_metadata(self): + m = self.m + invalid = ( + ("", "graph-inferred", "graph_literal", None), + ("", "recorded", "graph_literal", None), + ("", "inferred", "runtime_return", None), + (None, "inferred", "graph_literal", None), + ("", "unknown", "unknown", "unavailable"), + (None, "unknown", "unknown", None), + (None, "unknown", "unknown", "an arbitrary exception message"), + ) + for fields in invalid: + with self.subTest(fields=fields): + evidence = m.TextEvidence(*fields) + snapshot = m.PromptSnapshot(evidence, evidence, evidence, evidence) + with self.assertRaises(ValueError): + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext("1", snapshot, None, "inferred"),), + "partial", + "", + ) + ) + for provenance, source in ( + ("recorded", "runtime_return"), + ("recorded", "host_cache_output"), + ("inferred", "graph_literal"), + ("inferred", "graph_semantics"), + ): + evidence = m.TextEvidence("", provenance, source, None) + snapshot = m.PromptSnapshot(evidence, evidence, evidence, evidence) + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext("1", snapshot, None, "inferred"),), "complete", "" + ) + ) + + def test_bundle_state_and_declared_role_contract(self): + m = self.m + for state, role in ( + ("unknown", "inferred"), + ("complete", "recorded"), + ("complete", "declared"), + ): + with self.subTest(state=state, role=role): + with self.assertRaises(ValueError): + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext("1", None, None, role),), state, "" + ) + ) + for state in ("complete", "partial", "ambiguous", "unavailable", "conflicting"): + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext("1", None, None, "inferred"),), state, "" + ) + ) + def test_identifier_limit_counts_characters(self): m = self.m m.validate_bundle( m.ContextBundle( - (m.RecipeContext("é" * 128, None, None, "declared"),), "unknown", "" + (m.RecipeContext("é" * 128, None, None, "inferred"),), "unavailable", "" ) ) with self.assertRaises(ValueError): m.validate_bundle( m.ContextBundle( - (m.RecipeContext("é" * 129, None, None, "declared"),), "unknown", "" + (m.RecipeContext("é" * 129, None, None, "inferred"),), + "unavailable", + "", ) ) @@ -129,11 +198,11 @@ class ExactContextsTests(unittest.TestCase): with self.assertRaises(ValueError): m.validate_bundle( m.ContextBundle( - (m.RecipeContext("1", large, large, "declared"),), "complete", "" + (m.RecipeContext("1", large, large, "inferred"),), "complete", "" ) ) contexts = tuple( - m.RecipeContext(str(i), large, None, "declared") for i in range(8) + m.RecipeContext(str(i), large, None, "inferred") for i in range(8) ) with self.assertRaises(ValueError): m.validate_bundle(m.ContextBundle(contexts, "complete", "")) @@ -143,7 +212,7 @@ class ExactContextsTests(unittest.TestCase): # remain separate invalid records even if their prompt values are equal. m = self.m context = m.RecipeContext( - "1", self.snapshot("same"), self.snapshot("same"), "declared" + "1", self.snapshot("same"), self.snapshot("same"), "inferred" ) with self.assertRaises(ValueError): m.validate_bundle(m.ContextBundle((context, context), "complete", "")) diff --git a/utils/execution_context_graph.py b/utils/execution_context_graph.py index 93e18b4..4cc561f 100644 --- a/utils/execution_context_graph.py +++ b/utils/execution_context_graph.py @@ -51,16 +51,16 @@ class _Unknown(ValueError): def _identifier(value: object) -> str: if type(value) is int: if abs(value) > 10**127: - raise _Unknown("Invalid node identifier") + raise _Unknown("invalid_node_identifier") value = str(value) try: text_size(value, 512) if len(value) > 128: - raise ValueError("Oversized node identifier") + raise ValueError("oversized_node_identifier") except ValueError: - raise _Unknown("Invalid node identifier") from None + raise _Unknown("invalid_node_identifier") from None if not value: - raise _Unknown("Invalid node identifier") + raise _Unknown("invalid_node_identifier") return value @@ -71,7 +71,7 @@ def _link(value: object) -> tuple[str, int]: or type(value[1]) is not int or value[1] < 0 ): - raise _Unknown("Expected typed graph link") + raise _Unknown("expected_typed_graph_link") return _identifier(value[0]), value[1] @@ -85,17 +85,17 @@ class _Projector: def node(self, node_id: str, depth: int) -> tuple[str, dict]: self.visits += 1 if self.visits > 500 or depth > 64: - raise _Unknown("Graph resolution limit reached") + raise _Unknown("graph_resolution_limit_reached") node = self.graph.get(node_id) if type(node) is not dict or type(node.get("inputs")) is not dict: - raise _Unknown("Missing or malformed node") + raise _Unknown("missing_or_malformed_node") if len(node["inputs"]) > 500: - raise _Unknown("Node input limit reached") + raise _Unknown("node_input_limit_reached") kind = node.get("class_type") try: text_size(kind, 128) except ValueError: - raise _Unknown("Invalid node type") from None + raise _Unknown("invalid_node_type") from None return kind, node["inputs"] def text( @@ -107,13 +107,13 @@ class _Projector: node_id, slot = _link(value) key = (node_id, slot) if key in active: - raise _Unknown("Cyclic prompt graph") + raise _Unknown("cyclic_prompt_graph") name, inputs = self.node(node_id, depth) active = active | {key} if name == "Reroute" and slot == 0: keys = [key for key in ("value", "input", "in") if key in inputs] if len(keys) != 1: - raise _Unknown("Ambiguous reroute") + raise _Unknown("ambiguous_reroute") return self.text(inputs[keys[0]], kind, depth + 1, active) if ( name in ("StringConcatenate", "Text Concatenate") @@ -135,14 +135,14 @@ class _Projector: text_size(delimiter) clean_value = inputs.get("clean_whitespace", "true") if name == "Text Concatenate" and type(clean_value) not in (str, bool): - raise _Unknown("Invalid whitespace mode") + raise _Unknown("invalid_whitespace_mode") clean = name == "Text Concatenate" and ( clean_value is True or type(clean_value) is str and clean_value.lower() == "true" ) if not keys: - raise _Unknown("Missing concatenation inputs") + raise _Unknown("missing_concatenation_inputs") parts, sources, size = [], [], 0 for input_name in keys: child, refs = self.text( @@ -152,17 +152,19 @@ class _Projector: if part or name == "StringConcatenate": size += text_size(part) + (text_size(delimiter) if parts else 0) if size > 65536: - raise _Unknown("Text field limit reached") + raise _Unknown("text_field_limit_reached") parts.append(part) sources.extend(refs) if len(sources) > 64: - raise _Unknown("Source reference limit reached") + raise _Unknown("source_reference_limit_reached") self.transformed = self.transformed or bool(sources) text = delimiter.join(parts) - return self.snapshot(text, text, "", ""), tuple(sources) + return self.snapshot( + text, text, "", "", composed=True, raw_composed=True + ), tuple(sources) adapter = _DEFAULT_ADAPTERS.get((name, slot)) if adapter is None or adapter.kind != kind: - raise _Unknown("Unsupported node semantics or output slot") + raise _Unknown("unsupported_node_semantics_or_output_slot") child, sources = self.text( inputs.get(adapter.input_name), "string", depth + 1, active ) @@ -181,38 +183,118 @@ class _Projector: part for part in (prepend.strip(), raw, append.strip()) if part ) text_size(effective) - snapshot = self.snapshot(raw, effective, prepend, append) + snapshot = PromptSnapshot( + child.effective, + TextEvidence(effective, "inferred", "graph_semantics", None), + before.effective, + after.effective, + ) if name not in _PM: sources += before_sources + after_sources elif name in _PM: - snapshot = self.snapshot( - child.effective.value, child.effective.value, "", "" + snapshot = PromptSnapshot( + child.effective, + child.effective, + TextEvidence("", "inferred", "graph_literal", None), + TextEvidence("", "inferred", "graph_literal", None), ) else: snapshot = child if len(sources) > 64: - raise _Unknown("Source reference limit reached") + raise _Unknown("source_reference_limit_reached") return snapshot, sources + def discover_sources( + self, value: object, kind: str, depth: int = 1, active: frozenset = frozenset() + ) -> tuple[SourceRef, ...]: + if type(value) is str and kind == "string": + return () + node_id, slot = _link(value) + key = (node_id, slot) + if key in active: + raise _Unknown("cyclic_prompt_graph") + name, inputs = self.node(node_id, depth) + active = active | {key} + adapter = _DEFAULT_ADAPTERS.get((name, slot)) + if adapter is not None and adapter.kind == kind: + if name in _PM: + return (SourceRef(node_id, name, slot),) + return self.discover_sources( + inputs.get(adapter.input_name), "string", depth + 1, active + ) + if name == "Reroute" and slot == 0: + keys = [key for key in ("value", "input", "in") if key in inputs] + if len(keys) != 1: + raise _Unknown("ambiguous_reroute") + return self.discover_sources(inputs[keys[0]], kind, depth + 1, active) + if ( + name in ("StringConcatenate", "Text Concatenate") + and slot == 0 + and kind == "string" + ): + keys = ( + ["string_a", "string_b"] + if name == "StringConcatenate" + else sorted( + key + for key in inputs + if type(key) is str and re.fullmatch(r"text_[a-z]", key) + ) + ) + sources = [] + for key in keys: + sources.extend( + self.discover_sources( + inputs.get(key, ""), "string", depth + 1, active + ) + ) + if len(sources) > 64: + raise _Unknown("source_limit_reached") + self.transformed = self.transformed or bool(sources) + return tuple(sources) + return () + @staticmethod - def snapshot(raw: str, effective: str, prepend: str, append: str) -> PromptSnapshot: + def snapshot( + raw: str, + effective: str, + prepend: str, + append: str, + composed: bool = False, + raw_composed: bool = False, + ) -> PromptSnapshot: return PromptSnapshot( *( - TextEvidence(value, "graph-inferred", "graph", None) - for value in (raw, effective, prepend, append) + TextEvidence( + value, + "inferred", + ( + "graph_semantics" + if (composed and index == 1 or raw_composed and index == 0) + else "graph_literal" + ), + None, + ) + for index, value in enumerate((raw, effective, prepend, append)) ) ) def role(self, value: object) -> GraphRole: + sources = () try: self.transformed = False - snapshot, sources = self.text(value, "conditioning") + sources = self.discover_sources(value, "conditioning") + snapshot, _ = self.text(value, "conditioning") return GraphRole( sources, snapshot, "source-transformed" if self.transformed else "" ) except ValueError as error: - reason = str(error) if isinstance(error, _Unknown) else "Invalid text value" - return GraphRole((), None, reason) + reason = str(error) if isinstance(error, _Unknown) else "invalid_text_value" + return GraphRole( + sources, + None, + "source-transformed" if self.transformed and sources else reason, + ) def anchors(self, node_id: str, depth: int = 1) -> tuple[str, ...]: if node_id in self.selected_seen: @@ -226,7 +308,7 @@ class _Projector: parent, _ = _link(value) anchors.extend(self.anchors(parent, depth + 1)) if len(anchors) > 16: - raise _Unknown("Recipe context limit reached") + raise _Unknown("recipe_context_limit_reached") return tuple(anchors) @@ -238,10 +320,10 @@ def project_contexts(graph: object, outputs: object) -> GraphProjection: """ try: if type(graph) is not dict or len(graph) > 500: - raise _Unknown("Invalid or oversized graph") + raise _Unknown("invalid_or_oversized_graph") normalized = {_identifier(key): value for key, value in graph.items()} if len(normalized) != len(graph): - raise _Unknown("Duplicate graph identifiers") + raise _Unknown("duplicate_graph_identifiers") resolver = _Projector(normalized) if outputs is None: anchors = [ @@ -257,9 +339,9 @@ def project_contexts(graph: object, outputs: object) -> GraphProjection: anchors.extend(resolver.anchors(_identifier(output))) anchors = list(dict.fromkeys(anchors)) else: - raise _Unknown("Invalid output selection") + raise _Unknown("invalid_output_selection") if len(anchors) > 16: - raise _Unknown("Recipe context limit reached") + raise _Unknown("recipe_context_limit_reached") recipes, sources = [], [] for anchor in anchors: _, inputs = resolver.node(anchor, 1) @@ -269,15 +351,15 @@ def project_contexts(graph: object, outputs: object) -> GraphProjection: sources.extend(positive.sources + negative.sources) sources = tuple(dict.fromkeys(sources)) if len(sources) > 64: - raise _Unknown("Source reference limit reached") - reason = "" if recipes else "No selected sampler anchors" + raise _Unknown("source_reference_limit_reached") + reason = "" if recipes else "no_selected_sampler_anchors" if any( role.reason for recipe in recipes for role in (recipe.positive, recipe.negative) ): - reason = "Some roles require inferred or unresolved text" + reason = "some_roles_require_inferred_or_unresolved_text" return GraphProjection(tuple(recipes), sources, reason) except ValueError as error: - reason = str(error) if isinstance(error, _Unknown) else "Invalid text value" + reason = str(error) if isinstance(error, _Unknown) else "invalid_text_value" return GraphProjection((), (), reason) diff --git a/utils/execution_contexts.py b/utils/execution_contexts.py index 144c318..37eb35c 100644 --- a/utils/execution_contexts.py +++ b/utils/execution_contexts.py @@ -2,6 +2,7 @@ import hashlib import json +import re from dataclasses import dataclass @@ -93,6 +94,20 @@ def context_digest(context: RecipeContext) -> str: ) +BUNDLE_STATES = frozenset( + {"complete", "partial", "ambiguous", "unavailable", "conflicting"} +) + +UNKNOWN_REASONS = frozenset( + {"unavailable", "unsupported", "ambiguous", "limit_reached", "missing"} +) +EVIDENCE_SOURCES = { + "recorded": frozenset({"runtime_return", "host_cache_output"}), + "inferred": frozenset({"graph_literal", "graph_semantics"}), + "unknown": frozenset({"unknown"}), +} + + def validate_bundle(bundle: ContextBundle) -> None: if ( type(bundle) is not ContextBundle @@ -103,6 +118,10 @@ def validate_bundle(bundle: ContextBundle) -> None: ): raise ValueError("Invalid context bundle") total = text_size(bundle.state, 128) + text_size(bundle.reason, 128) + if bundle.state not in BUNDLE_STATES: + raise ValueError("Invalid bundle state") + if bundle.reason and re.fullmatch(r"[a-z][a-z0-9_-]*", bundle.reason) is None: + raise ValueError("Invalid bundle reason code") anchors = set() for context in bundle.contexts: if type(context) is not RecipeContext: @@ -114,6 +133,8 @@ def validate_bundle(bundle: ContextBundle) -> None: raise ValueError("Duplicate or empty recipe anchor") anchors.add(context.anchor_id) size += text_size(context.role_provenance, 128) + if context.role_provenance != "inferred": + raise ValueError("Invalid declared role provenance") for snapshot in (context.positive, context.negative): if snapshot is None: continue @@ -127,6 +148,21 @@ def validate_bundle(bundle: ContextBundle) -> None: ): if type(field) is not TextEvidence: raise ValueError("Invalid text evidence") + if ( + type(field.provenance) is not str + or type(field.source) is not str + or field.source not in EVIDENCE_SOURCES.get(field.provenance, ()) + ): + raise ValueError("Invalid evidence provenance or source") + if field.value is None: + if ( + field.provenance != "unknown" + or type(field.reason) is not str + or field.reason not in UNKNOWN_REASONS + ): + raise ValueError("Unknown evidence requires a reason code") + elif field.provenance == "unknown" or field.reason is not None: + raise ValueError("Known evidence requires a known provenance") if field.value is not None: size += text_size(field.value) size += text_size(field.provenance, 128) + text_size(field.source, 128)