From fbd9fba7e3ffeb211eb09c0bdeddb4f5258e70c3 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Mon, 19 Oct 2026 10:51:47 -0700 Subject: [PATCH] feat(inspect): support common text encoders and string sources --- TESTING.md | 16 +- tests/fixtures/README.md | 2 +- tests/fixtures/inspection_parity/f21.png | Bin 0 -> 1281 bytes tests/fixtures/inspection_parity/f22.png | Bin 0 -> 1104 bytes tests/fixtures/inspection_parity/f24.png | Bin 0 -> 989 bytes tests/fixtures/inspection_parity/vectors.json | 197 +++++++- tests/inspection_parity.py | 67 ++- .../legacy-metadata-characterization.test.js | 6 +- tests/test_api_prompt_inspector.py | 10 +- tests/test_prompt_pair_resolver.py | 440 ++++++++++++++++++ utils/prompt_pair_resolver.py | 151 +++++- 11 files changed, 864 insertions(+), 25 deletions(-) create mode 100644 tests/fixtures/inspection_parity/f21.png create mode 100644 tests/fixtures/inspection_parity/f22.png create mode 100644 tests/fixtures/inspection_parity/f24.png diff --git a/TESTING.md b/TESTING.md index 895a19f..039237d 100644 --- a/TESTING.md +++ b/TESTING.md @@ -246,7 +246,21 @@ Results that depend on a forced choice are never memoized. A `FaceDetailer` ancestry are compared with it. Equal texts resolve with the detailer as the sampler; different texts are `ambiguous` with `multistage-differs` and one candidate per stage; an unreadable stage makes the chain `unresolved`. Unanchored, a sampler -whose output only feeds detailers (or previews) is not a candidate of its own. If +whose output only feeds detailers (or previews) is not a candidate of its own. +Text encoders: `single` encoders (the Qwen image encoders, `TextEncodeQwenImage21` +slot 0 `prompt` and slot 1 `negative_prompt`, `KikoTextEncode`) read one string +input unchanged. `parts-equal` encoders (CLIPTextEncodeSDXL, Flux, SD3) resolve when +every part is identical or exactly one part is non-empty (exact `""`, whitespace is +text); otherwise `composite`. String sources reproduce the pinned KJNodes code from +literal settings only: `StringConstant` is its `string`; `StringConstantMultiline` +with a literal true `strip_newlines` is `replace("\n", "").strip()`; +`JoinStringMulti` keeps `string_1` (even empty), appends `string_2`..`string_N` +(`N` = `inputcount`) after the literal delimiter unless missing or exactly `""`, and +ignores later inputs. A linked setting is `unsupported-node`; a list output +(`return_list` true or linked) and `PromptSearchList` are `runtime-value` ("Text is +decided at run time"), never one list item. Unverified text nodes (LoRAfromText, +RegexReplace, XYZPrompt, ImpactWildcardProcessor) stay `unsupported-node`. Every +intermediate text is length-checked and charged to the scanned-character budget. If the table cannot be loaded, graph inspection is `unresolved` ("Inspection node semantics are unavailable") and ingestion is unaffected, because it never imports the table. diff --git a/tests/fixtures/README.md b/tests/fixtures/README.md index cab86c8..9cf262b 100644 --- a/tests/fixtures/README.md +++ b/tests/fixtures/README.md @@ -15,7 +15,7 @@ Browser tests use the committed file; FFmpeg is not required to run them. # Inspection parity fixtures -`inspection_parity/` holds small synthetic PNGs (`f01b.png` ... `f23.png`, each +`inspection_parity/` holds small synthetic PNGs (`f01b.png` ... `f24.png`, each under 32 KiB) and `vectors.json`, used by the #240 inspection parity tests. Every graph, prompt text, model name and filename prefix in them is original synthetic content written for this repository. Nothing comes from user output. They are diff --git a/tests/fixtures/inspection_parity/f21.png b/tests/fixtures/inspection_parity/f21.png new file mode 100644 index 0000000000000000000000000000000000000000..26ce75c5e74948b3e43adfaa601b2d185700c009 GIT binary patch literal 1281 zcmbVM&ubGw6kbaOsnnAvy`%#kgc7o8(pFE6P0(mtgQ+5fx=wc|n`viv#+ljtu;icz zq5py>PenY42;THB5D$Vk?;eDL=vna1Y}-h>2P?R?5=@L12lNJH+1>oW^>E zazRedcXqMe+)_i6*h)Qla4xQpy+>yYQd9Xp<#U4rX(V_asrKby@Ut8T7ZB*mPTtwObJqKy;}H zR7w+JOemHOZ##s?`Mh~nY3h+JBsm~beuo5h0rtjRMwl7?fW(|C+s1qW))6#{0b2{k z;-D*nE2?SN#I(@-AsUkqAE66}D6@`w)LLrd%l3`!>Y~n7-nyEHr#7wIvBx%z=WTIsrwFh_Wz=v$fso>(MW|)oRMc;&T z(bSU0;cCJ(wb`RnVGoGUlh{m`@d_GBBdWnqP=qyHlITRn9;3*RYR!^N&z^%FueXvfh8N|Jj{(W}L~&(x;D=b3Y&b*nbx6efa+N*Vphp`^|2P7kf*q nPo7VdPj5|qIsfuYt`4+9zOj8jNpxF literal 0 HcmV?d00001 diff --git a/tests/fixtures/inspection_parity/f22.png b/tests/fixtures/inspection_parity/f22.png new file mode 100644 index 0000000000000000000000000000000000000000..f2d2e66cb427a38bd07d9e0ad72388ed8775fa63 GIT binary patch literal 1104 zcmbVLF^dyH6kbmSiNwmvDh^l(M@Vu>?zBowf-c50m{Wuh*U9c=cP_KLGM~$<}?zqtNHl*hgh_HMfs!hJ~=?loJS1!z1>HhcuAw5IZF5 zQlB$|)^mskd@RMRWW=QvU|&}VCuY_F3z8RjNF)hF7zvalP*O=DXaV=Mjp3X$eeAA6 z?@>p3P*yCI%7EYz>Uwe+53sa7OVG!Px~NI2DI|@y+1VknYz21ctWrqWfSMjE7T{im zKG4{Va-Awerr6iGoBxT625cNf)Abw+_aY*cbZPJ#DeyPDSg9SY$SnwQ;B>hJP(l)} zOej?p?$Cv0HLuoUR(fm;8P9=)?GZ-@U~dpcK4uEvAwfummN8#|*ax8)p!HxZ=Bn^e zP)R51Ob=BT*?_qCIJ>#XGVQ2EwWaEOMek@Ow*X*u&hFJ)O_E~H>>BEZJ+!d^pWs;! zsKOg-#dgdH{`2OzKSNA>p&2ICThqYTzUaVD4?`FQ7 ee|@>r{ntc4aZm8w(c31F8fI&|`R3Zgr+)wzL~vWhBOvQ_C4PBeee6;W_xwzT0VC!XIWOheZ6@H zf9v{lY7W0Ioc_LTS;sG_)_s+vB2JXGizLFWFmnfH$*dU}oVs4j6C`uB50xm@=oyi0twHX2})E+MroKx+XgwxbR z>Z=H03kG`*u~YEvK#Xk4ZOO&3VEdp5WxCr>H4}%BC^`W55Y=VG+aL_e5 zGIL#*n)7Ownwd9N7jbmj>b*v*3Hn%im!n>d4efBceFxVQ7 { diff --git a/tests/test_api_prompt_inspector.py b/tests/test_api_prompt_inspector.py index 40846da..f36c962 100644 --- a/tests/test_api_prompt_inspector.py +++ b/tests/test_api_prompt_inspector.py @@ -792,13 +792,17 @@ class TestPromptInspector(ImageAPITestCase): ) async def test_inspect_route_exposes_reason_code_and_blocking_node(self): - self.parity_fixture("f06a") + # f06a's encoder is supported since Task 2.4; an unknown encoder is not. + built = parity.build("f06a") + built["prompt"]["6"]["class_type"] = "FutureTextEncode" + parity.write_png(self.output_dir / "f06a.png", built) _, result = await self.inspect("f06a.png") resolution = result["resolution"] self.assertEqual(resolution["state"], "unresolved") self.assertEqual(resolution["reason_code"], "unsupported-node") - self.assertEqual(set(resolution["node"]), {"id", "class_type"}) - self.assertIsInstance(resolution["node"]["class_type"], str) + self.assertEqual( + resolution["node"], {"id": "6", "class_type": "FutureTextEncode"} + ) png(self.output_dir / "image.png", prompt="{broken") _, result = await self.inspect() self.assertEqual(result["resolution"]["reason_code"], "malformed") diff --git a/tests/test_prompt_pair_resolver.py b/tests/test_prompt_pair_resolver.py index ea1b684..6f21b1b 100644 --- a/tests/test_prompt_pair_resolver.py +++ b/tests/test_prompt_pair_resolver.py @@ -1652,6 +1652,446 @@ class TestConditioningSemantics(unittest.TestCase): self.assertIsNone(result["reason_code"]) +def _primitive(value: str) -> dict: + return {"class_type": "PrimitiveString", "inputs": {"value": value}} + + +def _text_encoded(link: list) -> dict: + """A CLIPTextEncode (node 6 in ``_with_positive``) reading a linked string.""" + return {"class_type": "CLIPTextEncode", "inputs": {"text": link, "clip": ["4", 1]}} + + +def _string_positive(node: dict, extra: dict = None) -> dict: + """``f02`` whose positive encoder reads string node 10 (``node``).""" + return _with_positive(_text_encoded(["10", 0]), {"10": node, **(extra or {})}) + + +def _multiline(text, flag) -> dict: + return { + "class_type": "StringConstantMultiline", + "inputs": {"string": text, "strip_newlines": flag}, + } + + +def _join(count, delimiter, return_list=False, **strings) -> dict: + return { + "class_type": "JoinStringMulti", + "inputs": { + "inputcount": count, + "delimiter": delimiter, + "return_list": return_list, + **strings, + }, + } + + +def _sd3(clip_l: str, clip_g: str, t5xxl: str) -> dict: + return { + "class_type": "CLIPTextEncodeSD3", + "inputs": { + "clip": ["4", 1], + "clip_l": clip_l, + "clip_g": clip_g, + "t5xxl": t5xxl, + "empty_padding": "none", + }, + } + + +def _positive(result: dict) -> tuple: + return result["state"], _role_text_of(result, "positive"), result["reason_code"] + + +def _role_text_of(result: dict, role: str): + side = result[role] + return side["text"] if isinstance(side, dict) else None + + +F01B_POSITIVE = f"{parity.POS}, Scene 0: unique shot description soft winter light," + + +class TestEncoderAndStringSemantics(unittest.TestCase): + """Text encoders, string sources and runtime nodes (inspection, Task 2.4).""" + + def test_sdxl_equal_parts_resolve(self): + result = _inspect(parity.build("f06a")["prompt"]) + self.assertEqual(result["state"], "resolved") + self.assertIsNone(result["reason_code"]) + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + provenance = result["positive"]["provenance"] + self.assertEqual(provenance["path"], [{"node_id": "6", "output_slot": 0}]) + self.assertEqual(provenance["text_source"], "graph-inferred") + self.assertEqual(result["positive"]["storage_text"], parity.POS) + # Linked parts are followed; equal texts still resolve, path deduplicated. + source = parity.build("f06a")["prompt"] + source["6"]["inputs"]["text_g"] = ["10", 0] + source["6"]["inputs"]["text_l"] = ["10", 0] + source["10"] = _primitive(parity.POS) + result = _inspect(source) + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + self.assertEqual( + result["positive"]["provenance"]["path"], + [{"node_id": "6", "output_slot": 0}, {"node_id": "10", "output_slot": 0}], + ) + + def test_sdxl_distinct_parts_are_composite(self): + source = parity.build("f06a")["prompt"] + source["6"]["inputs"]["text_l"] = "a second style" + result = _inspect(source) + self.assertEqual(result["state"], "unresolved") + self.assertEqual(result["reason_code"], "composite") + self.assertEqual(result["reason"], "Encoder parts have different texts") + self.assertEqual( + result["node"], {"id": "6", "class_type": "CLIPTextEncodeSDXL"} + ) + self.assertIsNone(result["positive"]) + # Whitespace is text: emptiness is exact, never trimmed into agreement. + source["6"]["inputs"]["text_l"] = " " + self.assertEqual(_inspect(source)["reason_code"], "composite") + source["6"]["inputs"]["text_l"] = parity.POS + " " + self.assertEqual(_inspect(source)["reason_code"], "composite") + + def test_sdxl_single_nonempty_part_resolves(self): + for empty in ("text_l", "text_g"): + with self.subTest(empty=empty): + source = parity.build("f06a")["prompt"] + source["6"]["inputs"][empty] = "" + result = _inspect(source) + self.assertEqual(result["state"], "resolved") + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + self.assertEqual(result["positive"]["storage_text"], parity.POS) + # A linked empty part counts as empty; both paths are recorded. + source = parity.build("f06a")["prompt"] + source["6"]["inputs"]["text_l"] = ["10", 0] + source["10"] = _primitive("") + result = _inspect(source) + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + self.assertEqual( + result["positive"]["provenance"]["path"], + [{"node_id": "6", "output_slot": 0}, {"node_id": "10", "output_slot": 0}], + ) + # All parts empty: an empty positive, never a resolved prompt. + source["6"]["inputs"]["text_g"] = "" + self.assertEqual(_inspect(source)["reason"], "Positive prompt is empty") + # A missing part is malformed, never read as empty. + source = parity.build("f06a")["prompt"] + del source["6"]["inputs"]["text_l"] + self.assertEqual(_inspect(source)["reason_code"], "malformed") + + def test_flux_and_sd3_encoders(self): + result = _inspect(parity.build("f06b")["prompt"]) + self.assertEqual(result["state"], "resolved") + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + source = parity.build("f06b")["prompt"] + source["6"]["inputs"]["t5xxl"] = "a longer natural language description" + result = _inspect(source) + self.assertEqual(result["reason_code"], "composite") + self.assertEqual(result["node"]["class_type"], "CLIPTextEncodeFlux") + cases = [ + ((parity.POS, parity.POS, parity.POS), ("resolved", parity.POS, None)), + (("", "", parity.POS), ("resolved", parity.POS, None)), + (("", parity.POS, ""), ("resolved", parity.POS, None)), + ((parity.POS, "", WOLF), ("unresolved", None, "composite")), + ((parity.POS, parity.POS, WOLF), ("unresolved", None, "composite")), + # Spec 7.2: only "all equal" or "exactly one non-empty" resolve. + ((parity.POS, parity.POS, ""), ("unresolved", None, "composite")), + ] + for parts, expected in cases: + with self.subTest(parts=parts): + self.assertEqual( + _positive(_inspect(_with_positive(_sd3(*parts)))), expected + ) + + def test_qwen_edit_plus_prompt(self): + extra = { + "13": {"class_type": "LoadImage", "inputs": {"image": "reference.png"}} + } + for name in ("TextEncodeQwenImageEditPlus", "TextEncodeQwenImageEdit"): + with self.subTest(node=name): + inputs = {"clip": ["4", 1], "prompt": parity.POS, "vae": ["4", 2]} + images = ("image1", "image2") if name.endswith("Plus") else ("image",) + inputs.update({image: ["13", 0] for image in images}) + node = {"class_type": name, "inputs": inputs} + result = _inspect(_with_positive(node, extra)) + self.assertEqual(result["state"], "resolved") + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + self.assertEqual( + result["positive"]["provenance"]["path"], + [{"node_id": "6", "output_slot": 0}], + ) + + def test_qwen_image21_two_outputs(self): + source = parity.build("f24")["prompt"] + result = _inspect(source) + self.assertEqual(result["state"], "resolved") + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + self.assertEqual( + result["negative"]["provenance"]["path"], + [{"node_id": "6", "output_slot": 1}], + ) + self.assertEqual(result["negative"]["storage_text"], parity.NEG) + # Slot 2 is the latent output, never a prompt. + source["3"]["inputs"]["negative"] = ["6", 2] + result = _inspect(source) + self.assertEqual(result["reason_code"], "unsupported-node") + self.assertEqual( + result["node"], {"id": "6", "class_type": "TextEncodeQwenImage21"} + ) + + def test_kiko_text_encode_plain_text(self): + node = {"class_type": "KikoTextEncode", "inputs": {"clip": ["4", 1]}} + node["inputs"]["text"] = parity.POS + result = _inspect(_with_positive(node)) + self.assertEqual(result["state"], "resolved") + self.assertEqual(_sides(result), (parity.POS, parity.NEG)) + self.assertEqual(result["positive"]["storage_text"], parity.POS) + # No composition: a linked text arrives unchanged. + node["inputs"]["text"] = ["10", 0] + extra = {"10": {"class_type": "PrimitiveStringMultiline"}} + extra["10"]["inputs"] = {"value": " " + parity.POS + "\n"} + result = _inspect(_with_positive(node, extra)) + self.assertEqual(result["positive"]["text"], " " + parity.POS + "\n") + + def test_string_constant_multiline_strip_newlines_exact(self): + result = _inspect(_string_positive(_multiline("a\n b \n", True))) + self.assertEqual(_positive(result), ("resolved", "a b", None)) + self.assertEqual(result["positive"]["storage_text"], "a b") + self.assertEqual( + result["positive"]["provenance"]["path"], + [{"node_id": "6", "output_slot": 0}, {"node_id": "10", "output_slot": 0}], + ) + result = _inspect(_string_positive(_multiline("a\n b \n", False))) + self.assertEqual(_positive(result), ("resolved", "a\n b \n", None)) + # Only "\n" is removed; other whitespace inside the text is kept. + result = _inspect(_string_positive(_multiline("\ta\r\n\tb\n\n", True))) + self.assertEqual(result["positive"]["text"], "a\r\tb") + # A linked text is followed; a linked flag is decided at run time. + linked = _string_positive( + _multiline(["11", 0], True), {"11": _primitive("x\ny")} + ) + self.assertEqual(_inspect(linked)["positive"]["text"], "xy") + flag = _string_positive( + _multiline("a\nb", ["12", 0]), + {"12": {"class_type": "PrimitiveBoolean", "inputs": {"value": True}}}, + ) + result = _inspect(flag) + self.assertEqual(result["reason_code"], "unsupported-node") + self.assertEqual( + result["node"], {"id": "10", "class_type": "StringConstantMultiline"} + ) + for value in ("true", 1, None): + with self.subTest(flag=value): + result = _inspect(_string_positive(_multiline("a\nb", value))) + self.assertEqual(result["reason_code"], "malformed") + missing = _multiline("a", True) + del missing["inputs"]["strip_newlines"] + self.assertEqual( + _inspect(_string_positive(missing))["reason_code"], "malformed" + ) + # StringConstant is the literal strategy: its string, unchanged. + constant = {"class_type": "StringConstant", "inputs": {"string": " x\n"}} + result = _inspect(_string_positive(constant)) + self.assertEqual(_positive(result), ("resolved", " x\n", None)) + + def test_join_string_multi_exact_semantics(self): + strings = { + "string_1": ["11", 0], + "string_2": "b", + "string_3": ["13", 0], + "string_4": ["14", 0], + "string_6": "beyond the count", + } + extra = { + "11": _primitive(""), + "13": _primitive(""), + "14": _primitive(" d "), + } + join = _join(5, "\\n|", **strings) + result = _inspect(_string_positive(join, extra)) + # string_1 is kept although empty; empty and missing later parts are + # skipped; inputs past ``inputcount`` are ignored; the delimiter is + # used exactly as written (no escape processing, no trimming). + self.assertEqual(_positive(result), ("resolved", "\\n|b\\n| d ", None)) + self.assertEqual(result["positive"]["storage_text"], "\\n|b\\n| d ") + self.assertEqual( + result["positive"]["provenance"]["path"], + [ + {"node_id": "6", "output_slot": 0}, + {"node_id": "10", "output_slot": 0}, + {"node_id": "11", "output_slot": 0}, + {"node_id": "13", "output_slot": 0}, + {"node_id": "14", "output_slot": 0}, + ], + ) + two = _join(2, " ", string_1="a red fox", string_2="in snow") + self.assertEqual( + _inspect(_string_positive(two))["positive"]["text"], "a red fox in snow" + ) + # Settings must be literal values the node would accept. + linked = {"12": {"class_type": "PrimitiveInt", "inputs": {"value": 2}}} + for field, value, code in ( + ("inputcount", ["12", 0], "unsupported-node"), + ("delimiter", ["12", 0], "unsupported-node"), + ("inputcount", 0, "malformed"), + ("inputcount", 1001, "malformed"), + ("inputcount", True, "malformed"), + ("inputcount", "2", "malformed"), + ("delimiter", 1, "malformed"), + ("return_list", "false", "malformed"), + ("return_list", 0, "malformed"), + ): + with self.subTest(field=field, value=value): + node = _join(2, " ", string_1="a", string_2="b") + node["inputs"][field] = value + result = _inspect(_string_positive(node, linked)) + self.assertEqual(result["reason_code"], code) + if code == "unsupported-node": + self.assertEqual(result["node"]["class_type"], "JoinStringMulti") + for field in ("inputcount", "delimiter", "return_list", "string_1"): + with self.subTest(missing=field): + node = _join(2, " ", string_1="a", string_2="b") + del node["inputs"][field] + result = _inspect(_string_positive(node)) + self.assertEqual(result["reason_code"], "malformed") + + def test_join_string_multi_return_list_is_runtime_value(self): + source = parity.build("f21")["prompt"] + result = _inspect(source) + self.assertEqual(result["state"], "unresolved") + self.assertEqual(result["reason_code"], "runtime-value") + self.assertEqual(result["reason"], "Text is decided at run time") + self.assertEqual(result["node"], {"id": "10", "class_type": "JoinStringMulti"}) + # A linked list flag is a run-time value too. + source["10"]["inputs"]["return_list"] = ["20", 0] + source["20"] = {"class_type": "PrimitiveBoolean", "inputs": {"value": False}} + self.assertEqual(_inspect(source)["reason_code"], "runtime-value") + # With the flag literally false the same parts join. + source["10"]["inputs"]["return_list"] = False + self.assertEqual(_inspect(source)["positive"]["text"], "a red fox, in snow") + + def test_prompt_search_list_is_runtime_value(self): + from utils.prompt_inspector import inspect_prompt_evidence + + source = parity.build("f22")["prompt"] + result = _inspect(source) + self.assertEqual(result["state"], "unresolved") + self.assertEqual(result["reason_code"], "runtime-value") + self.assertEqual(result["node"], {"id": "10", "class_type": "PromptSearchList"}) + self.assertIsNone(result["positive"]) + # Never a list item: every output slot is decided at run time. + source["6"]["inputs"]["text"] = ["10", 3] + self.assertEqual(_inspect(source)["reason_code"], "runtime-value") + evidence = inspect_prompt_evidence({"prompt": parity.build("f22")["prompt"]}) + self.assertEqual(evidence["prompts"]["positive"]["status"], "unresolved") + self.assertIsNone(evidence["prompts"]["positive"]["text"]) + + @unittest.expectedFailure # needs Task 2.5 (EmptyFlux2LatentImage origin) + def test_f01b_single_flux2_branch_resolves_with_absent_negative(self): + result = _inspect(parity.build("f01b")["prompt"]) + self.assertEqual(result["state"], "absent") + self.assertEqual(result["positive"]["text"], F01B_POSITIVE) + self.assertIsNone(result["negative"]) + + def test_f01b_branch_text_is_the_exact_join(self): + # Until Task 2.5 the flux2 latent origin is unsupported, so this swaps + # it for an EmptyLatentImage and anchors at the branch's saver: the + # join, multiline strings and encoder alone must give the exact text. + source = parity.build("f01b")["prompt"] + source["1300:7"] = { + "class_type": "EmptyLatentImage", + "inputs": {"width": 1024, "height": 1024, "batch_size": 1}, + } + result = _inspect(source, output_node_id="1400") + self.assertEqual(result["state"], "absent") + self.assertEqual(result["reason"], "Guider has no negative input") + self.assertEqual(result["positive"]["text"], F01B_POSITIVE) + self.assertEqual(result["positive"]["storage_text"], F01B_POSITIVE) + self.assertIsNone(result["negative"]) + + def test_encoder_and_string_outputs_keep_their_kinds(self): + sdxl = parity.build("f06a")["prompt"]["6"] + # A conditioning output read as a string, and a string output read as + # conditioning, are malformed links, not unsupported nodes. + join = _join(2, " ", string_1=["11", 0], string_2="b") + result = _inspect(_string_positive(join, {"11": sdxl})) + self.assertEqual(result["reason_code"], "malformed") + self.assertEqual(result["reason"], "Output slot has an incompatible type") + for node in (_multiline("a", True), _join(2, " ", string_1="a")): + with self.subTest(node=node["class_type"]): + result = _inspect(_with_positive(node)) + self.assertEqual(result["reason_code"], "malformed") + + def test_unverified_text_nodes_stay_unsupported_and_named(self): + for name in ("LoRAfromText", "RegexReplace", "XYZPrompt", "TextBoxMira"): + with self.subTest(node=name): + node = {"class_type": name, "inputs": {"text": parity.POS}} + result = _inspect(_string_positive(node)) + self.assertEqual(result["reason_code"], "unsupported-node") + self.assertEqual(result["node"], {"id": "10", "class_type": name}) + + def test_string_strategies_bound_every_intermediate(self): + limited = replace(resolver.INSPECTION_PROFILE, max_text_length=10) + join = _join(2, " ", string_1="abcdef", string_2="ghijk") + result = resolve_prompt_pair( + {"prompt": _string_positive(join)}, profile=limited + ) + self.assertEqual(result["reason_code"], "text-limit") + # Exactly at the limit is fine. + join["inputs"]["string_2"] = "ghi" + result = resolve_prompt_pair( + {"prompt": _string_positive(join)}, profile=limited + ) + self.assertEqual(result["positive"]["text"], "abcdef ghi") + # The scanned-character budget covers stripping, joining and comparing. + long_text = "x\n" * 5_000 + cases = { + "strip": _string_positive(_multiline(long_text, True)), + "join": _string_positive( + _join(2, " ", string_1=long_text, string_2=long_text) + ), + "parts": _with_positive(_sd3(long_text, long_text, long_text)), + } + scanned = replace(resolver.INSPECTION_PROFILE, max_scanned_units=8_000) + for label, source in cases.items(): + with self.subTest(strategy=label): + self.assertEqual(_inspect(source)["state"], "resolved") + result = resolve_prompt_pair({"prompt": source}, profile=scanned) + self.assertEqual(result["reason_code"], "work-limit") + + def test_listed_strategies_without_code_here_are_unsupported(self): + # ``pm-compose`` and ``concat`` are schema-valid but implemented by the + # existing adapters only; a table row using them is never guessed at. + table = inspection_semantics.load_semantics() + encoders = dict(table.encoders) + encoders[("CLIPTextEncodeSDXL", 0)] = { + "strategy": "pm-compose", + "inputs": ("text",), + } + strings = dict(table.strings) + strings[("StringConstant", 0)] = {"strategy": "concat", "inputs": ("string",)} + patched = replace(table, encoders=encoders, strings=strings) + constant = {"class_type": "StringConstant", "inputs": {"string": "x"}} + with mock.patch.object(resolver, "_inspection_semantics", return_value=patched): + for source, name in ( + (parity.build("f06a")["prompt"], "CLIPTextEncodeSDXL"), + (_string_positive(constant), "StringConstant"), + ): + with self.subTest(node=name): + result = _inspect(source) + self.assertEqual(result["reason_code"], "unsupported-node") + self.assertEqual(result["node"]["class_type"], name) + + def test_default_profile_keeps_encoder_and_string_nodes_unsupported(self): + for name in ("f06a", "f06b", "f21", "f22", "f24"): + with self.subTest(fixture=name): + result = resolve_prompt_pair({"prompt": parity.build(name)["prompt"]}) + self.assertEqual(result["state"], "unresolved") + self.assertEqual( + result["reason"], "Unsupported node semantics or output slot" + ) + self.assertIsNone(result["reason_code"]) + + def _detailed(same_prompt: bool = False) -> dict: """``f19``: KSampler 3 -> VAEDecode 8 -> FaceDetailer 50 -> SaveImage 9.""" source = parity.build("f19")["prompt"] diff --git a/utils/prompt_pair_resolver.py b/utils/prompt_pair_resolver.py index cf8d8e8..3a11ead 100644 --- a/utils/prompt_pair_resolver.py +++ b/utils/prompt_pair_resolver.py @@ -148,6 +148,11 @@ _GUIDER_ABSENT_REASON = "Guider has no negative input" _ZEROED_SOURCE = "zeroed-conditioning" # Runtime selector choices expanded per sampler (the spec's candidate cap). _MAX_BRANCH_CANDIDATES = 32 +# Statically undecidable text (a search, a list output): never pick an item. +_RUNTIME_REASON = "Text is decided at run time" +_INCOMPATIBLE_SLOT = "Output slot has an incompatible type" +# JoinStringMulti's own ``inputcount`` range; anything else never executed. +_MAX_JOIN_INPUTS = 1000 _SAMPLER_LIKE_INPUTS = ("latent_image", "guider", "positive", "negative") _NO_SAMPLER_NAMES = 3 _CLASS_NAME_LENGTH = 64 @@ -427,9 +432,7 @@ class _GraphResolver: elif (name, slot) in self.adapters: adapter = self.adapters[(name, slot)] if not isinstance(adapter, OutputAdapter) or adapter.kind != kind: - raise _Unresolved( - "Output slot has an incompatible type", "malformed" - ) + raise _Unresolved(_INCOMPATIBLE_SLOT, "malformed") evidence = ( self.execution_record(node_id, inputs) if name == "PromptManagerPair" @@ -548,6 +551,10 @@ class _GraphResolver: def follows(self, name: str, slot: int, kind: str) -> bool: """Whether the semantics table traces this output (inspection only).""" semantics = self.semantics + if name in semantics.runtime: + return True + if (name, slot) in semantics.encoders or (name, slot) in semantics.strings: + return True if slot == 0 and name in semantics.selectors: return True return kind == "conditioning" and ( @@ -558,11 +565,23 @@ class _GraphResolver: def semantic( self, node_id: str, name: str, slot: int, kind: str, inputs: dict, depth: int ) -> Tuple[str, list, str]: - """Trace a table-listed pass-through, zeroed, selector or composite output. + """Trace a table-listed output: runtime, encoder, string, pass-through, + zeroed, selector or composite. The caller (``text``) prepends this node to the returned path. """ semantics = self.semantics + if name in semantics.runtime: + raise self.named(_RUNTIME_REASON, "runtime-value", node_id) + for table, expected, build in ( + (semantics.encoders, "conditioning", self.encoded), + (semantics.strings, "string", self.stringed), + ): + listed = table.get((name, slot)) + if listed is not None: + if kind != expected: + raise _Unresolved(_INCOMPATIBLE_SLOT, "malformed") + return build(node_id, listed, inputs, depth) entry = ( semantics.conditioning.get((name, slot)) if kind == "conditioning" else None ) @@ -621,9 +640,15 @@ class _GraphResolver: def merged(self, node_id: str, slot: int, parts: list) -> Tuple[str, list, str]: """One text reached through several followed paths (deduplicated).""" - text = parts[0][0] - storages = {part[2] for part in parts} - storage_text = parts[0][2] if len(storages) == 1 else text + text, merged, storage_text = self.joined_paths(parts, parts) + self.inherit_source(node_id, slot, [part[1] for part in parts]) + return text, merged, storage_text + + def joined_paths(self, chosen: list, parts: list) -> Tuple[str, list, str]: + """``chosen``'s one text with every part's followed path, deduplicated.""" + text = chosen[0][0] + storages = {part[2] for part in chosen} + storage_text = chosen[0][2] if len(storages) == 1 else text paths = [part[1] for part in parts] self.scan(sum(len(path) for path in paths)) merged, seen = [], set() @@ -633,9 +658,119 @@ class _GraphResolver: if identity not in seen: seen.add(identity) merged.append(step) - self.inherit_source(node_id, slot, paths) return text, merged, storage_text + def encoded( + self, node_id: str, entry: Mapping, inputs: dict, depth: int + ) -> Tuple[str, list, str]: + """A text encoder's prompt (``single`` or ``parts-equal``). + + ``single`` reads one string input unchanged; image, VAE and other inputs + are not text. ``parts-equal`` reads every listed part: identical parts + are that text, exactly one non-empty part (exact ``""`` emptiness) is + that text, and anything else is several texts in effect at once + (``composite``, D6). Like ``CLIPTextEncode``, an encoder's text source + stays ``graph-inferred``. + """ + names = entry["inputs"] + if entry["strategy"] == "single": + return self.text(inputs.get(names[0]), "string", depth + 1) + if entry["strategy"] != "parts-equal": + raise self.unsupported("Unsupported encoder strategy", node_id) + parts = [self.text(inputs.get(name), "string", depth + 1) for name in names] + # Comparing the parts reads every character once. + self.scan(sum(len(part[0]) for part in parts)) + if len({part[0] for part in parts}) == 1: + return self.joined_paths(parts, parts) + filled = [part for part in parts if part[0] != ""] + if len(filled) == 1: + return self.joined_paths(filled, parts) + raise self.named("Encoder parts have different texts", "composite", node_id) + + def stringed( + self, node_id: str, entry: Mapping, inputs: dict, depth: int + ) -> Tuple[str, list, str]: + """A string source: ``literal``, ``strip-newlines`` or ``join-multi``. + + Each reproduces the pinned node's code exactly, and only from literal + settings; a linked setting is never guessed. + """ + strategy = entry["strategy"] + source = entry["inputs"][0] + if strategy == "literal": + return self.text(inputs.get(source), "string", depth + 1) + if strategy == "strip-newlines": + # KJNodes StringConstantMultiline: replace("\n", "").strip(). + flag = self.setting(node_id, inputs, entry["params"]["flag"], bool) + text, path, storage_text = self.text( + inputs.get(source), "string", depth + 1 + ) + if not flag: + return text, path, storage_text + self.scan(len(text)) + text = self.built(self.bounded(text.replace("\n", "").strip())) + return text, path, text + if strategy == "join-multi": + return self.join_multi(node_id, entry, inputs, depth) + raise self.unsupported("Unsupported string strategy", node_id) + + def setting(self, node_id: str, inputs: dict, name: str, kind: type) -> Any: + """A literal node setting; linked is ``unsupported-node``, else malformed.""" + if name not in inputs: + raise _Unresolved("Missing string node setting", "malformed") + value = inputs[name] + if _link(value): + raise self.unsupported("Linked string node setting", node_id) + if type(value) is not kind: + raise _Unresolved("Malformed string node setting", "malformed") + return value + + def join_multi( + self, node_id: str, entry: Mapping, inputs: dict, depth: int + ) -> Tuple[str, list, str]: + """KJNodes JoinStringMulti with a literal ``return_list`` of false. + + ``string_1`` is always kept (even empty); ``string_2`` .. ``string_N`` + (``N`` = ``inputcount``) are appended after the literal delimiter unless + missing or exactly ``""``; inputs past ``N`` are ignored. A list output + (or a linked flag) is decided at run time. + """ + params = entry["params"] + flag = params["list_flag"] + if flag not in inputs: + raise _Unresolved("Missing string node setting", "malformed") + if _link(inputs[flag]) or inputs[flag] is True: + raise self.named(_RUNTIME_REASON, "runtime-value", node_id) + if inputs[flag] is not False: + raise _Unresolved("Malformed string node setting", "malformed") + count = self.setting(node_id, inputs, params["count"], int) + delimiter = self.setting(node_id, inputs, params["delimiter"], str) + if not 1 <= count <= _MAX_JOIN_INPUTS: + raise _Unresolved("Malformed string node setting", "malformed") + self.scan(count) + names = [entry["inputs"][0]] + [ + f"{params['input_prefix']}{index}" for index in range(2, count + 1) + ] + parts: list = [] + path: list = [] + size = 0 + for position, input_name in enumerate(names): + if position and input_name not in inputs: + continue + part, followed, _ = self.text(inputs.get(input_name), "string", depth + 1) + self.scan(len(part) + len(followed)) + path.extend(followed) + if position and not part: + continue + size += len(part) + (len(delimiter) if parts else 0) + if size > self.max_text_length: + raise _Unresolved( + "Resolved text exceeds the length limit", "text-limit" + ) + parts.append(part) + text = self.built(delimiter.join(parts)) + return text, path, text + def source_of(self, path: list) -> str: if not path: return "graph-inferred"