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).
This commit is contained in:
Vito Sansevero
@@ -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 "<instance>:<inner>" 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):
|
||||
|
||||
@@ -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
|
||||
|
||||
Reference in New Issue
Block a user