diff --git a/tests/test_execution_context_graph.py b/tests/test_execution_context_graph.py new file mode 100644 index 0000000..8f6310a --- /dev/null +++ b/tests/test_execution_context_graph.py @@ -0,0 +1,174 @@ +"""Projection must retain declared roles without claiming sampler consumption.""" + +import unittest +from test_execution_contexts import load_module + + +def node(kind, **inputs): + return {"class_type": kind, "inputs": inputs} + + +def graph(kind="KSampler", source="PromptManagerPair"): + return { + "p": node( + source, + text=" cat ", + negative_text="", + prepend_text="pre", + append_text="post", + ), + "s": node(kind, positive=["p", 0], negative=["p", 1]), + } + + +class ProjectionTests(unittest.TestCase): + def setUp(self): + self.m = load_module("execution_context_graph") + + def test_projection_preserves_roles(self): + for kind in ("KSampler", "KSamplerAdvanced", "SamplerCustom"): + g = graph(kind) + result = self.m.project_contexts(g, None) + self.assertEqual(len(result.recipes), 1) + recipe = result.recipes[0] + self.assertEqual(recipe.anchor_id, "s") + self.assertEqual(recipe.positive.inferred.raw.value, " cat ") + self.assertEqual(recipe.positive.inferred.effective.value, "pre cat post") + self.assertEqual(recipe.negative.inferred.effective.value, "") + self.assertEqual(recipe.positive.sources[0].output_slot, 0) + g["s"]["inputs"]["positive"], g["s"]["inputs"]["negative"] = ["p", 1], [ + "p", + 0, + ] + crossed = self.m.project_contexts(g, None).recipes[0] + self.assertEqual(crossed.positive.inferred.raw.value, "") + self.assertEqual(crossed.negative.sources[0].output_slot, 0) + for source in ("PromptManager", "PromptManagerFocused", "PromptManagerSplit"): + g = graph(source=source) + g["clip"] = node("CLIPTextEncode", text=["p", 1]) + g["s"]["inputs"]["negative"] = ["clip", 0] + recipe = self.m.project_contexts(g, None).recipes[0] + self.assertEqual(recipe.negative.inferred.raw.value, " cat ") + self.assertEqual(recipe.negative.sources[0].output_slot, 1) + for source, slot in ( + ("PromptManagerText", 0), + ("PromptManagerPair", 2), + ("PromptManagerPair", 3), + ): + g = graph(source=source) + g["clip"] = node("CLIPTextEncode", text=["p", slot]) + g["s"]["inputs"] = {"positive": ["clip", 0], "negative": ["clip", 0]} + self.assertEqual( + self.m.project_contexts(g, None) + .recipes[0] + .positive.sources[0] + .output_slot, + slot, + ) + g = graph() + g["p"] = node("CLIPTextEncode", text=["literal", 0]) + g["literal"] = node("PrimitiveString", value=" exact ") + g["s"]["inputs"]["negative"] = ["p", 0] + self.assertEqual( + self.m.project_contexts(g, None) + .recipes[0] + .positive.inferred.effective.value, + " exact ", + ) + g["p"] = node("UnknownTransform", text="secret") + role = self.m.project_contexts(g, None).recipes[0].positive + self.assertIsNone(role.inferred) + self.assertTrue(role.reason) + + def test_output_selection_and_scalar_isolation(self): + g = graph() + g["other"] = node("KSampler", positive=["p", 0], negative=["p", 1]) + g["decode"] = node("VAEDecode", samples=["s", 0]) + g["save"] = node("SaveImage", images=["decode", 0]) + result = self.m.project_contexts(g, ["save"]) + self.assertEqual(tuple(r.anchor_id for r in result.recipes), ("s",)) + g["p"]["inputs"]["text"] = "changed" + self.assertEqual(result.recipes[0].positive.inferred.raw.value, " cat ") + self.assertEqual(self.m.project_contexts(g, []).recipes, ()) + for outputs in (["missing"], [object()], "save"): + result = self.m.project_contexts(g, outputs) + self.assertEqual(result.recipes, ()) + self.assertTrue(result.reason) + + def test_projection_limits(self): + for count, succeeds in ((500, True), (501, False)): + g = graph() + g.update({str(i): node("Unused") for i in range(count - 2)}) + self.assertEqual(bool(self.m.project_contexts(g, None).recipes), succeeds) + for count, succeeds in ((16, True), (17, False)): + g = {"p": graph()["p"]} + g.update({str(i): graph()["s"] for i in range(count)}) + self.assertEqual(bool(self.m.project_contexts(g, None).recipes), succeeds) + for depth, succeeds in ((64, True), (65, False)): + g = graph() + # The CLIP plus reroutes plus literal give exactly depth visited links. + g["p"] = node("CLIPTextEncode", text=["0", 0]) + for i in range(depth - 2): + g[str(i)] = node("Reroute", value=[str(i + 1), 0]) + g[str(depth - 2)] = node("PrimitiveString", value="known") + role = self.m.project_contexts(g, None).recipes[0].positive + self.assertEqual(role.inferred is not None, succeeds) + g = graph() + g["s" * 129] = g.pop("s") + self.assertFalse(self.m.project_contexts(g, None).recipes) + + def test_selected_custom_outputs_and_multistage_ancestry(self): + g = graph() + g["earlier"] = node("SamplerCustom", positive=["p", 0], negative=["p", 1]) + g["s"]["inputs"]["latent_image"] = ["earlier", 0] + g["adapter"] = node("UnknownImageAdapter", source=["s", 0]) + g["custom"] = node("CustomOutput", images=["adapter", 0]) + result = self.m.project_contexts(g, ["custom"]) + self.assertEqual(tuple(r.anchor_id for r in result.recipes), ("s", "earlier")) + + def test_total_resolution_visits_are_bounded(self): + # Sixteen recipes each resolving two independent chains of length 16 + # exceed 500 total visits even though the graph has fewer than 500 nodes. + g = {"v": node("PrimitiveString", value="known")} + for i in range(15): + g[str(i)] = node("Reroute", value=[str(i + 1), 0] if i < 14 else ["v", 0]) + g["clip"] = node("CLIPTextEncode", text=["0", 0]) + for i in range(16): + g["s" + str(i)] = node( + "KSampler", positive=["clip", 0], negative=["clip", 0] + ) + result = self.m.project_contexts(g, None) + self.assertTrue(result.reason) + self.assertFalse( + result.recipes + and all( + r.positive.inferred is not None and r.negative.inferred is not None + for r in result.recipes + ) + ) + + def test_known_concatenation_remains_inferred(self): + g = graph(source="PromptManagerText") + g["join"] = node( + "StringConcatenate", string_a=["p", 0], string_b="!", delimiter="" + ) + g["clip"] = node("CLIPTextEncode", text=["join", 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.effective.value, "pre cat post!") + self.assertEqual(role.inferred.effective.provenance, "graph-inferred") + self.assertEqual(role.sources[0].node_id, "p") + self.assertEqual(role.reason, "source-transformed") + + def test_malformed_inputs_do_not_stringify(self): + class HostValue: + def __str__(self): + raise AssertionError("host object stringified") + + for g in ( + {HostValue(): node("KSampler")}, + {"s": node(HostValue())}, + {"s": node("KSampler", positive=HostValue())}, + ): + self.m.project_contexts(g, None) diff --git a/tests/test_execution_contexts.py b/tests/test_execution_contexts.py new file mode 100644 index 0000000..5f48626 --- /dev/null +++ b/tests/test_execution_contexts.py @@ -0,0 +1,149 @@ +"""Exact identities and bounded scalar records; normalization would lose history.""" + +import hashlib +import importlib.util +import sys +import types +import unittest +from pathlib import Path + + +def load_module(name): + package = "execution_context_test_utils" + if package not in sys.modules: + module = types.ModuleType(package) + module.__path__ = [str(Path(__file__).parents[1] / "utils")] + sys.modules[package] = module + qualified = package + "." + name + if qualified not in sys.modules: + spec = importlib.util.spec_from_file_location( + qualified, Path(__file__).parents[1] / "utils" / (name + ".py") + ) + module = importlib.util.module_from_spec(spec) + sys.modules[qualified] = module + spec.loader.exec_module(module) + return sys.modules[qualified] + + +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 snapshot(self, value): + return self.m.PromptSnapshot( + *(self.evidence(v) for v in (value, value, "", "")) + ) + + def test_exact_identity(self): + m = self.m + self.assertEqual( + m.canonical_content("snapshot", ("é", None, "", "")), + '[1,"snapshot","é",null,"",""]'.encode(), + ) + values = ("cat", "Cat", " cat", "cat ", "é", "e\u0301", None, "") + digests = {m.content_digest("snapshot", (v, v, "", "")) for v in values} + self.assertEqual(len(digests), len(values)) + self.assertEqual( + m.content_digest("snapshot", ("cat", "cat", "", "")), + hashlib.sha256(b'[1,"snapshot","cat","cat","",""]').hexdigest(), + ) + a = self.snapshot("cat") + b = m.PromptSnapshot( + *(self.evidence(v, "execution") 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")), + ) + + def test_invalid_text_and_bounds(self): + m = self.m + for value in ("\ud800", "x" * 65537, "é" * 32769, object(), True): + with self.subTest(value_type=type(value).__name__): + with self.assertRaises(ValueError): + m.canonical_content("snapshot", (value,)) + m.canonical_content("snapshot", ("é" * 32768,)) + for identifier in ("x" * 129, object()): + with self.assertRaises(ValueError): + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext(identifier, None, None, "declared"),), + "unknown", + "missing", + ) + ) + with self.assertRaises(ValueError): + m.validate_bundle( + m.ContextBundle( + tuple( + m.RecipeContext(str(i), None, None, "declared") + for i in range(17) + ), + "partial", + "limit", + ) + ) + with self.assertRaises(ValueError): + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext("1", self.snapshot("\ud800"), None, "declared"),), + "complete", + "", + ) + ) + + def test_bundle_preserves_unknown_and_empty(self): + m = self.m + bundle = m.ContextBundle( + (m.RecipeContext("1", self.snapshot(""), self.snapshot(None), "declared"),), + "partial", + "unknown", + ) + self.assertIsNone(m.validate_bundle(bundle)) + with self.assertRaises(ValueError): + m.validate_bundle( + m.ContextBundle(bundle.contexts, "partial", "", format_version=2) + ) + + def test_identifier_limit_counts_characters(self): + m = self.m + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext("é" * 128, None, None, "declared"),), "unknown", "" + ) + ) + with self.assertRaises(ValueError): + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext("é" * 129, None, None, "declared"),), "unknown", "" + ) + ) + + def test_aggregate_payload_limits(self): + m = self.m + large = self.snapshot("x" * 65536) + with self.assertRaises(ValueError): + m.validate_bundle( + m.ContextBundle( + (m.RecipeContext("1", large, large, "declared"),), "complete", "" + ) + ) + contexts = tuple( + m.RecipeContext(str(i), large, None, "declared") for i in range(8) + ) + with self.assertRaises(ValueError): + m.validate_bundle(m.ContextBundle(contexts, "complete", "")) + + def test_equal_repeated_samples_are_ambiguous(self): + # Projection contains references, never runtime samples. Duplicate anchors + # 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" + ) + 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 new file mode 100644 index 0000000..93e18b4 --- /dev/null +++ b/utils/execution_context_graph.py @@ -0,0 +1,283 @@ +"""Bounded projection of declared sampler roles, independent of the host.""" + +import re +from dataclasses import dataclass +from .execution_contexts import PromptSnapshot, TextEvidence, text_size +from .prompt_pair_resolver import _DEFAULT_ADAPTERS + + +@dataclass(frozen=True) +class SourceRef: + node_id: str + node_type: str + output_slot: int + + +@dataclass(frozen=True) +class GraphRole: + sources: tuple[SourceRef, ...] + inferred: PromptSnapshot | None + reason: str + + +@dataclass(frozen=True) +class DeclaredRecipe: + anchor_id: str + positive: GraphRole + negative: GraphRole + + +@dataclass(frozen=True) +class GraphProjection: + recipes: tuple[DeclaredRecipe, ...] + sources: tuple[SourceRef, ...] + reason: str + + +_SAMPLERS = {"KSampler", "KSamplerAdvanced", "SamplerCustom"} +_PM = { + "PromptManager", + "PromptManagerFocused", + "PromptManagerSplit", + "PromptManagerText", + "PromptManagerPair", +} + + +class _Unknown(ValueError): + pass + + +def _identifier(value: object) -> str: + if type(value) is int: + if abs(value) > 10**127: + raise _Unknown("Invalid node identifier") + value = str(value) + try: + text_size(value, 512) + if len(value) > 128: + raise ValueError("Oversized node identifier") + except ValueError: + raise _Unknown("Invalid node identifier") from None + if not value: + raise _Unknown("Invalid node identifier") + return value + + +def _link(value: object) -> tuple[str, int]: + if ( + type(value) not in (tuple, list) + or len(value) != 2 + or type(value[1]) is not int + or value[1] < 0 + ): + raise _Unknown("Expected typed graph link") + return _identifier(value[0]), value[1] + + +class _Projector: + def __init__(self, graph: dict) -> None: + self.graph = graph + self.visits = 0 + self.transformed = False + self.selected_seen = set() + + 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") + 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") + if len(node["inputs"]) > 500: + 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 + return kind, node["inputs"] + + def text( + self, value: object, kind: str, depth: int = 1, active: frozenset = frozenset() + ) -> tuple[PromptSnapshot, tuple]: + if type(value) is str and kind == "string": + text_size(value) + return self.snapshot(value, value, "", ""), () + 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} + 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.text(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) + ) + ) + delimiter = inputs.get( + "delimiter", "" if name == "StringConcatenate" else " " + ) + 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") + 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") + parts, sources, size = [], [], 0 + for input_name in keys: + child, refs = self.text( + inputs.get(input_name, ""), "string", depth + 1, active + ) + part = child.effective.value.strip() if clean else child.effective.value + 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") + parts.append(part) + 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, "", ""), tuple(sources) + adapter = _DEFAULT_ADAPTERS.get((name, slot)) + if adapter is None or adapter.kind != kind: + raise _Unknown("Unsupported node semantics or output slot") + child, sources = self.text( + inputs.get(adapter.input_name), "string", depth + 1, active + ) + if name in _PM: + sources = (SourceRef(node_id, name, slot),) + if adapter.compose: + before, before_sources = self.text( + inputs.get("prepend_text", ""), "string", depth + 1, active + ) + after, after_sources = self.text( + inputs.get("append_text", ""), "string", depth + 1, active + ) + raw = child.effective.value + prepend, append = before.effective.value, after.effective.value + effective = " ".join( + part for part in (prepend.strip(), raw, append.strip()) if part + ) + text_size(effective) + snapshot = self.snapshot(raw, effective, prepend, append) + if name not in _PM: + sources += before_sources + after_sources + elif name in _PM: + snapshot = self.snapshot( + child.effective.value, child.effective.value, "", "" + ) + else: + snapshot = child + if len(sources) > 64: + raise _Unknown("Source reference limit reached") + return snapshot, sources + + @staticmethod + def snapshot(raw: str, effective: str, prepend: str, append: str) -> PromptSnapshot: + return PromptSnapshot( + *( + TextEvidence(value, "graph-inferred", "graph", None) + for value in (raw, effective, prepend, append) + ) + ) + + def role(self, value: object) -> GraphRole: + try: + self.transformed = False + snapshot, sources = 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) + + def anchors(self, node_id: str, depth: int = 1) -> tuple[str, ...]: + if node_id in self.selected_seen: + return () + self.selected_seen.add(node_id) + name, inputs = self.node(node_id, depth) + anchors = [node_id] if name in _SAMPLERS else [] + for value in inputs.values(): + if type(value) not in (tuple, list): + continue + parent, _ = _link(value) + anchors.extend(self.anchors(parent, depth + 1)) + if len(anchors) > 16: + raise _Unknown("Recipe context limit reached") + return tuple(anchors) + + +def project_contexts(graph: object, outputs: object) -> GraphProjection: + """Keep declared roles and graph inference, never evidence of consumption. + + Sources are PM capture points. An inferred snapshot may compose text; its + presence never promotes a source observation to recorded effective text. + """ + try: + if type(graph) is not dict or len(graph) > 500: + 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") + resolver = _Projector(normalized) + if outputs is None: + anchors = [ + key + for key, value in normalized.items() + if type(value) is dict + and type(value.get("class_type")) is str + and value["class_type"] in _SAMPLERS + ] + elif type(outputs) in (list, tuple) and len(outputs) <= 500: + anchors = [] + for output in outputs: + anchors.extend(resolver.anchors(_identifier(output))) + anchors = list(dict.fromkeys(anchors)) + else: + raise _Unknown("Invalid output selection") + if len(anchors) > 16: + raise _Unknown("Recipe context limit reached") + recipes, sources = [], [] + for anchor in anchors: + _, inputs = resolver.node(anchor, 1) + positive = resolver.role(inputs.get("positive")) + negative = resolver.role(inputs.get("negative")) + recipes.append(DeclaredRecipe(anchor, positive, negative)) + 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" + if any( + role.reason + for recipe in recipes + for role in (recipe.positive, recipe.negative) + ): + 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" + return GraphProjection((), (), reason) diff --git a/utils/execution_contexts.py b/utils/execution_contexts.py new file mode 100644 index 0000000..144c318 --- /dev/null +++ b/utils/execution_contexts.py @@ -0,0 +1,139 @@ +"""Scalar contracts for bounded execution context capture.""" + +import hashlib +import json +from dataclasses import dataclass + + +@dataclass(frozen=True) +class TextEvidence: + value: str | None + provenance: str + source: str + reason: str | None + + +@dataclass(frozen=True) +class PromptSnapshot: + raw: TextEvidence + effective: TextEvidence + prepend: TextEvidence + append: TextEvidence + + +@dataclass(frozen=True) +class RecipeContext: + anchor_id: str + positive: PromptSnapshot | None + negative: PromptSnapshot | None + role_provenance: str + + +@dataclass(frozen=True) +class ContextBundle: + contexts: tuple[RecipeContext, ...] + state: str + reason: str + format_version: int = 1 + + +MAX_FIELD_BYTES = 64 * 1024 +MAX_CONTEXT_BYTES = 256 * 1024 +MAX_ATTEMPT_BYTES = 1024 * 1024 + + +def text_size(value: str, limit: int = MAX_FIELD_BYTES) -> int: + """Reject oversize strings before allocating encoded copies.""" + if type(value) is not str or len(value) > limit: + raise ValueError("Invalid or oversized text") + size = 0 + for char in value: + code = ord(char) + if 0xD800 <= code <= 0xDFFF: + raise ValueError("Invalid Unicode") + size += 1 if code < 128 else 2 if code < 2048 else 3 if code < 65536 else 4 + if size > limit: + raise ValueError("Oversized text") + return size + + +def canonical_content(kind: str, fields: tuple[str | None, ...]) -> bytes: + text_size(kind, 128) + if type(fields) is not tuple: + raise ValueError("Invalid canonical fields") + for value in fields: + if value is not None: + text_size(value) + return json.dumps( + [1, kind, *fields], ensure_ascii=False, allow_nan=False, separators=(",", ":") + ).encode("utf-8") + + +def content_digest(kind: str, fields: tuple[str | None, ...]) -> str: + return hashlib.sha256(canonical_content(kind, fields)).hexdigest() + + +def snapshot_digest(snapshot: PromptSnapshot) -> str: + return content_digest( + "snapshot", + tuple( + getattr(snapshot, field).value + for field in ("raw", "effective", "prepend", "append") + ), + ) + + +def context_digest(context: RecipeContext) -> str: + return content_digest( + "context", + tuple( + snapshot_digest(side) if side is not None else None + for side in (context.positive, context.negative) + ), + ) + + +def validate_bundle(bundle: ContextBundle) -> None: + if ( + type(bundle) is not ContextBundle + or type(bundle.format_version) is not int + or bundle.format_version != 1 + or type(bundle.contexts) is not tuple + or len(bundle.contexts) > 16 + ): + raise ValueError("Invalid context bundle") + total = text_size(bundle.state, 128) + text_size(bundle.reason, 128) + anchors = set() + for context in bundle.contexts: + if type(context) is not RecipeContext: + raise ValueError("Invalid recipe context") + size = text_size(context.anchor_id, 512) + if len(context.anchor_id) > 128: + raise ValueError("Oversized recipe anchor") + if not context.anchor_id or context.anchor_id in anchors: + raise ValueError("Duplicate or empty recipe anchor") + anchors.add(context.anchor_id) + size += text_size(context.role_provenance, 128) + for snapshot in (context.positive, context.negative): + if snapshot is None: + continue + if type(snapshot) is not PromptSnapshot: + raise ValueError("Invalid prompt snapshot") + for field in ( + snapshot.raw, + snapshot.effective, + snapshot.prepend, + snapshot.append, + ): + if type(field) is not TextEvidence: + raise ValueError("Invalid text evidence") + if field.value is not None: + size += text_size(field.value) + size += text_size(field.provenance, 128) + text_size(field.source, 128) + if field.reason is not None: + size += text_size(field.reason, 128) + if size > MAX_CONTEXT_BYTES: + raise ValueError("Context payload limit reached") + total += size + if total > MAX_ATTEMPT_BYTES: + raise ValueError("Attempt payload limit reached")