From baf788b8633c4254bc24908f17f8d4c53cabae85 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Mon, 19 Oct 2026 14:17:08 -0700 Subject: [PATCH] feat(inspect): port legacy parser rules the new resolver missed (A1111) R3: an accumulating batch saver repeats the A1111 parameters record once per batch item, identical except "Batch index: N". The bounded PNG reader now keeps the last copy (Pillow's rule, which #165 relied on) instead of calling the file unreadable. Any other difference, including a missing or second batch index, is still a conflict, and decoding the copies draws from the duplicate role budget. R4: parameters text that starts with "Negative prompt:" has an empty positive (#165), instead of showing the delimiter line as the positive. R5: the settings footer may be followed by sd-dynamic-prompts' one-line "Template:" and "Negative Template:" records, and may be "Steps: N" alone when there is no full footer. R10: commas inside A1111's quoted settings values no longer split them, so an "ADetailer prompt" with a comma keeps the other facts known. The facts' A1111 reader mirrors R4, R5 and R10; the default (ingestion) profile keeps the historical split. Also from the Task 2.7 review: a guard test fails on any uncoded unresolved reason without a user line (M1), and the encoder bounds role lines to the client's 512 code units (M4). --- tests/js/inspection-legacy-ports.test.js | 59 ++++++++ tests/test_inspection_legacy_ports.py | 178 ++++++++++++++++++++++- tests/test_inspection_outcomes.py | 131 +++++++++++++++++ utils/inspection_outcomes.py | 2 + utils/inspection_response.py | 29 +++- utils/prompt_inspector.py | 36 ++++- utils/prompt_pair_resolver.py | 61 +++++++- web/js/comfy-metadata.js | 43 +++++- 8 files changed, 523 insertions(+), 16 deletions(-) create mode 100644 tests/js/inspection-legacy-ports.test.js diff --git a/tests/js/inspection-legacy-ports.test.js b/tests/js/inspection-legacy-ports.test.js new file mode 100644 index 0000000..04283c3 --- /dev/null +++ b/tests/js/inspection-legacy-ports.test.js @@ -0,0 +1,59 @@ +// Run with: node --test tests/js/ +// Legacy parser rules ported to the inspection facts (#240, Task 2.7b). Synthetic texts. +const test = require("node:test"); +const assert = require("node:assert/strict"); +const { extractGenerationFacts } = require("../../web/js/comfy-metadata.js"); + +const fields = ["seed", "steps", "cfg", "sampler", "scheduler", "model"]; +const FOOTER = "Steps: 20, Sampler: Euler, CFG scale: 7, Seed: 5"; + +function parameterFacts(parameters) { + const facts = extractGenerationFacts({ + prompt_graph: { status: "absent", value: null }, + workflow_graph: { status: "absent", value: null }, + parameters, + recordFormats: { parameters: "original-text" }, + metadataStatus: "available", + }); + return Object.fromEntries(fields.map(field => [field, [facts[field].status, facts[field].value]])); +} + +const FOUR_KNOWN = { + seed: ["known", "5"], steps: ["known", "20"], cfg: ["known", "7"], sampler: ["known", "Euler"], + scheduler: ["not-recorded", null], model: ["not-recorded", null], +}; +const ALL_UNRESOLVED = Object.fromEntries(fields.map(field => [field, ["unresolved", null]])); + +test("R4: text starting with the negative delimiter has no positive, as with a leading newline", () => { + for (const text of [`Negative prompt: bad\n${FOOTER}`, `\nNegative prompt: bad\n${FOOTER}`, `Negative prompt: a\nNegative prompt: b\n${FOOTER}`]) { + assert.deepEqual(parameterFacts(text), ALL_UNRESOLVED, text); + } + assert.deepEqual(parameterFacts(`a cat\nNegative prompt: bad\n${FOOTER}`), FOUR_KNOWN); +}); + +test("R5: template records after the footer and a bare Steps footer", () => { + const base = `a cat\nNegative prompt: bad\n${FOOTER}`; + for (const trailer of ["\nTemplate: a {cat|dog}", "\nTemplate: a {cat|dog}\nNegative Template: bad", "\nNegative Template: bad\n"]) { + assert.deepEqual(parameterFacts(base + trailer), FOUR_KNOWN, trailer); + } + for (const trailer of ["\nTemplate: a\nTemplate: b", "\nNegative Template: bad\nTemplate: a", "\nTemplate: a\nmore", "\ntrailing"]) { + assert.deepEqual(parameterFacts(base + trailer), ALL_UNRESOLVED, trailer); + } + const bare = parameterFacts("a cat\nNegative prompt: bad\nSteps: 20\nTemplate: a {cat|dog}"); + assert.deepEqual(bare.steps, ["known", "20"]); + assert.ok(["seed", "cfg", "sampler", "scheduler", "model"].every(field => bare[field][0] === "not-recorded")); + // A full footer wins over a bare `Steps: N` line in the prompt; two bare ones are not unique. + assert.deepEqual(parameterFacts(`a cat\nSteps: 3\nNegative prompt: bad\n${FOOTER}`), FOUR_KNOWN); + assert.deepEqual(parameterFacts("a cat\nSteps: 3\nNegative prompt: bad\nSteps: 20"), ALL_UNRESOLVED); +}); + +test("R10: commas inside quoted settings values do not split them", () => { + assert.deepEqual(parameterFacts(`a cat\nNegative prompt: bad\n${FOOTER}, ADetailer prompt: "face, smile"`), FOUR_KNOWN); + assert.deepEqual(parameterFacts(`a cat\n${FOOTER}, Note: "say \\"hi, there\\"", Model: base`), { ...FOUR_KNOWN, model: ["known", "base"] }); + // A quoted value of a fact is its JSON string. + assert.deepEqual(parameterFacts(`a cat\n${FOOTER}, Model: "base, v2"`).model, ["known", "base, v2"]); + // Unquoted JSON (comfy-image-saver hashes) keeps the historical split, field by field. + assert.deepEqual(parameterFacts(`a cat\n${FOOTER}, Hashes: {"model": "abc", "lora:x": "def"}, Model: base`).model, ["known", "base"]); + // An unbalanced quote is not a quoted value: the historical split, as A1111 reads it. + assert.deepEqual(parameterFacts(`a cat\n${FOOTER}, Note: "open, Model: base`).model, ["known", "base"]); +}); diff --git a/tests/test_inspection_legacy_ports.py b/tests/test_inspection_legacy_ports.py index a2c259d..35268b7 100644 --- a/tests/test_inspection_legacy_ports.py +++ b/tests/test_inspection_legacy_ports.py @@ -6,11 +6,22 @@ the rule from the parser reconciliation (R1, R3, ...) it ports. The default ``DEFAULT_PROFILE`` ignores it. """ +import io import json +import tempfile import unittest +import zlib +from pathlib import Path + +from PIL import Image from utils import prompt_pair_resolver as resolver -from utils.prompt_inspector import inspect_prompt_evidence +from utils.prompt_inspector import ( + DUPLICATE_ROLE_BYTES, + MetadataReadError, + inspect_prompt_evidence, + read_inspection_file, +) from utils.prompt_pair_resolver import resolve_prompt_pair try: @@ -274,5 +285,170 @@ class TestHistoricalPromptManagerNodes(unittest.TestCase): self.assertEqual(result["reason"], "Unsupported node semantics or output slot") +FOOTER = "Steps: 20, Sampler: Euler, CFG scale: 7, Seed: 5" + + +def _parameters(text: str, *, legacy: bool = False) -> dict: + profile = resolver.DEFAULT_PROFILE if legacy else resolver.INSPECTION_PROFILE + return resolve_prompt_pair({"parameters": text}, profile=profile) + + +class TestA1111Text(unittest.TestCase): + """R4 and R5 (#165, py/api/__init__.py:738-750) for inspection.""" + + def test_text_starting_with_the_negative_delimiter_has_no_positive(self): + for text in ( + f"Negative prompt: bad\n{FOOTER}", + f"\nNegative prompt: bad\n{FOOTER}", + ): + with self.subTest(text=text): + result = _parameters(text) + self.assertEqual(result["state"], "unresolved") + self.assertEqual(result["reason"], resolver.EMPTY_POSITIVE_REASON) + self.assertIsNone(result["positive"]) + # Two delimiters, one at the start, are still two. + result = _parameters(f"Negative prompt: a\nNegative prompt: b\n{FOOTER}") + self.assertEqual(result["state"], "ambiguous") + + def test_dynamic_prompts_template_lines_follow_the_footer(self): + base = f"a cat\nNegative prompt: bad\n{FOOTER}" + for trailer in ( + "\nTemplate: a {cat|dog}", + "\nTemplate: a {cat|dog}\nNegative Template: bad", + "\nNegative Template: bad", + "\nTemplate: a {cat|dog}\nNegative Template: bad\n", + ): + with self.subTest(trailer=trailer): + result = _parameters(base + trailer) + self.assertEqual(result["state"], "resolved") + self.assertEqual(_texts(result), ("a cat", "bad")) + for trailer in ( + "\nTemplate: a\nTemplate: b", + "\nNegative Template: bad\nTemplate: a", + "\nTemplate: a\nmore template text", + "\ntrailing", + ): + with self.subTest(trailer=trailer): + result = _parameters(base + trailer) + self.assertEqual(result["reason_code"], "malformed") + + def test_footer_with_only_steps(self): + result = _parameters("a cat\nNegative prompt: bad\nSteps: 20") + self.assertEqual(_texts(result), ("a cat", "bad")) + result = _parameters("a cat\nSteps: 20\nTemplate: a {cat|dog}") + self.assertEqual((result["state"], _texts(result)[0]), ("absent", "a cat")) + # A full footer wins over a bare ``Steps: N`` line in the prompt text. + result = _parameters(f"a cat\nSteps: 3\nNegative prompt: bad\n{FOOTER}") + self.assertEqual(_texts(result), ("a cat\nSteps: 3", "bad")) + result = _parameters("a cat\nSteps: 3\nNegative prompt: bad\nSteps: 20") + self.assertEqual(result["reason_code"], "malformed") + + def test_default_profile_keeps_the_historical_split(self): + result = _parameters(f"Negative prompt: bad\n{FOOTER}", legacy=True) + self.assertEqual(_texts(result), ("Negative prompt: bad", None)) + for text in ( + f"a cat\nNegative prompt: bad\n{FOOTER}\nTemplate: a {{cat|dog}}", + "a cat\nNegative prompt: bad\nSteps: 20", + ): + with self.subTest(text=text): + result = _parameters(text, legacy=True) + self.assertEqual( + result["reason"], "A1111 settings footer is not unique" + ) + + +def _chunk(keyword: str, text: str, kind: bytes = b"tEXt") -> bytes: + body = text.encode("latin-1") + if kind == b"zTXt": + body = b"\0" + zlib.compress(body) + elif kind == b"iTXt": + body = b"\0\0\0\0" + text.encode("utf-8") + return parity._raw_chunk(kind, keyword.encode("latin-1") + b"\0" + body) + + +def _batch(index: int, positive: str = "a cat", seed: int = 5) -> str: + return ( + f"{positive}\nNegative prompt: bad\nSteps: 20, Sampler: Euler, " + f"CFG scale: 7, Seed: {seed}, Batch index: {index}, Batch size: 12" + ) + + +class TestBatchIndexDuplicates(unittest.TestCase): + """R3: an accumulating batch saver repeats ``parameters`` per batch item.""" + + def setUp(self): + temporary = tempfile.TemporaryDirectory() + self.addCleanup(temporary.cleanup) + self.directory = Path(temporary.name) + + def png(self, chunks, name="batch.png"): + buffer = io.BytesIO() + with Image.new("RGB", (4, 4), (40, 80, 120)) as image: + image.save(buffer, format="PNG") + data = buffer.getvalue() + iend = data.rindex(b"IEND") - 4 + path = self.directory / name + path.write_bytes(data[:iend] + b"".join(chunks) + data[iend:]) + return path + + def test_copies_differing_only_in_batch_index_read_as_the_last(self): + cases = { + "same length": [0, 1, 2], + "growing length": [8, 9, 10, 11], + "any order": [3, 0], + "repeat of an earlier copy": [0, 1, 0], + } + for label, indexes in cases.items(): + with self.subTest(label): + path = self.png([_chunk("parameters", _batch(i)) for i in indexes]) + read = read_inspection_file(path) + self.assertEqual(read.metadata["parameters"], _batch(indexes[-1])) + self.assertEqual(read.record_formats["parameters"], "original-text") + evidence = inspect_prompt_evidence(read.metadata) + self.assertEqual(evidence["resolution"]["state"], "resolved") + self.assertEqual(evidence["prompts"]["positive"]["text"], "a cat") + # Encodings may differ; the decoded copies are compared. + path = self.png( + [ + _chunk("parameters", _batch(0)), + _chunk("parameters", _batch(1), b"zTXt"), + _chunk("parameters", _batch(2), b"iTXt"), + ] + ) + self.assertEqual(read_inspection_file(path).metadata["parameters"], _batch(2)) + + def test_any_other_difference_is_still_a_conflict(self): + cases = { + "positive": [_batch(0), _batch(1, positive="a dog")], + "seed": [_batch(0), _batch(1, seed=6)], + "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")], + } + for label, texts in cases.items(): + with self.subTest(label): + path = self.png([_chunk("parameters", text) for text in texts]) + with self.assertRaisesRegex( + MetadataReadError, "Duplicate prompt metadata" + ): + read_inspection_file(path) + # The rule is only for ``parameters``: prompt graphs stay byte-identical. + prompt = json.dumps(parity.build("f02")["prompt"]) + path = self.png([_chunk("prompt", prompt), _chunk("prompt", prompt + " ")]) + with self.assertRaisesRegex(MetadataReadError, "Duplicate prompt metadata"): + read_inspection_file(path) + + def test_decoded_copies_are_bounded(self): + # Small compressed copies must not decode unbounded text. + text = _batch(0) + "," + " " * (DUPLICATE_ROLE_BYTES // 8) + chunks = [ + _chunk("parameters", text.replace("index: 0", f"index: {i}"), b"zTXt") + for i in range(10) + ] + path = self.png(chunks) + with self.assertRaisesRegex(MetadataReadError, "exceeds inspection limits"): + read_inspection_file(path) + + if __name__ == "__main__": unittest.main() diff --git a/tests/test_inspection_outcomes.py b/tests/test_inspection_outcomes.py index 2420ea9..c7670b6 100644 --- a/tests/test_inspection_outcomes.py +++ b/tests/test_inspection_outcomes.py @@ -1,11 +1,15 @@ """Non-resolved prompt outcomes carry one short, truthful line per role.""" +import ast import json import unittest +from pathlib import Path from dataclasses import replace from unittest import mock +from utils import prompt_pair_resolver from utils.inspection_outcomes import ( + _UNCODED_LINES, CLASS_NAME_LENGTH, REASON_CODES, outcome_message, @@ -351,5 +355,132 @@ class TestResolutionProjection(unittest.TestCase): self.assertIsNone(body["resolution"]["node"]) +# Uncoded unresolved reasons only the default (ingestion) profile emits; the +# inspector never shows them, so they need no line. +_DEFAULT_PROFILE_ONLY = frozenset( + {"Cyclic or excessive latent ancestry", "Resolution limit reached"} +) + + +def _uncoded_unresolved_reasons() -> set: + """Every literal reason the resolver can return as unresolved without a code. + + Reads the source: ``_Unresolved(reason)`` without a code, and + ``_result("unresolved", ..., reason=...)`` without ``reason_code``. A reason + is a string literal or a module constant naming one. + """ + source = Path(prompt_pair_resolver.__file__).read_text(encoding="utf-8") + tree = ast.parse(source) + + def literal(node): + if isinstance(node, ast.Constant) and isinstance(node.value, str): + return node.value + if isinstance(node, ast.Name): + value = getattr(prompt_pair_resolver, node.id, None) + return value if isinstance(value, str) else None + return None + + found = set() + for node in ast.walk(tree): + if not isinstance(node, ast.Call) or not isinstance(node.func, ast.Name): + continue + keywords = {item.arg: item.value for item in node.keywords} + if node.func.id == "_Unresolved": + if len(node.args) < 2 and "code" not in keywords and node.args: + reason = literal(node.args[0]) + if reason is None: + raise AssertionError(f"line {node.lineno}: unreadable reason") + found.add(reason) + elif node.func.id == "_result" and node.args: + state = literal(node.args[0]) + coded = "reason_code" in keywords or len(node.args) > 6 + reason = keywords.get( + "reason", node.args[4] if len(node.args) > 4 else None + ) + if state == "unresolved" and not coded and reason is not None: + text = literal(reason) + if text is None: + raise AssertionError(f"line {node.lineno}: unreadable reason") + found.add(text) + return found + + +class TestEveryUncodedOutcomeHasALine(unittest.TestCase): + """Task 2.7 review M1: a new uncoded unresolved reason fails here.""" + + def test_resolver_reasons_without_a_code_have_a_line(self): + reasons = _uncoded_unresolved_reasons() + self.assertIn(EMPTY_POSITIVE_REASON, reasons, "the scan sees the source") + missing = reasons - set(_UNCODED_LINES) - _DEFAULT_PROFILE_ONLY + self.assertEqual(missing, set(), "add a code or a line in inspection_outcomes") + # The allowance names only reasons the source still emits. + self.assertLessEqual(_DEFAULT_PROFILE_ONLY, reasons) + + def test_default_profile_only_reasons_are_coded_in_inspection(self): + # The limit outcome is uncoded only on the default profile. + source = {"prompt": {"1": {"class_type": "KSampler", "inputs": {}}}} + for limited in ({"max_depth": 0}, {"max_text_length": 0}): + with self.subTest(**limited): + default = prompt_pair_resolver.resolve_prompt_pair(source, **limited) + inspected = prompt_pair_resolver.resolve_prompt_pair( + source, profile=INSPECTION_PROFILE, **limited + ) + self.assertEqual(default["reason"], "Resolution limit reached") + self.assertIsNone(default["reason_code"]) + self.assertEqual(inspected["reason"], "Resolution limit reached") + self.assertIsNotNone(inspected["reason_code"]) + + +class TestRoleMessageBound(unittest.TestCase): + """Task 2.7 review M4: the encoder bounds role lines as the client does.""" + + def encode(self, prompts: dict) -> dict: + return json.loads( + encode_inspection_response( + image={"root_index": 0, "path": "a.png", "filename": "a.png"}, + root_scope="a" * 64, + file={ + "observation": "current-file", + "source_label": "output", + "media_type": "image", + "extension": ".png", + "size": 1, + "modified_time": 1, + }, + prompts=prompts, + resolution={"state": "unresolved", "reason": "x"}, + inspection=None, + library={"status": "none", "tags": []}, + ) + )["prompts"] + + def test_role_messages_are_bounded_text_on_unresolved_roles(self): + side = {"status": "unresolved", "text": None, "source": None} + prompts = self.encode( + { + "positive": {**side, "message": "x" * 600}, + "negative": {**side, "message": "\U0001f600" * 300}, + } + ) + self.assertEqual(prompts["positive"]["message"], "x" * 512) + # The client counts UTF-16 code units. + self.assertLessEqual( + len(prompts["negative"]["message"].encode("utf-16-le")), 1024 + ) + self.assertEqual(prompts["negative"]["message"], "\U0001f600" * 256) + for message in ("", None, 7): + with self.subTest(message=message): + prompts = self.encode( + {"positive": {**side, "message": message}, "negative": side} + ) + self.assertNotIn("message", prompts["positive"]) + resolved = {"status": "resolved", "text": "a", "source": "graph-inferred"} + prompts = self.encode( + {"positive": {**resolved, "message": "line"}, "negative": side} + ) + self.assertNotIn("message", prompts["positive"]) + self.assertEqual(prompts["negative"], side) + + if __name__ == "__main__": unittest.main() diff --git a/utils/inspection_outcomes.py b/utils/inspection_outcomes.py index 27eadf6..295daf1 100644 --- a/utils/inspection_outcomes.py +++ b/utils/inspection_outcomes.py @@ -34,6 +34,8 @@ LIMIT_CODES = frozenset({"work-limit", "depth-limit", "text-limit", "graph-ceili NAMED_CODES = frozenset({"unsupported-node", "runtime-value", "composite"}) # Bound for a node ID or class name on the wire and in a line. CLASS_NAME_LENGTH = 200 +# Bound for a role's line on the wire, in UTF-16 code units (the client's count). +MESSAGE_LENGTH = 512 UNREADABLE_METADATA_REASON = "Media metadata could not be read" _NODE_KEYS = frozenset({"id", "class_type"}) diff --git a/utils/inspection_response.py b/utils/inspection_response.py index 357e97b..02c37d3 100644 --- a/utils/inspection_response.py +++ b/utils/inspection_response.py @@ -3,9 +3,9 @@ import json import math from decimal import Decimal, InvalidOperation -from typing import TYPE_CHECKING, Literal +from typing import TYPE_CHECKING, Any, Literal, Optional -from .inspection_outcomes import project_resolution +from .inspection_outcomes import MESSAGE_LENGTH, project_resolution if TYPE_CHECKING: from .prompt_inspector import InspectionRead @@ -314,6 +314,29 @@ def _library_projection(library: dict) -> dict: return result +def _bounded_message(value: Any) -> Optional[str]: + """A role line as the client accepts it: 1 to 512 UTF-16 code units.""" + if not isinstance(value, str) or not value: + return None + value = value[:MESSAGE_LENGTH] + while len(value.encode("utf-16-le", "surrogatepass")) > 2 * MESSAGE_LENGTH: + value = value[:-1] + return value + + +def _bounded_messages(prompts: dict) -> dict: + """Keep a role ``message`` only as bounded text on a non-resolved role.""" + bounded = {} + for role, side in prompts.items(): + if isinstance(side, dict) and "message" in side: + side = {key: item for key, item in side.items() if key != "message"} + message = _bounded_message(prompts[role]["message"]) + if message is not None and side.get("status") != "resolved": + side["message"] = message + bounded[role] = side + return bounded + + def _encode_inspection( *, identity: dict, @@ -330,7 +353,7 @@ def _encode_inspection( "inspection_version": 1, **identity, "file": _file_projection(file, inspection), - "prompts": prompts, + "prompts": _bounded_messages(prompts), "resolution": project_resolution(resolution), "metadata": metadata, "library": _library_projection(library), diff --git a/utils/prompt_inspector.py b/utils/prompt_inspector.py index 4778e36..488c5fb 100644 --- a/utils/prompt_inspector.py +++ b/utils/prompt_inspector.py @@ -3,6 +3,7 @@ import hashlib import io import json +import re import struct import subprocess # nosec B404 import threading @@ -41,6 +42,9 @@ MAX_PNG_CHUNKS = 10_000 _ROLE_METADATA_KEYS = frozenset({"prompt", "parameters", EXECUTION_KEY}) _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+(?=, |$)") _SIGNATURE = b"\x89PNG\r\n\x1a\n" @@ -123,13 +127,14 @@ class _PngRecords: self.digests: dict[str, tuple[int, bytes]] = {} self.role_encoded = self.role_decoded = 0 self.duplicate_workflow = self.duplicate_role = 0 + self.duplicate_decoded = 0 def skip(self, key: str, length: int) -> bool: """Decide from the keyword and length alone whether to seek past a chunk.""" if key not in _METADATA_KEYS or key in self.limited: return True # never retained, or a duplicate of a limited workflow if key in self.digests: - if self.digests[key][0] != length: + if self.digests[key][0] != length and key != "parameters": raise MetadataReadError("Duplicate prompt metadata") if key == "workflow": if self.duplicate_workflow + length > DUPLICATE_WORKFLOW_BYTES: @@ -157,7 +162,11 @@ class _PngRecords: digest = hashlib.sha256(kind + data).digest() if key in self.digests: if self.digests[key][1] != digest: - raise MetadataReadError("Duplicate prompt metadata") + if key != "parameters": + raise MetadataReadError("Duplicate prompt metadata") + self.batch_copy(kind, data) + # The kept copy is the last one; later repeats compare with it. + self.digests[key] = (len(data), digest) return self.digests[key] = (len(data), digest) workflow = key == "workflow" @@ -178,6 +187,29 @@ class _PngRecords: raise MetadataReadError("Metadata exceeds inspection limits") self.found[key] = value + def batch_copy(self, kind: bytes, data: bytes) -> None: + """Keep a later ``parameters`` copy that differs only in its batch index. + + Each copy must carry exactly one ``Batch index: N`` settings field; any + other difference is still a conflict. Decoding every copy draws from the + duplicate role budget, so small compressed copies stay bounded. + """ + _, value = _png_text(kind, data, ROLE_METADATA_BYTES) + size = ROLE_METADATA_BYTES + 1 if value is None else len(value.encode()) + self.duplicate_decoded += size + 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: + raise MetadataReadError("Duplicate prompt metadata") + self.role_decoded += max(0, size - len(kept.encode())) + if self.role_decoded > ROLE_METADATA_BYTES: + raise MetadataReadError("Metadata exceeds inspection limits") + self.found["parameters"] = value + def _png_metadata(stream: Any, size: int) -> tuple[dict, frozenset, int, int]: """CRC-check retained text and structural chunks, seeking past everything else. diff --git a/utils/prompt_pair_resolver.py b/utils/prompt_pair_resolver.py index 46df248..b8ec3fa 100644 --- a/utils/prompt_pair_resolver.py +++ b/utils/prompt_pair_resolver.py @@ -1394,7 +1394,53 @@ class _GraphResolver: raise _Unresolved("Output ancestry limit reached", code) -def _parameters(value: Any, max_text_length: int) -> dict: +_FOOTER = re.compile(r"(?m)^Steps:\s*\d+\s*,[^\n]*$") +# Inspection also reads a footer of ``Steps: N`` alone (no other settings). +_BARE_FOOTER = re.compile(r"(?m)^Steps:\s*\d+\s*$") +# sd-dynamic-prompts writes these one-line records after the settings footer. +_FOOTER_TRAILERS = ("Template:", "Negative Template:") + + +def _trailer_lines(text: str) -> bool: + """Whether ``text`` after the footer is empty or only template records. + + At most one ``Template:`` line, then at most one ``Negative Template:`` + line; anything else (a multi-line template included) is not recognized. + """ + lines = re.split(r"\r?\n|\r", text.strip()) if text.strip() else [] + prefixes = [ + next( + ( + index + for index, name in enumerate(_FOOTER_TRAILERS) + if line.startswith(name) + ), + None, + ) + for line in lines + ] + return None not in prefixes and prefixes == sorted(set(prefixes)) + + +def _footer(value: str, inspection: bool) -> Optional[re.Match]: + """The unique A1111 settings footer, or ``None``. + + The default profile keeps the historical rule (one ``Steps: N, ...`` line, + and it is the last line). Inspection (R5) accepts a bare ``Steps: N`` + footer when there is no full one, and template records after it. + """ + footers = list(_FOOTER.finditer(value)) + if inspection and not footers: + footers = list(_BARE_FOOTER.finditer(value)) + if len(footers) != 1: + return None + rest = value[footers[0].end() :] + if inspection and _trailer_lines(rest): + return footers[0] + return None if rest.strip() else footers[0] + + +def _parameters(value: Any, max_text_length: int, inspection: bool = False) -> dict: if not isinstance(value, str) or not value: return _result("absent", reason="No A1111 parameters") if len(value) > max_text_length: @@ -1403,14 +1449,17 @@ def _parameters(value: Any, max_text_length: int) -> dict: reason="Parameters exceed the length limit", reason_code="text-limit", ) - footers = list(re.finditer(r"(?m)^Steps:\s*\d+\s*,[^\n]*$", value)) - if len(footers) != 1 or value[footers[0].end() :].strip(): + footer = _footer(value, inspection) + if footer is None: return _result( "unresolved", reason="A1111 settings footer is not unique", reason_code="malformed", ) - body = value[: footers[0].start()].rstrip("\r\n") + body = value[: footer.start()].rstrip("\r\n") + if inspection: + # R4: text that starts with the delimiter has an empty positive (#165). + body = "\n" + body delimiter = "\nNegative prompt:" if body.count(delimiter) > 1: return _result("ambiguous", reason="Multiple negative prompt delimiters") @@ -1552,7 +1601,9 @@ def _resolve( if "prompt" in metadata: graph = metadata["prompt"] elif "parameters" in metadata: - return _parameters(metadata["parameters"], profile.max_text_length) + return _parameters( + metadata["parameters"], profile.max_text_length, not profile.legacy + ) else: graph = metadata if not isinstance(graph, dict) or not graph: diff --git a/web/js/comfy-metadata.js b/web/js/comfy-metadata.js index f345075..d2fdbf6 100644 --- a/web/js/comfy-metadata.js +++ b/web/js/comfy-metadata.js @@ -662,18 +662,51 @@ return true; } + // The inspection resolver's `_footer` (Python): R5 also reads a bare `Steps: N` + // footer, and sd-dynamic-prompts' one-line template records after it. + const FOOTER_TRAILERS = ["Template:", "Negative Template:"]; + function trailerLines(text) { + const lines = text.trim().split(/\r?\n|\r/).filter((line, index, all) => all.length > 1 || line); + const order = lines.map(line => FOOTER_TRAILERS.findIndex(name => line.startsWith(name))); + return !order.includes(-1) && order.every((value, index) => index === 0 || value > order[index - 1]); + } + function settingsFooter(parameters) { + let matches = [...parameters.matchAll(/^Steps:\s*\d+\s*,[^\n]*$/gm)]; + if (!matches.length) matches = [...parameters.matchAll(/^Steps:\s*\d+\s*$/gm)]; + if (matches.length !== 1) return null; + return trailerLines(parameters.slice(matches[0].index + matches[0][0].length)) ? matches[0] : null; + } + /** R10: split settings on commas outside A1111's quoted values; unbalanced quotes split as before. */ + function settingsParts(line) { + const parts = []; + let start = 0, quoted = false; + for (let index = 0; index < line.length; index++) { + if (quoted && line[index] === "\\") index++; + else if (line[index] === '"') quoted = !quoted; + else if (line[index] === "," && !quoted) { parts.push(line.slice(start, index)); start = index + 1; } + } + if (quoted) return line.split(","); + parts.push(line.slice(start)); + return parts; + } + function settingValue(value) { + if (!/^"(?:\\.|[^\\"])*"$/.test(value)) return value; + try { return JSON.parse(value); } catch { return value; } + } + function parameterFacts(parameters) { const source = "embedded-parameters"; - const matches = [...parameters.matchAll(/^Steps:\s*\d+\s*,[^\n]*$/gm)]; - if (matches.length !== 1 || parameters.slice(matches[0].index + matches[0][0].length).trim()) return factSet("unresolved", source); - const body = parameters.slice(0, matches[0].index).replace(/[\r\n]+$/, ""); + const footer = settingsFooter(parameters); + if (!footer) return factSet("unresolved", source); + // R4: like the resolver, text starting with the delimiter has an empty positive. + const body = "\n" + parameters.slice(0, footer.index).replace(/[\r\n]+$/, ""); if (!body.split("\nNegative prompt:")[0].trim() || body.split("\nNegative prompt:").length > 2) return factSet("unresolved", source); const settings = new Map(); - for (const part of matches[0][0].split(",")) { + for (const part of settingsParts(footer[0])) { const match = part.match(/^\s*([^:]+):\s*(.*?)\s*$/); if (!match) return factSet("unresolved", source); const [, key, value] = match; - settings.set(key, settings.has(key) ? null : value); + settings.set(key, settings.has(key) ? null : settingValue(value)); } const keys = { seed: "Seed", steps: "Steps", cfg: "CFG scale", sampler: "Sampler", scheduler: "Schedule type", model: "Model" }; return Object.fromEntries(FACT_FIELDS.map(field => {