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).
This commit is contained in:
Vito Sansevero
@@ -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"]);
|
||||
});
|
||||
@@ -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()
|
||||
|
||||
@@ -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()
|
||||
|
||||
@@ -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"})
|
||||
|
||||
|
||||
@@ -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),
|
||||
|
||||
@@ -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.
|
||||
|
||||
@@ -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:
|
||||
|
||||
@@ -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 => {
|
||||
|
||||
Reference in New Issue
Block a user