From decb69864a2c3215c8c45bf7696667fe9b4db8c5 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Sat, 17 Oct 2026 06:59:08 -0700 Subject: [PATCH] fix(metadata): preserve bounded paired execution evidence (#193) --- prompt_manager_pair.py | 14 ++ tests/test_pair_execution.py | 322 ++++++++++++++++++++++++++++++++++ utils/image_monitor.py | 23 +++ utils/metadata_extractor.py | 1 + utils/pair_execution.py | 145 +++++++++++++++ utils/prompt_pair_resolver.py | 115 +++++++++--- utils/usage_tracking.py | 4 + 7 files changed, 603 insertions(+), 21 deletions(-) create mode 100644 tests/test_pair_execution.py create mode 100644 utils/pair_execution.py diff --git a/prompt_manager_pair.py b/prompt_manager_pair.py index 0bea74c..6c85dde 100644 --- a/prompt_manager_pair.py +++ b/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) diff --git a/tests/test_pair_execution.py b/tests/test_pair_execution.py new file mode 100644 index 0000000..fd46d84 --- /dev/null +++ b/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") diff --git a/utils/image_monitor.py b/utils/image_monitor.py index 69b2b8f..a42504e 100644 --- a/utils/image_monitor.py +++ b/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") diff --git a/utils/metadata_extractor.py b/utils/metadata_extractor.py index c6e8541..c1c044a 100644 --- a/utils/metadata_extractor.py +++ b/utils/metadata_extractor.py @@ -116,6 +116,7 @@ class ComfyUIMetadataExtractor: # Extract other common metadata fields metadata_fields = [ + "comfyassets_prompt_pairs", "parameters", "model", "sampler", diff --git a/utils/pair_execution.py b/utils/pair_execution.py new file mode 100644 index 0000000..1dc04e3 --- /dev/null +++ b/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) diff --git a/utils/prompt_pair_resolver.py b/utils/prompt_pair_resolver.py index c9bd353..4db5507 100644 --- a/utils/prompt_pair_resolver.py +++ b/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: diff --git a/utils/usage_tracking.py b/utils/usage_tracking.py index 217241d..674b1fe 100644 --- a/utils/usage_tracking.py +++ b/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)