From c86002ec05f366c6e31032a7dbfd1717964c8283 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Mon, 19 Oct 2026 14:59:00 -0700 Subject: [PATCH] fix(inspect): match the batch index only in the settings footer Review I-1: duplicate A1111 parameters copies were compared after blanking "Batch index: N" anywhere in the record, so copies whose prompts differ only by that phrase were accepted. The field is now blanked only inside the record's one "Steps: N, ..." footer line, which must hold it exactly once; any other difference stays a conflict. Review M-4: the inspector parses the workflow record only when the API graph can show the overwritten-PromptManager signature (a PromptManager class, or a node with PromptManager's search_text input). --- tests/test_inspection_legacy_ports.py | 104 ++++++++++++++++++++++++++ utils/prompt_inspector.py | 48 ++++++++++-- 2 files changed, 146 insertions(+), 6 deletions(-) diff --git a/tests/test_inspection_legacy_ports.py b/tests/test_inspection_legacy_ports.py index d300a4a..13e6c54 100644 --- a/tests/test_inspection_legacy_ports.py +++ b/tests/test_inspection_legacy_ports.py @@ -148,6 +148,91 @@ class TestOverwrittenPromptManagerText(unittest.TestCase): metadata["prompt"]["3"]["inputs"]["positive"] = ["7", 0] self.assertEqual(_texts(_inspect(metadata)), (parity.NEG, parity.NEG)) + def test_overwritten_nodes_inside_subgraphs_are_refused(self): + # Review M-1: subgraph nodes run as ":" API IDs. + def nested(depth: int = 1) -> dict: + metadata = _overwritten() + prompt, workflow = metadata["prompt"], metadata["workflow"] + prefix = ":".join(["50", "60", "70"][:depth]) + ":" + for node_id in ("17", "113", "119"): + prompt[prefix + node_id] = prompt.pop(node_id) + prompt["3"]["inputs"]["positive"] = [prefix + "17", 0] + inner = [n for n in workflow["nodes"] if n["type"] == "PromptManager"] + outer = [n for n in workflow["nodes"] if n["type"] != "PromptManager"] + subgraphs = [] + for level in range(depth, 0, -1): + nodes = ( + inner + if level == depth + else [{"id": [50, 60, 70][level], "type": f"sub-{level + 1}"}] + ) + subgraphs.append({"id": f"sub-{level}", "nodes": nodes}) + metadata["workflow"] = { + "nodes": outer + [{"id": 50, "type": "sub-1", "widgets_values": []}], + "links": [], + "definitions": {"subgraphs": subgraphs}, + } + return metadata + + for depth in (1, 2, 3): + with self.subTest(depth=depth): + for anchor in (None, "9"): + result = _inspect(nested(depth), output_node_id=anchor) + self.assertEqual( + result["reason"], resolver.OVERWRITTEN_PROMPT_REASON, anchor + ) + # A subgraph definition with no instance runs nothing. + metadata = nested() + metadata["workflow"]["nodes"] = metadata["workflow"]["nodes"][:-1] + self.assertEqual(_inspect(metadata)["state"], "resolved") + # Malformed definitions never raise. + for definitions in ("x", {"subgraphs": "x"}, {"subgraphs": [1, {"id": 3}]}): + metadata = nested() + metadata["workflow"]["definitions"] = definitions + self.assertEqual(_inspect(metadata)["state"], "resolved", definitions) + # A self-referencing subgraph is bounded. + metadata = nested() + metadata["workflow"]["definitions"]["subgraphs"].append( + {"id": "loop", "nodes": [{"id": 1, "type": "loop"}]} + ) + metadata["workflow"]["nodes"].append({"id": 99, "type": "loop"}) + self.assertEqual( + _inspect(metadata)["reason"], resolver.OVERWRITTEN_PROMPT_REASON + ) + + def test_object_widget_values_name_the_text(self): + metadata = _overwritten() + for node in metadata["workflow"]["nodes"]: + node["widgets_values"] = {"text": node["widgets_values"][0]} + self.assertEqual( + _inspect(metadata)["reason"], resolver.OVERWRITTEN_PROMPT_REASON + ) + + def test_the_workflow_is_parsed_only_for_prompt_manager_graphs(self): + # Review M-4: no PromptManager-like node, no workflow parse. + from unittest import mock + + from utils import prompt_inspector + + prompt = parity.build("f02")["prompt"] + with mock.patch.object( + prompt_inspector, + "_parsed_workflow", + wraps=prompt_inspector._parsed_workflow, + ) as parsed: + inspect_prompt_evidence( + {"prompt": json.dumps(prompt), "workflow": json.dumps({"nodes": []})} + ) + self.assertEqual(parsed.call_count, 0) + metadata = _overwritten() + inspect_prompt_evidence( + { + "prompt": json.dumps(metadata["prompt"]), + "workflow": json.dumps(metadata["workflow"]), + } + ) + self.assertEqual(parsed.call_count, 1) + def test_default_profile_ignores_the_workflow(self): result = resolve_prompt_pair(_overwritten()) self.assertEqual(result["state"], "resolved") @@ -194,6 +279,8 @@ class TestHistoricalPromptManagerNodes(unittest.TestCase): ({"text": "", "prepend_text": "trig", "append_text": "sky"}, "trig, sky"), ({"text": " ", "prepend_text": " ", "append_text": "sky"}, "sky"), ({"text": "a cat"}, "a cat"), + # Review M-5: an explicit null part is skipped, as ``if prepend_text:``. + ({"text": "a cat", "prepend_text": None, "append_text": None}, "a cat"), ) for inputs, expected in cases: with self.subTest(inputs=inputs): @@ -427,6 +514,23 @@ class TestBatchIndexDuplicates(unittest.TestCase): "no batch index": [_batch(0), _batch(0).replace(", Batch index: 0", "")], "two batch indexes": [_batch(0), _batch(1) + ", Batch index: 2"], "prose": [_batch(0), _batch(1).replace("a cat", "a cat Batch index: 1")], + # Review I-1: the field is matched only in the settings footer. + "prompt field": [ + _batch(0).replace("a cat", "a cat, Batch index: 0"), + _batch(0).replace("a cat", "a cat, Batch index: 1"), + ], + "prompt only": [ + "a cat, Batch index: 0\nNegative prompt: bad\nSteps: 20, Seed: 5", + "a cat, Batch index: 1\nNegative prompt: bad\nSteps: 20, Seed: 5", + ], + "negative field": [ + _batch(0).replace("bad", "bad, Batch index: 0"), + _batch(1).replace("bad", "bad, Batch index: 1"), + ], + "no footer": [ + "a cat\nBatch index: 0, Batch size: 2", + "a cat\nBatch index: 1, Batch size: 2", + ], } for label, texts in cases.items(): with self.subTest(label): diff --git a/utils/prompt_inspector.py b/utils/prompt_inspector.py index 488c5fb..62b9515 100644 --- a/utils/prompt_inspector.py +++ b/utils/prompt_inspector.py @@ -17,6 +17,7 @@ from PIL import Image from .inspection_outcomes import UNREADABLE_METADATA_REASON, outcome_message from .pair_execution import EXECUTION_KEY from .prompt_pair_resolver import ( + _FOOTER, EMPTY_POSITIVE_REASON, INSPECTION_PROFILE, ResolverProfile, @@ -44,7 +45,21 @@ _METADATA_KEYS = _ROLE_METADATA_KEYS | {"workflow"} _TEXT_CHUNKS = (b"tEXt", b"zTXt", b"iTXt") # An accumulating batch saver writes one A1111 ``parameters`` record per batch # item, identical except this field (R3); Pillow keeps the last one. -_BATCH_INDEX = re.compile(r"(?m)(?:^|(?<=, ))Batch index: \d+(?=, |$)") +# Matched only inside the record's one ``Steps: N, ...`` settings footer line. +_BATCH_INDEX = re.compile(r"(?:^|(?<=, ))Batch index: \d+(?=, |$)") + + +def _batch_normalized(text: str) -> str | None: + """``text`` with its footer's single batch index blanked, else ``None``.""" + footers = list(_FOOTER.finditer(text)) + if len(footers) != 1: + return None + line, count = _BATCH_INDEX.subn("Batch index: #", footers[0].group()) + if count != 1: + return None + return text[: footers[0].start()] + line + text[footers[0].end() :] + + _SIGNATURE = b"\x89PNG\r\n\x1a\n" @@ -200,10 +215,8 @@ class _PngRecords: if value is None or self.duplicate_decoded > DUPLICATE_ROLE_BYTES: raise MetadataReadError("Metadata exceeds inspection limits") kept = self.found["parameters"] - normalized = [ - _BATCH_INDEX.subn("Batch index: #", text) for text in (kept, value) - ] - if normalized[0] != normalized[1] or normalized[0][1] != 1: + normalized = _batch_normalized(kept) + if normalized is None or normalized != _batch_normalized(value): raise MetadataReadError("Duplicate prompt metadata") self.role_decoded += max(0, size - len(kept.encode())) if self.role_decoded > ROLE_METADATA_BYTES: @@ -467,6 +480,27 @@ def unreadable_evidence() -> dict: return {"prompts": _explained(prompts, resolution), "resolution": resolution} +def _may_hold_prompt_manager(metadata: dict) -> bool: + """Whether the API graph can show the R1 signature, so the workflow is needed. + + PromptManager classes, or a ``CLIPTextEncode`` carrying PromptManager's own + ``search_text`` input (an older save rewrote the class). + """ + graph = metadata.get("prompt") + if not isinstance(graph, dict): + return False + for node in graph.values(): + if not isinstance(node, dict): + continue + name = node.get("class_type") + inputs = node.get("inputs") + if isinstance(name, str) and "promptmanager" in name.lower(): + return True + if isinstance(inputs, dict) and "search_text" in inputs: + return True + return False + + def _parsed_workflow(value: Any) -> dict | None: """The UI workflow as an object, or ``None`` (absent, invalid or not JSON).""" if isinstance(value, str): @@ -539,7 +573,9 @@ def inspect_prompt_evidence( canonical["prompt"] = json.loads(canonical["prompt"]) except (ValueError, RecursionError): return malformed - workflow = _parsed_workflow(workflow) + workflow = ( + _parsed_workflow(workflow) if _may_hold_prompt_manager(canonical) else None + ) if workflow is not None and "prompt" in canonical: # Only to recognize overwritten PromptManager text (R1); never a role. canonical["workflow"] = workflow