2403 lines
104 KiB
Python
2403 lines
104 KiB
Python
"""Conservative prompt-pair resolution contracts, independent of ComfyUI."""
|
|
|
|
import random
|
|
import subprocess # nosec B404 - fixed interpreter command, no shell
|
|
import sys
|
|
import tempfile
|
|
import unittest
|
|
from dataclasses import replace
|
|
from pathlib import Path
|
|
from unittest import mock
|
|
|
|
from utils import inspection_semantics
|
|
from utils import prompt_pair_resolver as resolver
|
|
from utils.prompt_pair_resolver import OutputAdapter, resolve_prompt_pair
|
|
|
|
try:
|
|
from tests import inspection_parity as parity
|
|
except ImportError: # discovered with ``-s tests``
|
|
import inspection_parity as parity
|
|
|
|
|
|
def graph(positive: str = "portrait", negative: str = "blurry") -> dict:
|
|
return {
|
|
"1": {"class_type": "CLIPTextEncode", "inputs": {"text": positive}},
|
|
"2": {"class_type": "CLIPTextEncode", "inputs": {"text": negative}},
|
|
"3": {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["1", 0], "negative": ["2", 0]},
|
|
},
|
|
}
|
|
|
|
|
|
class TestPromptPairResolver(unittest.TestCase):
|
|
def test_resolves_same_sampler_and_preserves_output_provenance(self):
|
|
result = resolve_prompt_pair({"prompt": graph()})
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(result["positive"]["text"], "portrait")
|
|
self.assertEqual(result["negative"]["text"], "blurry")
|
|
self.assertEqual(result["positive"]["provenance"]["node_id"], "1")
|
|
self.assertEqual(result["positive"]["provenance"]["output_slot"], 0)
|
|
self.assertEqual(result["sampler_id"], "3")
|
|
|
|
def test_explicit_empty_negative_is_resolved_but_missing_is_absent(self):
|
|
self.assertEqual(resolve_prompt_pair(graph(negative=""))["state"], "resolved")
|
|
self.assertEqual(
|
|
resolve_prompt_pair(graph(negative=""))["negative"]["text"], ""
|
|
)
|
|
source = graph()
|
|
del source["3"]["inputs"]["negative"]
|
|
result = resolve_prompt_pair(source)
|
|
self.assertEqual(result["state"], "absent")
|
|
self.assertEqual(result["positive"]["text"], "portrait")
|
|
self.assertIsNone(result["negative"])
|
|
|
|
def test_multiple_samplers_are_ambiguous_even_when_texts_match(self):
|
|
source = graph()
|
|
source["4"] = {
|
|
"class_type": "KSamplerAdvanced",
|
|
"inputs": source["3"]["inputs"],
|
|
}
|
|
result = resolve_prompt_pair(source)
|
|
self.assertEqual(result["state"], "ambiguous")
|
|
self.assertEqual(len(result["candidates"]), 2)
|
|
self.assertIsNone(result["positive"])
|
|
|
|
def test_saver_anchor_selects_only_its_sampler(self):
|
|
source = graph()
|
|
source.update(
|
|
{
|
|
"4": {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["2", 0], "negative": ["1", 0]},
|
|
},
|
|
"5": {"class_type": "VAEDecode", "inputs": {"samples": ["4", 0]}},
|
|
"6": {"class_type": "SaveImage", "inputs": {"images": ["5", 0]}},
|
|
}
|
|
)
|
|
result = resolve_prompt_pair(source, output_node_id="6")
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(result["sampler_id"], "4")
|
|
self.assertEqual(result["positive"]["text"], "blurry")
|
|
|
|
def test_unknown_anchor_transform_and_multistage_are_unresolved(self):
|
|
source = graph()
|
|
source["5"] = {
|
|
"class_type": "MysteryImageTransform",
|
|
"inputs": {"images": ["3", 0]},
|
|
}
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source, output_node_id="5")["state"], "unresolved"
|
|
)
|
|
source["4"] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {**source["3"]["inputs"], "latent_image": ["3", 0]},
|
|
}
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source, output_node_id="4")["state"], "unresolved"
|
|
)
|
|
|
|
def test_combines_unknown_encoders_and_sdxl_channels_are_not_guessed(self):
|
|
for name in ("ConditioningCombine", "CLIPTextEncodeSDXL", "UnknownEncoder"):
|
|
with self.subTest(name=name):
|
|
source = graph()
|
|
source["1"] = {
|
|
"class_type": name,
|
|
"inputs": {
|
|
"text": "guess",
|
|
"text_g": "global",
|
|
"text_l": "local",
|
|
"conditioning_1": ["2", 0],
|
|
},
|
|
}
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_prompt_manager_composition_and_string_slot_are_resolved(self):
|
|
source = graph()
|
|
source["8"] = {
|
|
"class_type": "PromptManagerSplit",
|
|
"inputs": {
|
|
"text": "portrait",
|
|
"prepend_text": " trigger ",
|
|
"append_text": " detail ",
|
|
},
|
|
}
|
|
source["1"]["inputs"]["text"] = ["8", 1]
|
|
result = resolve_prompt_pair(source)
|
|
self.assertEqual(result["positive"]["text"], "trigger portrait detail")
|
|
self.assertIn(
|
|
{"node_id": "8", "output_slot": 1}, result["positive"]["provenance"]["path"]
|
|
)
|
|
source["3"]["inputs"]["positive"] = ["8", 0]
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source)["positive"]["text"], "trigger portrait detail"
|
|
)
|
|
source["1"]["inputs"]["text"] = ["8", 0]
|
|
source["3"]["inputs"]["positive"] = ["1", 0]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_concat_and_reroute_resolve_safe_linked_strings(self):
|
|
source = graph()
|
|
source["8"] = {"class_type": "PrimitiveString", "inputs": {"value": "portrait"}}
|
|
source["9"] = {
|
|
"class_type": "StringConcatenate",
|
|
"inputs": {"string_a": ["8", 0], "string_b": ["8", 0], "delimiter": ", "},
|
|
}
|
|
source["10"] = {"class_type": "Reroute", "inputs": {"value": ["9", 0]}}
|
|
source["1"]["inputs"]["text"] = ["10", 0]
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source)["positive"]["text"], "portrait, portrait"
|
|
)
|
|
|
|
def test_slot_adapter_does_not_conflate_two_outputs_of_one_node(self):
|
|
source = graph()
|
|
source["8"] = {
|
|
"class_type": "FuturePair",
|
|
"inputs": {"first": "positive", "second": "negative"},
|
|
}
|
|
source["3"]["inputs"] = {"positive": ["8", 0], "negative": ["8", 1]}
|
|
adapters = {
|
|
("FuturePair", 0): OutputAdapter("first", "conditioning"),
|
|
("FuturePair", 1): OutputAdapter("second", "conditioning"),
|
|
}
|
|
result = resolve_prompt_pair(source, adapters=adapters)
|
|
self.assertEqual(result["positive"]["text"], "positive")
|
|
self.assertEqual(result["negative"]["text"], "negative")
|
|
self.assertNotEqual(
|
|
result["positive"]["provenance"]["output_slot"],
|
|
result["negative"]["provenance"]["output_slot"],
|
|
)
|
|
|
|
def test_prompt_manager_text_has_only_its_real_string_output_zero(self):
|
|
source = graph()
|
|
source["8"] = {
|
|
"class_type": "PromptManagerText",
|
|
"inputs": {"text": "text node", "prepend_text": "before"},
|
|
}
|
|
source["1"]["inputs"]["text"] = ["8", 0]
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source)["positive"]["text"], "before text node"
|
|
)
|
|
source["1"]["inputs"]["text"] = ["8", 1]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
source["3"]["inputs"]["positive"] = ["8", 0]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_malformed_class_names_and_unknown_samplers_do_not_claim_absence(self):
|
|
for name in ([], {}, None, "UnknownSampler"):
|
|
source = graph()
|
|
source["3"]["class_type"] = name
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_pair_adapter_composes_only_positive_and_retains_recipe_text(self):
|
|
source = graph()
|
|
source["8"] = {
|
|
"class_type": "PromptManagerPair",
|
|
"inputs": {
|
|
"text": "positive",
|
|
"negative_text": "negative",
|
|
"prepend_text": "trigger",
|
|
"append_text": "detail",
|
|
},
|
|
}
|
|
source["3"]["inputs"] = {"positive": ["8", 0], "negative": ["8", 1]}
|
|
result = resolve_prompt_pair(source)
|
|
self.assertEqual(result["positive"]["text"], "trigger positive detail")
|
|
self.assertEqual(result["positive"]["storage_text"], "positive")
|
|
self.assertEqual(result["negative"]["text"], "negative")
|
|
self.assertEqual(result["negative"]["storage_text"], "negative")
|
|
source["1"]["inputs"]["text"] = ["8", 2]
|
|
source["2"]["inputs"]["text"] = ["8", 3]
|
|
source["3"]["inputs"] = {"positive": ["1", 0], "negative": ["2", 0]}
|
|
result = resolve_prompt_pair(source)
|
|
self.assertEqual(result["positive"]["storage_text"], "positive")
|
|
self.assertEqual(result["negative"]["text"], "negative")
|
|
source["3"]["inputs"]["negative"] = ["8", 3]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_linked_recipe_and_ordinary_clip_storage_values_are_explicit(self):
|
|
source = graph()
|
|
result = resolve_prompt_pair(source)
|
|
self.assertEqual(result["positive"]["storage_text"], "portrait")
|
|
source["8"] = {
|
|
"class_type": "PrimitiveString",
|
|
"inputs": {"value": "linked main"},
|
|
}
|
|
source["1"] = {
|
|
"class_type": "PromptManager",
|
|
"inputs": {"text": ["8", 0], "prepend_text": "before"},
|
|
}
|
|
result = resolve_prompt_pair(source)
|
|
self.assertEqual(result["positive"]["text"], "before linked main")
|
|
self.assertEqual(result["positive"]["storage_text"], "linked main")
|
|
a1111 = resolve_prompt_pair(
|
|
{
|
|
"parameters": (
|
|
"portrait\nNegative prompt: blurry\n" "Steps: 20, Sampler: Euler"
|
|
)
|
|
}
|
|
)
|
|
self.assertEqual(a1111["positive"]["storage_text"], "portrait")
|
|
self.assertEqual(a1111["negative"]["storage_text"], "blurry")
|
|
|
|
def test_cycles_depth_visits_and_text_growth_are_bounded(self):
|
|
source = graph()
|
|
source["8"] = {"class_type": "Reroute", "inputs": {"value": ["8", 0]}}
|
|
source["1"]["inputs"]["text"] = ["8", 0]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
self.assertEqual(
|
|
resolve_prompt_pair(graph(), max_visits=1)["state"], "unresolved"
|
|
)
|
|
self.assertEqual(
|
|
resolve_prompt_pair(graph(), max_depth=0)["state"], "unresolved"
|
|
)
|
|
self.assertEqual(
|
|
resolve_prompt_pair(graph(), max_text_length=3)["state"], "unresolved"
|
|
)
|
|
|
|
def test_malformed_metadata_and_links_never_raise(self):
|
|
for value in (
|
|
None,
|
|
[],
|
|
"raw text",
|
|
{},
|
|
{"prompt": []},
|
|
{"prompt": {"1": None}},
|
|
):
|
|
with self.subTest(value=value):
|
|
self.assertIn(
|
|
resolve_prompt_pair(value)["state"], ("absent", "unresolved")
|
|
)
|
|
for link in (["missing", 0], ["1", -1], ["1", True], ["1", 17], ["1"], 5):
|
|
source = graph()
|
|
source["3"]["inputs"]["positive"] = link
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_a1111_explicit_parameters_and_empty_negative(self):
|
|
for negative in ("bad anatomy\nblurry", ""):
|
|
result = resolve_prompt_pair(
|
|
{
|
|
"parameters": (
|
|
f"portrait\nNegative prompt: {negative}\n"
|
|
"Steps: 20, Sampler: Euler, Seed: 1"
|
|
)
|
|
}
|
|
)
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(result["negative"]["text"], negative)
|
|
self.assertEqual(result["positive"]["provenance"]["format"], "a1111")
|
|
self.assertEqual(
|
|
resolve_prompt_pair({"parameters": "portrait\nSteps: 20, Sampler: Euler"})[
|
|
"state"
|
|
],
|
|
"absent",
|
|
)
|
|
self.assertEqual(
|
|
resolve_prompt_pair({"parameters": "unstructured arbitrary text"})["state"],
|
|
"unresolved",
|
|
)
|
|
|
|
def test_was_concat_preserves_configured_whitespace_and_rejects_bad_delimiters(
|
|
self,
|
|
):
|
|
source = graph()
|
|
source["8"] = {
|
|
"class_type": "Text Concatenate",
|
|
"inputs": {
|
|
"text_a": " portrait ",
|
|
"text_b": "",
|
|
"text_c": "studio",
|
|
"delimiter": ", ",
|
|
},
|
|
}
|
|
source["1"]["inputs"]["text"] = ["8", 0]
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source)["positive"]["text"], "portrait, studio"
|
|
)
|
|
source["8"]["inputs"]["clean_whitespace"] = "false"
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source)["positive"]["text"], " portrait , studio"
|
|
)
|
|
source["8"]["inputs"]["delimiter"] = []
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
source["8"]["inputs"] = {1: "bad key"}
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_valid_empty_latent_and_reroute_but_not_unknown_or_cyclic_latents(self):
|
|
source = graph()
|
|
source["8"] = {"class_type": "EmptyLatentImage", "inputs": {}}
|
|
source["9"] = {"class_type": "Reroute", "inputs": {"value": ["8", 0]}}
|
|
source["3"]["inputs"]["latent_image"] = ["9", 0]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "resolved")
|
|
source["9"]["inputs"]["value"] = ["9", 0]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
source["3"]["inputs"]["latent_image"] = ["8", 0]
|
|
source["8"]["class_type"] = "UnknownLatentTransform"
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
source["3"]["inputs"]["latent_image"] = None
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_cyclic_or_invalid_output_anchors_never_guess(self):
|
|
source = graph()
|
|
source["8"] = {"class_type": "Reroute", "inputs": {"value": ["8", 0]}}
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source, output_node_id="8")["state"], "unresolved"
|
|
)
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source, output_node_id="missing")["state"], "unresolved"
|
|
)
|
|
source["8"] = {"class_type": "SaveImage", "inputs": {"images": ["3", 1]}}
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source, output_node_id="8")["state"], "unresolved"
|
|
)
|
|
source["8"] = {"class_type": "SaveImage", "inputs": {"images": ["3", 0]}}
|
|
self.assertEqual(
|
|
resolve_prompt_pair(source, output_node_id="8", max_depth=1)["state"],
|
|
"unresolved",
|
|
)
|
|
|
|
def test_malformed_reroute_empty_positive_and_duplicate_string_ids(self):
|
|
source = graph(positive="")
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
source = graph()
|
|
source["8"] = {
|
|
"class_type": "Reroute",
|
|
"inputs": {"value": ["1", 0], "input": ["2", 0]},
|
|
}
|
|
source["3"]["inputs"]["positive"] = ["8", 0]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
source[3] = source["3"]
|
|
self.assertEqual(resolve_prompt_pair(source)["state"], "unresolved")
|
|
|
|
def test_a1111_malformed_ambiguous_and_bounded_parameters(self):
|
|
self.assertEqual(resolve_prompt_pair({"parameters": None})["state"], "absent")
|
|
self.assertEqual(
|
|
resolve_prompt_pair({"parameters": "x" * 20}, max_text_length=10)["state"],
|
|
"unresolved",
|
|
)
|
|
self.assertEqual(
|
|
resolve_prompt_pair(
|
|
{"parameters": "\nNegative prompt: bad\nSteps: 20, Sampler: Euler"}
|
|
)["state"],
|
|
"unresolved",
|
|
)
|
|
self.assertEqual(
|
|
resolve_prompt_pair(
|
|
{
|
|
"parameters": (
|
|
"portrait\nNegative prompt: bad\nNegative prompt: worse\n"
|
|
"Steps: 20, Sampler: Euler"
|
|
)
|
|
}
|
|
)["state"],
|
|
"ambiguous",
|
|
)
|
|
self.assertEqual(
|
|
resolve_prompt_pair(
|
|
{
|
|
"parameters": (
|
|
"portrait\nNegative prompt: bad\n"
|
|
"Steps: 20, Sampler: Euler\nextra text"
|
|
)
|
|
}
|
|
)["state"],
|
|
"unresolved",
|
|
)
|
|
|
|
def test_exponential_concat_text_is_bounded_before_building_a_large_string(self):
|
|
source = graph()
|
|
source["8"] = {"class_type": "PrimitiveString", "inputs": {"value": "xx"}}
|
|
for index in range(9, 19):
|
|
source[str(index)] = {
|
|
"class_type": "StringConcatenate",
|
|
"inputs": {
|
|
"string_a": [str(index - 1), 0],
|
|
"string_b": [str(index - 1), 0],
|
|
},
|
|
}
|
|
source["1"]["inputs"]["text"] = ["18", 0]
|
|
result = resolve_prompt_pair(source, max_text_length=100)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertIn("length limit", result["reason"])
|
|
self.assertEqual(
|
|
resolve_prompt_pair(graph(), max_depth=1)["state"], "unresolved"
|
|
)
|
|
|
|
def test_api_graph_is_authoritative_over_unrelated_parameters(self):
|
|
source = graph()
|
|
source["1"]["class_type"] = "UnknownEncoder"
|
|
result = resolve_prompt_pair(
|
|
{
|
|
"prompt": source,
|
|
"parameters": (
|
|
"fallback\nNegative prompt: wrong\n" "Steps: 20, Sampler: Euler"
|
|
),
|
|
}
|
|
)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
|
|
|
|
def _f03_metadata() -> dict:
|
|
return {"prompt": parity.build("f03")["prompt"]}
|
|
|
|
|
|
def _vector(name: str) -> dict:
|
|
return next(v for v in parity.load_vectors() if v["name"] == name)["expect"]
|
|
|
|
|
|
def _diamond(prefix: str = "d", leaves: int = 575, middles: int = 24) -> dict:
|
|
"""A ~600-node string diamond: 24 fans over 575 leaves, one leaf shared."""
|
|
letters = "abcdefghijklmnopqrstuvwxyz"
|
|
nodes = {
|
|
f"{prefix}l{index}": {"class_type": "PrimitiveString", "inputs": {"value": "a"}}
|
|
for index in range(leaves)
|
|
}
|
|
for middle in range(middles):
|
|
nodes[f"{prefix}m{middle}"] = {
|
|
"class_type": "Text Concatenate",
|
|
"inputs": {
|
|
f"text_{letters[index]}": [
|
|
f"{prefix}l{(middle * middles + index) % leaves}",
|
|
0,
|
|
]
|
|
for index in range(middles)
|
|
},
|
|
}
|
|
nodes[f"{prefix}r"] = {
|
|
"class_type": "Text Concatenate",
|
|
"inputs": {
|
|
f"text_{letters[index]}": [f"{prefix}m{index}", 0]
|
|
for index in range(middles)
|
|
},
|
|
}
|
|
return nodes
|
|
|
|
|
|
def _diamond_samplers(count: int) -> dict:
|
|
"""``count`` samplers, each encoding the same diamond through its own encoder."""
|
|
source = {"neg": {"class_type": "CLIPTextEncode", "inputs": {"text": "bad"}}}
|
|
source.update(_diamond())
|
|
for index in range(count):
|
|
source[f"e{index}"] = {
|
|
"class_type": "CLIPTextEncode",
|
|
"inputs": {"text": ["dr", 0]},
|
|
}
|
|
source[f"s{index}"] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": [f"e{index}", 0], "negative": ["neg", 0]},
|
|
}
|
|
return source
|
|
|
|
|
|
def _distinct_diamond_samplers(count: int, chain: int = 60) -> dict:
|
|
"""``count`` samplers, each with its own 590-node diamond, sharing a latent chain.
|
|
|
|
Real work exceeds the shared 20,000-visit total while the graph stays under
|
|
the node ceiling: every candidate re-walks the (unmemoized) latent chain.
|
|
"""
|
|
source = {
|
|
"neg": {"class_type": "CLIPTextEncode", "inputs": {"text": "bad"}},
|
|
"z0": {"class_type": "EmptyLatentImage", "inputs": {}},
|
|
}
|
|
for index in range(1, chain + 1):
|
|
source[f"z{index}"] = {
|
|
"class_type": "Reroute",
|
|
"inputs": {"value": [f"z{index - 1}", 0]},
|
|
}
|
|
for index in range(count):
|
|
prefix = f"d{index}_"
|
|
source.update(_diamond(prefix, leaves=565))
|
|
source[f"e{index}"] = {
|
|
"class_type": "CLIPTextEncode",
|
|
"inputs": {"text": [f"{prefix}r", 0]},
|
|
}
|
|
source[f"s{index}"] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {
|
|
"positive": [f"e{index}", 0],
|
|
"negative": ["neg", 0],
|
|
"latent_image": [f"z{chain}", 0],
|
|
},
|
|
}
|
|
return source
|
|
|
|
|
|
def _depth_straddling_samplers() -> dict:
|
|
"""A at depth 9 below its sampler, B four Reroutes deeper on the same chain."""
|
|
source = {"p": {"class_type": "PrimitiveString", "inputs": {"value": "cat"}}}
|
|
previous = "p"
|
|
for index in range(6):
|
|
source[f"c{index}"] = {
|
|
"class_type": "Reroute",
|
|
"inputs": {"value": [previous, 0]},
|
|
}
|
|
previous = f"c{index}"
|
|
source["encA"] = {"class_type": "CLIPTextEncode", "inputs": {"text": [previous, 0]}}
|
|
source["A"] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["encA", 0], "negative": ["encA", 0]},
|
|
}
|
|
for index in range(4):
|
|
source[f"b{index}"] = {
|
|
"class_type": "Reroute",
|
|
"inputs": {"value": [previous, 0]},
|
|
}
|
|
previous = f"b{index}"
|
|
source["encB"] = {"class_type": "CLIPTextEncode", "inputs": {"text": [previous, 0]}}
|
|
source["B"] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["encB", 0], "negative": ["encB", 0]},
|
|
}
|
|
return source
|
|
|
|
|
|
def _nested_shared_chain_samplers() -> dict:
|
|
"""A, B and C share one chain; each later sampler hangs below a memo hit."""
|
|
source = {"p": {"class_type": "PrimitiveString", "inputs": {"value": "cat"}}}
|
|
|
|
def chain(start: str, prefix: str, count: int) -> str:
|
|
previous = start
|
|
for index in range(count):
|
|
source[f"{prefix}{index}"] = {
|
|
"class_type": "Reroute",
|
|
"inputs": {"value": [previous, 0]},
|
|
}
|
|
previous = f"{prefix}{index}"
|
|
return previous
|
|
|
|
def sampler(name: str, text_source: str) -> None:
|
|
source[f"enc{name}"] = {
|
|
"class_type": "CLIPTextEncode",
|
|
"inputs": {"text": [text_source, 0]},
|
|
}
|
|
source[name] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": [f"enc{name}", 0], "negative": [f"enc{name}", 0]},
|
|
}
|
|
|
|
top = chain("p", "c", 6) # A memoizes c5 with height 6
|
|
sampler("A", top)
|
|
middle = chain(top, "r", 1) # r0 is built over a memo hit on c5
|
|
sampler("B", middle)
|
|
sampler("C", chain(middle, "d", 3)) # C reaches r0 three Reroutes deeper
|
|
return source
|
|
|
|
|
|
def _random_graph(rng: random.Random) -> dict:
|
|
ids = [str(index) for index in range(rng.randrange(5, 60))]
|
|
classes = (
|
|
"StringConcatenate",
|
|
"Text Concatenate",
|
|
"Reroute",
|
|
"PrimitiveString",
|
|
"CLIPTextEncode",
|
|
"KSampler",
|
|
"PromptManager",
|
|
"EmptyLatentImage",
|
|
)
|
|
|
|
def link():
|
|
return [rng.choice(ids), 0]
|
|
|
|
source = {}
|
|
for node_id in ids:
|
|
name = rng.choice(classes)
|
|
if name == "StringConcatenate":
|
|
inputs = {"string_a": link(), "string_b": rng.choice([link(), "x"])}
|
|
elif name == "Text Concatenate":
|
|
inputs = {"text_a": link(), "text_b": link(), "delimiter": ""}
|
|
elif name == "Reroute":
|
|
inputs = {"value": link()}
|
|
elif name == "PrimitiveString":
|
|
inputs = {"value": rng.choice(["a", " b "])}
|
|
elif name == "CLIPTextEncode":
|
|
inputs = {"text": rng.choice([link(), "p"])}
|
|
elif name == "KSampler":
|
|
inputs = {"positive": link(), "negative": link()}
|
|
if rng.random() < 0.3:
|
|
inputs["latent_image"] = link()
|
|
elif name == "PromptManager":
|
|
inputs = {"text": rng.choice(["q", link()]), "prepend_text": ""}
|
|
else:
|
|
inputs = {}
|
|
source[node_id] = {"class_type": name, "inputs": inputs}
|
|
return source
|
|
|
|
|
|
def _outcomes(result: dict) -> dict:
|
|
items = result["candidates"] or [result]
|
|
return {
|
|
item["sampler_id"]: (item["state"], item["reason_code"], item["reason"])
|
|
for item in items
|
|
}
|
|
|
|
|
|
class _CountingInputs(dict):
|
|
"""Node inputs that count how often a resolver iterates (scans) their keys."""
|
|
|
|
scans = 0
|
|
|
|
def __iter__(self):
|
|
type(self).scans += 1
|
|
return super().__iter__()
|
|
|
|
|
|
class _CountingText(str):
|
|
"""A literal that tallies the characters a resolver strips from it."""
|
|
|
|
stripped = 0
|
|
|
|
def strip(self, *args):
|
|
type(self).stripped += len(self)
|
|
return super().strip(*args)
|
|
|
|
|
|
class TestResolverProfiles(unittest.TestCase):
|
|
"""Inspection bounds work, not graph size; the default profile is pinned."""
|
|
|
|
def test_inspection_profile_resolves_reachable_branch_in_506_node_graph(self):
|
|
expected = _vector("f03-default")
|
|
self.assertGreater(len(_f03_metadata()["prompt"]), 500)
|
|
result = resolve_prompt_pair(
|
|
_f03_metadata(), profile=resolver.INSPECTION_PROFILE
|
|
)
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertIsNone(result["reason_code"])
|
|
self.assertEqual(result["positive"]["text"], parity.POS)
|
|
self.assertEqual(result["negative"]["text"], parity.NEG)
|
|
self.assertEqual(expected["state"], "unresolved") # ingestion baseline
|
|
|
|
def test_default_profile_keeps_500_node_gate(self):
|
|
for result in (
|
|
resolve_prompt_pair(_f03_metadata()),
|
|
resolve_prompt_pair(_f03_metadata(), profile=resolver.DEFAULT_PROFILE),
|
|
):
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason"], "Resolution limit reached")
|
|
self.assertIsNone(result["reason_code"])
|
|
self.assertNotIn("node", result)
|
|
|
|
def test_total_visit_budget_is_shared_across_candidates(self):
|
|
source = _distinct_diamond_samplers(33)
|
|
self.assertEqual(resolver.INSPECTION_PROFILE.total_visits, 20_000)
|
|
self.assertLess(len(source), resolver.INSPECTION_PROFILE.node_ceiling)
|
|
result = resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE
|
|
)
|
|
# While selection is off (until PR3) many samplers are always ambiguous;
|
|
# PR3 must add "work-limit never upgrades to resolved" once one is chosen.
|
|
self.assertNotEqual(result["state"], "resolved")
|
|
self.assertEqual(result["state"], "ambiguous")
|
|
candidates = result["candidates"]
|
|
self.assertEqual(len(candidates), 33)
|
|
self.assertEqual(candidates[0]["state"], "resolved")
|
|
codes = [candidate["reason_code"] for candidate in candidates]
|
|
first_limited = codes.index("work-limit")
|
|
self.assertGreater(first_limited, 0)
|
|
self.assertTrue(all(code == "work-limit" for code in codes[first_limited:]))
|
|
self.assertTrue(
|
|
all(c["state"] == "unresolved" for c in candidates[first_limited:])
|
|
)
|
|
# The default profile keeps its historical gate on the same graph.
|
|
self.assertEqual(
|
|
resolve_prompt_pair({"prompt": source})["reason"],
|
|
"Resolution limit reached",
|
|
)
|
|
|
|
def test_memo_hit_respects_the_depth_left_at_the_reusing_candidate(self):
|
|
source = _depth_straddling_samplers()
|
|
reverse = dict(reversed(list(source.items())))
|
|
profile = resolver.INSPECTION_PROFILE
|
|
forward = _outcomes(
|
|
resolve_prompt_pair({"prompt": source}, profile=profile, max_depth=10)
|
|
)
|
|
backward = _outcomes(
|
|
resolve_prompt_pair({"prompt": reverse}, profile=profile, max_depth=10)
|
|
)
|
|
self.assertEqual(forward, backward)
|
|
self.assertEqual(forward["A"][:2], ("resolved", None))
|
|
self.assertEqual(forward["B"][:2], ("unresolved", "depth-limit"))
|
|
anchored = resolve_prompt_pair(
|
|
{"prompt": source}, profile=profile, max_depth=10, output_node_id="B"
|
|
)
|
|
self.assertEqual(
|
|
(anchored["state"], anchored["reason_code"]), ("unresolved", "depth-limit")
|
|
)
|
|
|
|
def test_memo_height_propagates_through_a_nested_memo_hit(self):
|
|
source = _nested_shared_chain_samplers()
|
|
reverse = dict(reversed(list(source.items())))
|
|
profile = resolver.INSPECTION_PROFILE
|
|
for graph in (source, reverse):
|
|
outcomes = _outcomes(
|
|
resolve_prompt_pair({"prompt": graph}, profile=profile, max_depth=12)
|
|
)
|
|
self.assertEqual(outcomes["C"][:2], ("unresolved", "depth-limit"))
|
|
anchored = resolve_prompt_pair(
|
|
{"prompt": source}, profile=profile, max_depth=12, output_node_id="C"
|
|
)
|
|
self.assertEqual(
|
|
(anchored["state"], anchored["reason_code"]), ("unresolved", "depth-limit")
|
|
)
|
|
|
|
def test_candidate_outcomes_match_anchored_and_ignore_node_order(self):
|
|
profile = resolver.INSPECTION_PROFILE
|
|
limits = ({}, {"max_depth": 4}, {"max_depth": 8}, {"max_visits": 6})
|
|
compared = 0
|
|
for seed in range(240, 246): # 6,000 deterministic graphs, about 1.5 s
|
|
rng = random.Random(seed)
|
|
for _ in range(1_000):
|
|
source = _random_graph(rng)
|
|
kwargs = rng.choice(limits)
|
|
base = _outcomes(
|
|
resolve_prompt_pair({"prompt": source}, profile=profile, **kwargs)
|
|
)
|
|
items = list(source.items())
|
|
rng.shuffle(items)
|
|
shuffled = _outcomes(
|
|
resolve_prompt_pair(
|
|
{"prompt": dict(items)}, profile=profile, **kwargs
|
|
)
|
|
)
|
|
self.assertEqual(base, shuffled, (seed, source, kwargs))
|
|
if len(base) < 2:
|
|
continue
|
|
for sampler_id, outcome in base.items():
|
|
anchored = resolve_prompt_pair(
|
|
{"prompt": source},
|
|
profile=profile,
|
|
output_node_id=sampler_id,
|
|
**kwargs,
|
|
)
|
|
self.assertEqual(
|
|
_outcomes(anchored)[sampler_id],
|
|
outcome,
|
|
(seed, sampler_id, source, kwargs),
|
|
)
|
|
compared += 1
|
|
self.assertGreater(compared, 1_000)
|
|
|
|
def test_provenance_entry_budget_reports_the_path_limit(self):
|
|
letters = "abcdefghijklmnopqrstuvwxyz"
|
|
source = {
|
|
"p": {"class_type": "PrimitiveString", "inputs": {"value": "cat"}},
|
|
"neg": {"class_type": "CLIPTextEncode", "inputs": {"text": "bad"}},
|
|
}
|
|
previous = "p"
|
|
for index in range(60):
|
|
source[f"c{index}"] = {
|
|
"class_type": "Reroute",
|
|
"inputs": {"value": [previous, 0]},
|
|
}
|
|
previous = f"c{index}"
|
|
for index in range(500):
|
|
source[f"w{index}"] = {
|
|
"class_type": "Text Concatenate",
|
|
"inputs": {f"text_{c}": [previous, 0] for c in letters},
|
|
}
|
|
source[f"e{index}"] = {
|
|
"class_type": "CLIPTextEncode",
|
|
"inputs": {"text": [f"w{index}", 0]},
|
|
}
|
|
source[f"s{index}"] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": [f"e{index}", 0], "negative": ["neg", 0]},
|
|
}
|
|
result = resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE
|
|
)
|
|
reasons = [candidate["reason"] for candidate in result["candidates"]]
|
|
self.assertEqual(result["candidates"][0]["state"], "resolved")
|
|
first = reasons.index("Provenance path limit reached")
|
|
self.assertGreater(first, 0)
|
|
self.assertLessEqual(
|
|
set(reasons[first:]),
|
|
{"Provenance path limit reached", "Resolution visit limit reached"},
|
|
)
|
|
self.assertEqual(reasons[first], "Provenance path limit reached")
|
|
self.assertEqual(
|
|
{c["reason_code"] for c in result["candidates"][first:]}, {"work-limit"}
|
|
)
|
|
# A small entry budget pins the same reason on one candidate.
|
|
small = replace(resolver.INSPECTION_PROFILE, max_path_entries=10)
|
|
result = resolve_prompt_pair({"prompt": source}, profile=small)
|
|
self.assertEqual(
|
|
(result["candidates"][0]["reason"], result["candidates"][0]["reason_code"]),
|
|
("Provenance path limit reached", "work-limit"),
|
|
)
|
|
|
|
def test_shared_node_is_scanned_once_however_many_candidates_use_it(self):
|
|
_CountingInputs.scans = 0
|
|
fat = _CountingInputs({"text": "portrait"})
|
|
fat.update({f"extra_{index}": 0 for index in range(20_000)})
|
|
source = {
|
|
"fat": {"class_type": "CLIPTextEncode", "inputs": fat},
|
|
"neg": {"class_type": "CLIPTextEncode", "inputs": {"text": "bad"}},
|
|
}
|
|
for index in range(1_000):
|
|
source[f"s{index}"] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {"positive": ["fat", 0], "negative": ["neg", 0]},
|
|
}
|
|
result = resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE
|
|
)
|
|
states = {candidate["state"] for candidate in result["candidates"]}
|
|
self.assertEqual(states, {"resolved"})
|
|
# Work is O(nodes), not O(candidates x node size).
|
|
self.assertLessEqual(_CountingInputs.scans, 2)
|
|
|
|
def test_stripped_concatenation_text_is_charged_to_a_work_budget(self):
|
|
_CountingText.stripped = 0
|
|
letters = "abcdefghijklmnopqrstuvwxyz"
|
|
hub = _CountingText(" " * 100_000)
|
|
source = graph()
|
|
source["hub"] = {"class_type": "PrimitiveString", "inputs": {"value": hub}}
|
|
for leaf in range(300):
|
|
source[f"l{leaf}"] = {
|
|
"class_type": "Text Concatenate",
|
|
"inputs": {f"text_{c}": ["hub", 0] for c in letters},
|
|
}
|
|
for middle in range(12):
|
|
source[f"m{middle}"] = {
|
|
"class_type": "Text Concatenate",
|
|
"inputs": {
|
|
f"text_{letters[i]}": [f"l{middle * 25 + i}", 0] for i in range(25)
|
|
},
|
|
}
|
|
source["root"] = {
|
|
"class_type": "Text Concatenate",
|
|
"inputs": {f"text_{letters[i]}": [f"m{i}", 0] for i in range(12)},
|
|
}
|
|
source["1"]["inputs"]["text"] = ["root", 0]
|
|
result = resolve_prompt_pair(source, profile=resolver.INSPECTION_PROFILE)
|
|
self.assertEqual(
|
|
(result["state"], result["reason_code"]), ("unresolved", "work-limit")
|
|
)
|
|
self.assertLessEqual(
|
|
_CountingText.stripped, resolver.INSPECTION_PROFILE.max_scanned_units
|
|
)
|
|
|
|
def test_per_candidate_visit_budget_resets_for_each_sampler(self):
|
|
source = _diamond_samplers(3)
|
|
result = resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE, max_visits=700
|
|
)
|
|
self.assertEqual(
|
|
[candidate["state"] for candidate in result["candidates"]],
|
|
["resolved"] * 3,
|
|
)
|
|
result = resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE, max_visits=300
|
|
)
|
|
self.assertEqual(
|
|
[candidate["reason_code"] for candidate in result["candidates"]],
|
|
["work-limit"] * 3,
|
|
)
|
|
|
|
def test_node_ceiling_reports_graph_ceiling(self):
|
|
source = graph()
|
|
for index in range(20_001 - len(source)):
|
|
source[f"p{index}"] = {"class_type": "PrimitiveInt", "inputs": {"value": 1}}
|
|
self.assertEqual(len(source), 20_001)
|
|
result = resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE
|
|
)
|
|
self.assertEqual(result["reason_code"], "graph-ceiling")
|
|
self.assertNotEqual(result["state"], "resolved")
|
|
del source["p0"]
|
|
result = resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE
|
|
)
|
|
self.assertEqual(result["state"], "resolved")
|
|
|
|
def test_prompt_graph_without_supported_sampler_is_unresolved_not_absent(self):
|
|
# f17 with its sampler renamed to a class the inspector does not know.
|
|
source = parity.build("f17")["prompt"]
|
|
source["13"]["class_type"] = "FutureSampler"
|
|
metadata = {"prompt": source}
|
|
result = resolve_prompt_pair(metadata, profile=resolver.INSPECTION_PROFILE)
|
|
self.assertEqual(
|
|
(result["state"], result["reason_code"]), ("unresolved", "no-sampler")
|
|
)
|
|
self.assertIn("FutureSampler", result["reason"])
|
|
self.assertIsNone(result["positive"])
|
|
default = resolve_prompt_pair(metadata)
|
|
self.assertEqual(default["state"], "absent")
|
|
self.assertEqual(default["reason"], "No supported sampler pair found")
|
|
self.assertIsNone(default["reason_code"])
|
|
|
|
def test_no_sampler_reason_names_at_most_three_sampler_like_classes(self):
|
|
source = {
|
|
str(index): {
|
|
"class_type": f"FutureSampler{index % 5}",
|
|
"inputs": {"latent_image": ["x", 0]},
|
|
}
|
|
for index in range(10)
|
|
}
|
|
result = resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE
|
|
)
|
|
self.assertEqual(result["reason_code"], "no-sampler")
|
|
self.assertEqual(
|
|
result["reason"],
|
|
"No supported sampler pair found "
|
|
"(saw FutureSampler0, FutureSampler1, FutureSampler2)",
|
|
)
|
|
|
|
def test_absent_only_without_prompt_or_parameters(self):
|
|
for value in (None, [], "raw text", {}, {"parameters": None}):
|
|
with self.subTest(value=value):
|
|
result = resolve_prompt_pair(value, profile=resolver.INSPECTION_PROFILE)
|
|
self.assertEqual(result["state"], "absent")
|
|
self.assertIsNone(result["positive"])
|
|
note = {"1": {"class_type": "Note", "inputs": {"text": "hello"}}}
|
|
for value in (
|
|
{"prompt": note},
|
|
note,
|
|
{"prompt": {}},
|
|
{"prompt": []},
|
|
{"prompt": {"1": None}},
|
|
{"parameters": "unstructured arbitrary text"},
|
|
):
|
|
with self.subTest(value=value):
|
|
result = resolve_prompt_pair(value, profile=resolver.INSPECTION_PROFILE)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertIsNotNone(result["reason_code"])
|
|
self.assertEqual(resolve_prompt_pair({"prompt": note})["state"], "absent")
|
|
# A pair-level "absent" names the missing negative; it always has a positive.
|
|
source = graph()
|
|
del source["3"]["inputs"]["negative"]
|
|
result = resolve_prompt_pair(source, profile=resolver.INSPECTION_PROFILE)
|
|
self.assertEqual(result["state"], "absent")
|
|
self.assertEqual(result["positive"]["text"], "portrait")
|
|
|
|
def test_existing_limits_map_to_reason_codes(self):
|
|
cases = (
|
|
({"max_visits": 1}, "work-limit", "Resolution visit limit reached"),
|
|
({"max_depth": 0}, "depth-limit", None),
|
|
({"max_depth": 1}, "depth-limit", "Resolution depth limit reached"),
|
|
({"max_text_length": 3}, "text-limit", None),
|
|
)
|
|
for limits, code, reason in cases:
|
|
with self.subTest(limits=limits):
|
|
default = resolve_prompt_pair(graph(), **limits)
|
|
self.assertEqual(default["state"], "unresolved")
|
|
self.assertIsNone(default["reason_code"])
|
|
result = resolve_prompt_pair(
|
|
graph(), profile=resolver.INSPECTION_PROFILE, **limits
|
|
)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason_code"], code)
|
|
if reason:
|
|
self.assertEqual(result["reason"], reason)
|
|
result = resolve_prompt_pair(
|
|
{"parameters": "x" * 20},
|
|
profile=resolver.INSPECTION_PROFILE,
|
|
max_text_length=10,
|
|
)
|
|
self.assertEqual(result["reason_code"], "text-limit")
|
|
|
|
def test_unsupported_node_names_the_blocking_node(self):
|
|
source = graph()
|
|
source["1"]["class_type"] = "UnknownEncoder"
|
|
result = resolve_prompt_pair(source, profile=resolver.INSPECTION_PROFILE)
|
|
self.assertEqual(result["reason_code"], "unsupported-node")
|
|
self.assertEqual(result["node"], {"id": "1", "class_type": "UnknownEncoder"})
|
|
default = resolve_prompt_pair(source)
|
|
self.assertIsNone(default["reason_code"])
|
|
self.assertNotIn("node", default)
|
|
self.assertEqual(default["reason"], result["reason"])
|
|
|
|
def test_default_profile_emits_only_a_null_reason_code_beyond_todays_keys(self):
|
|
source = graph()
|
|
source["4"] = {"class_type": "KSampler", "inputs": source["3"]["inputs"]}
|
|
for value in (graph(), source, {"parameters": "a\nSteps: 1, Sampler: x"}):
|
|
result = resolve_prompt_pair(value)
|
|
self.assertEqual(
|
|
set(result),
|
|
{
|
|
"state",
|
|
"positive",
|
|
"negative",
|
|
"sampler_id",
|
|
"reason",
|
|
"candidates",
|
|
"reason_code",
|
|
},
|
|
)
|
|
self.assertIsNone(result["reason_code"])
|
|
for candidate in result["candidates"]:
|
|
self.assertNotIn("node", candidate)
|
|
self.assertIsNone(candidate["reason_code"])
|
|
|
|
def test_diamond_provenance_paths_are_bounded_in_inspection(self):
|
|
source = graph()
|
|
source["8"] = {"class_type": "PrimitiveString", "inputs": {"value": ""}}
|
|
for index in range(9, 49):
|
|
source[str(index)] = {
|
|
"class_type": "StringConcatenate",
|
|
"inputs": {
|
|
"string_a": [str(index - 1), 0],
|
|
"string_b": [str(index - 1), 0],
|
|
"delimiter": "",
|
|
},
|
|
}
|
|
source["1"]["inputs"]["text"] = ["48", 0]
|
|
result = resolve_prompt_pair(source, profile=resolver.INSPECTION_PROFILE)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason_code"], "work-limit")
|
|
# A memo hit charges its unfolded cost: the 2^k unfolded visits exhaust
|
|
# the 2,000 allowance (per-entry paths never outgrow the allowance).
|
|
self.assertEqual(result["reason"], "Resolution visit limit reached")
|
|
|
|
|
|
def _guided(guider: dict = None) -> dict:
|
|
"""``f07b`` (SamplerCustomAdvanced + CFGGuider), optionally another guider."""
|
|
source = parity.build("f07b")["prompt"]
|
|
if guider is not None:
|
|
source["21"] = guider
|
|
return source
|
|
|
|
|
|
def _basic_guided() -> dict:
|
|
"""``f17`` with its BasicGuider reading the CLIPTextEncode directly."""
|
|
source = parity.build("f17")["prompt"]
|
|
source["22"]["inputs"]["conditioning"] = ["6", 0]
|
|
del source["26"]
|
|
return source
|
|
|
|
|
|
def _inspect(source: dict, **options) -> dict:
|
|
return resolve_prompt_pair(
|
|
{"prompt": source}, profile=resolver.INSPECTION_PROFILE, **options
|
|
)
|
|
|
|
|
|
class TestGuiderSamplers(unittest.TestCase):
|
|
"""SamplerCustomAdvanced follows its guider through the semantics table."""
|
|
|
|
def test_sampler_custom_advanced_with_cfg_guider_resolves(self):
|
|
result = _inspect(_guided())
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertIsNone(result["reason_code"])
|
|
self.assertEqual(result["sampler_id"], "3")
|
|
self.assertEqual(result["positive"]["text"], parity.POS)
|
|
self.assertEqual(result["negative"]["text"], parity.NEG)
|
|
provenance = result["positive"]["provenance"]
|
|
self.assertEqual(
|
|
(provenance["sampler_id"], provenance["guider_id"]), ("3", "21")
|
|
)
|
|
self.assertEqual(provenance["node_id"], "6")
|
|
self.assertEqual(provenance["path"], [{"node_id": "6", "output_slot": 0}])
|
|
self.assertEqual(result["negative"]["provenance"]["node_id"], "7")
|
|
# The saver anchor reaches the same sampler.
|
|
anchored = _inspect(_guided(), output_node_id="9")
|
|
self.assertEqual(anchored, result)
|
|
|
|
def test_basic_guider_negative_is_absent_by_design(self):
|
|
from utils.prompt_inspector import inspect_prompt_evidence
|
|
|
|
result = _inspect(_basic_guided())
|
|
self.assertEqual(result["state"], "absent")
|
|
self.assertEqual(result["reason"], "Guider has no negative input")
|
|
self.assertIsNone(result["reason_code"])
|
|
self.assertEqual(result["positive"]["text"], parity.POS)
|
|
self.assertIsNone(result["negative"])
|
|
evidence = inspect_prompt_evidence({"prompt": _basic_guided()})
|
|
self.assertEqual(evidence["prompts"]["positive"]["status"], "resolved")
|
|
self.assertEqual(evidence["prompts"]["positive"]["text"], parity.POS)
|
|
self.assertEqual(evidence["prompts"]["negative"]["status"], "absent")
|
|
self.assertEqual(
|
|
evidence["resolution"]["reason"], "Guider has no negative input"
|
|
)
|
|
|
|
def test_basic_guider_empty_positive_is_not_absent(self):
|
|
source = _basic_guided()
|
|
source["6"]["inputs"]["text"] = " "
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason"], "Positive prompt is empty")
|
|
|
|
def test_dual_cfg_guider_is_unsupported_node(self):
|
|
dual = {
|
|
"class_type": "DualCFGGuider",
|
|
"inputs": {
|
|
"model": ["4", 0],
|
|
"cond1": ["6", 0],
|
|
"cond2": ["6", 0],
|
|
"negative": ["7", 0],
|
|
"cfg_conds": 8.0,
|
|
"cfg_cond2_negative": 8.0,
|
|
},
|
|
}
|
|
result = _inspect(_guided(dual))
|
|
self.assertEqual(
|
|
(result["state"], result["reason_code"], result["node"]["class_type"]),
|
|
("unresolved", "unsupported-node", "DualCFGGuider"),
|
|
)
|
|
self.assertEqual(result["node"], {"id": "21", "class_type": "DualCFGGuider"})
|
|
self.assertEqual(result["sampler_id"], "3")
|
|
self.assertIsNone(result["positive"])
|
|
|
|
def test_unsupported_node_names_first_blocking_node(self):
|
|
source = _guided()
|
|
source["30"] = {
|
|
"class_type": "FutureConditioner",
|
|
"inputs": {"conditioning": ["6", 0]},
|
|
}
|
|
source["31"] = {
|
|
"class_type": "OtherConditioner",
|
|
"inputs": {"conditioning": ["7", 0]},
|
|
}
|
|
source["21"]["inputs"]["positive"] = ["30", 0]
|
|
source["21"]["inputs"]["negative"] = ["31", 0]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["reason_code"], "unsupported-node")
|
|
self.assertEqual(
|
|
result["node"], {"id": "30", "class_type": "FutureConditioner"}
|
|
)
|
|
# With the positive fixed, the negative's node is the first blocker.
|
|
source["21"]["inputs"]["positive"] = ["6", 0]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["node"], {"id": "31", "class_type": "OtherConditioner"})
|
|
self.assertEqual(result["positive"]["text"], parity.POS)
|
|
|
|
def test_default_profile_ignores_semantics(self):
|
|
for source in (_guided(), _basic_guided()):
|
|
result = resolve_prompt_pair({"prompt": source})
|
|
self.assertIn(result["state"], ("unresolved", "absent"))
|
|
self.assertEqual(result["reason"], "No supported sampler pair found")
|
|
self.assertIsNone(result["reason_code"])
|
|
self.assertIsNone(result["positive"])
|
|
# Without semantics the inspection bounds alone do not follow guiders.
|
|
plain = replace(resolver.INSPECTION_PROFILE, semantics=False)
|
|
result = resolve_prompt_pair({"prompt": _guided()}, profile=plain)
|
|
self.assertEqual(result["reason_code"], "no-sampler")
|
|
|
|
def test_guider_reached_through_a_reroute(self):
|
|
source = _guided()
|
|
source["40"] = {"class_type": "Reroute", "inputs": {"value": ["21", 0]}}
|
|
source["3"]["inputs"]["guider"] = ["40", 0]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(result["positive"]["provenance"]["guider_id"], "21")
|
|
|
|
def test_malformed_guider_links_and_slots_never_guess(self):
|
|
cases = {
|
|
"missing": None,
|
|
"literal": "CFGGuider",
|
|
"dangling": ["99", 0],
|
|
"slot": ["21", 1],
|
|
}
|
|
for label, value in cases.items():
|
|
with self.subTest(label=label):
|
|
source = _guided()
|
|
if value is None:
|
|
del source["3"]["inputs"]["guider"]
|
|
else:
|
|
source["3"]["inputs"]["guider"] = value
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertIsNone(result["positive"])
|
|
expected = "unsupported-node" if label == "slot" else "malformed"
|
|
self.assertEqual(result["reason_code"], expected)
|
|
source = _guided()
|
|
source["21"]["inputs"]["model"] = ["21", 0]
|
|
source["40"] = {"class_type": "Reroute", "inputs": {"value": ["40", 0]}}
|
|
source["3"]["inputs"]["guider"] = ["40", 0]
|
|
self.assertEqual(_inspect(source)["reason"], "Cyclic guider ancestry")
|
|
source["40"]["inputs"] = {"value": ["21", 0], "input": ["21", 0]}
|
|
result = _inspect(source)
|
|
self.assertEqual(result["reason"], "Reroute input is not uniquely identifiable")
|
|
self.assertEqual(result["reason_code"], "malformed")
|
|
|
|
def test_guider_hops_draw_from_the_work_budgets(self):
|
|
self.assertEqual(_inspect(_guided(), max_visits=2)["reason_code"], "work-limit")
|
|
self.assertEqual(_inspect(_guided())["state"], "resolved")
|
|
source = _guided()
|
|
previous = "21"
|
|
for index in range(70):
|
|
source[f"r{index}"] = {
|
|
"class_type": "Reroute",
|
|
"inputs": {"value": [previous, 0]},
|
|
}
|
|
previous = f"r{index}"
|
|
source["3"]["inputs"]["guider"] = [previous, 0]
|
|
self.assertEqual(_inspect(source)["reason_code"], "depth-limit")
|
|
|
|
def test_custom_advanced_upstream_of_a_sampler_is_multistage(self):
|
|
source = _guided()
|
|
source["41"] = {
|
|
"class_type": "KSampler",
|
|
"inputs": {
|
|
"positive": ["6", 0],
|
|
"negative": ["7", 0],
|
|
"latent_image": ["3", 0],
|
|
},
|
|
}
|
|
source["8"]["inputs"]["samples"] = ["41", 0]
|
|
result = _inspect(source, output_node_id="9")
|
|
self.assertEqual(result["sampler_id"], "41")
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason"], "Multistage sampler ancestry is ambiguous")
|
|
unanchored = _inspect(source)
|
|
self.assertEqual(unanchored["state"], "ambiguous")
|
|
self.assertEqual(
|
|
sorted(item["sampler_id"] for item in unanchored["candidates"]),
|
|
["3", "41"],
|
|
)
|
|
|
|
|
|
WOLF = "a grey wolf in fog"
|
|
|
|
|
|
def _encoded(text: str) -> dict:
|
|
return {"class_type": "CLIPTextEncode", "inputs": {"text": text, "clip": ["4", 1]}}
|
|
|
|
|
|
def _with_positive(node: dict, extra: dict = None) -> dict:
|
|
"""``f02``'s KSampler graph whose positive input reads ``node`` (as node 6)."""
|
|
source = parity.build("f02")["prompt"]
|
|
source["6"] = node
|
|
source.update(extra or {})
|
|
return source
|
|
|
|
|
|
def _if_else(condition, on_true, on_false) -> dict:
|
|
return {
|
|
"class_type": "easy ifElse",
|
|
"inputs": {"boolean": condition, "on_true": on_true, "on_false": on_false},
|
|
}
|
|
|
|
|
|
def _image_exists() -> dict:
|
|
return {"class_type": "ImageExists", "inputs": {"image": ["13", 0]}}
|
|
|
|
|
|
def _sides(result: dict) -> tuple:
|
|
return tuple(
|
|
result[role]["text"] if isinstance(result[role], dict) else None
|
|
for role in ("positive", "negative")
|
|
)
|
|
|
|
|
|
class TestConditioningSemantics(unittest.TestCase):
|
|
"""Pass-throughs, zeroed conditioning, selectors and composites (inspection)."""
|
|
|
|
def test_flux_guidance_passthrough_resolves(self):
|
|
result = _inspect(parity.build("f06c")["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}, {"node_id": "10", "output_slot": 0}],
|
|
)
|
|
self.assertEqual(provenance["text_source"], "graph-inferred")
|
|
# f17: BasicGuider <- FluxGuidance <- CLIPTextEncode, negative absent.
|
|
guided = _inspect(parity.build("f17")["prompt"])
|
|
self.assertEqual(guided["state"], "absent")
|
|
self.assertEqual(guided["reason"], "Guider has no negative input")
|
|
self.assertEqual(guided["positive"]["text"], parity.POS)
|
|
|
|
def test_controlnet_apply_advanced_maps_slots(self):
|
|
source = parity.build("f06e")["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": "7", "output_slot": 1}, {"node_id": "11", "output_slot": 0}],
|
|
)
|
|
# Slot 0 of the negative's node is its positive output, never the negative.
|
|
source["3"]["inputs"]["negative"] = ["7", 0]
|
|
self.assertEqual(_sides(_inspect(source)), (parity.POS, parity.POS))
|
|
# An output slot the table does not list is unsupported, named.
|
|
source["3"]["inputs"]["negative"] = ["7", 2]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["reason_code"], "unsupported-node")
|
|
self.assertEqual(
|
|
result["node"], {"id": "7", "class_type": "ControlNetApplyAdvanced"}
|
|
)
|
|
|
|
def test_reference_latent_and_set_area_pass_through(self):
|
|
chain = [
|
|
("ReferenceLatent", {"latent": ["5", 0]}),
|
|
("ConditioningSetArea", {"width": 64, "height": 64, "x": 0, "y": 0}),
|
|
("ConditioningSetAreaPercentage", {"width": 1.0, "height": 1.0}),
|
|
("ConditioningSetAreaStrength", {"strength": 1.0}),
|
|
("ConditioningSetMask", {"mask": ["40", 0], "strength": 1.0}),
|
|
("ConditioningSetTimestepRange", {"start": 0.0, "end": 1.0}),
|
|
("ControlNetApply", {"control_net": ["41", 0], "strength": 1.0}),
|
|
]
|
|
extra, previous = {"10": _encoded(parity.POS)}, "10"
|
|
for index, (name, inputs) in enumerate(chain):
|
|
node_id = f"2{index}"
|
|
extra[node_id] = {
|
|
"class_type": name,
|
|
"inputs": {**inputs, "conditioning": [previous, 0]},
|
|
}
|
|
previous = node_id
|
|
source = _with_positive(extra.pop(previous), extra)
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(_sides(result), (parity.POS, parity.NEG))
|
|
path = result["positive"]["provenance"]["path"]
|
|
self.assertEqual(
|
|
[entry["node_id"] for entry in path],
|
|
["6", "25", "24", "23", "22", "21", "20", "10"],
|
|
)
|
|
|
|
def test_inpaint_model_conditioning_slots(self):
|
|
inpaint = {
|
|
"class_type": "InpaintModelConditioning",
|
|
"inputs": {
|
|
"positive": ["10", 0],
|
|
"negative": ["11", 0],
|
|
"vae": ["4", 2],
|
|
"pixels": ["13", 0],
|
|
"mask": ["14", 0],
|
|
"noise_mask": True,
|
|
},
|
|
}
|
|
extra = {"10": _encoded(parity.POS), "11": _encoded(parity.NEG)}
|
|
source = _with_positive(inpaint, extra)
|
|
source["3"]["inputs"]["negative"] = ["6", 1]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(_sides(result), (parity.POS, parity.NEG))
|
|
self.assertEqual(result["positive"]["provenance"]["output_slot"], 0)
|
|
self.assertEqual(result["negative"]["provenance"]["output_slot"], 1)
|
|
|
|
def test_zeroed_negative_is_explicit_empty_with_distinct_source(self):
|
|
from utils.prompt_inspector import inspect_prompt_evidence
|
|
|
|
source = parity.build("f23")["prompt"]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "resolved")
|
|
negative = result["negative"]
|
|
self.assertEqual((negative["text"], negative["storage_text"]), ("", ""))
|
|
self.assertEqual(negative["provenance"]["text_source"], "zeroed-conditioning")
|
|
self.assertEqual(
|
|
negative["provenance"]["path"], [{"node_id": "7", "output_slot": 0}]
|
|
)
|
|
self.assertEqual(result["positive"]["text"], parity.POS)
|
|
self.assertEqual(
|
|
result["positive"]["provenance"]["text_source"], "graph-inferred"
|
|
)
|
|
evidence = inspect_prompt_evidence({"prompt": source})["prompts"]
|
|
self.assertEqual(
|
|
evidence["negative"],
|
|
{"status": "resolved", "text": "", "source": "zeroed-conditioning"},
|
|
)
|
|
# The source survives pass-throughs and reroutes in front of it.
|
|
source["30"] = {
|
|
"class_type": "FluxGuidance",
|
|
"inputs": {"guidance": 3.5, "conditioning": ["7", 0]},
|
|
}
|
|
source["31"] = {"class_type": "Reroute", "inputs": {"value": ["30", 0]}}
|
|
source["3"]["inputs"]["negative"] = ["31", 0]
|
|
negative = _inspect(source)["negative"]
|
|
self.assertEqual(negative["text"], "")
|
|
self.assertEqual(negative["provenance"]["text_source"], "zeroed-conditioning")
|
|
# A zeroed positive is an empty positive, never a resolved prompt.
|
|
source["3"]["inputs"]["positive"] = ["7", 0]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason"], "Positive prompt is empty")
|
|
|
|
def test_if_else_literal_boolean_selects_branch(self):
|
|
source = parity.build("f20")["prompt"]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(_sides(result), (parity.POS, parity.NEG))
|
|
self.assertEqual(
|
|
[entry["node_id"] for entry in result["positive"]["provenance"]["path"]],
|
|
["6", "10"],
|
|
)
|
|
source["6"]["inputs"]["boolean"] = False
|
|
self.assertEqual(_inspect(source)["positive"]["text"], WOLF)
|
|
# The branch that is not taken is never traced.
|
|
source["6"]["inputs"]["on_true"] = ["99", 0]
|
|
self.assertEqual(_inspect(source)["state"], "resolved")
|
|
for literal in (1, "true", None):
|
|
with self.subTest(literal=literal):
|
|
source["6"]["inputs"]["boolean"] = literal
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason_code"], "malformed")
|
|
self.assertIsNone(result["positive"])
|
|
|
|
def test_if_else_linked_boolean_converging_branches_resolve(self):
|
|
guidance = {
|
|
"class_type": "FluxGuidance",
|
|
"inputs": {"guidance": 3.5, "conditioning": ["10", 0]},
|
|
}
|
|
extra = {"10": _encoded(parity.POS), "11": guidance, "12": _image_exists()}
|
|
source = _with_positive(_if_else(["12", 1], ["11", 0], ["10", 0]), extra)
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(_sides(result), (parity.POS, parity.NEG))
|
|
self.assertEqual(result["sampler_id"], "3")
|
|
self.assertEqual(result["candidates"], [])
|
|
# Two encoders with the same text converge as well.
|
|
source["11"] = _encoded(parity.POS)
|
|
self.assertEqual(_inspect(source)["state"], "resolved")
|
|
|
|
def test_if_else_linked_boolean_diverging_branches_are_ambiguous_with_candidates(
|
|
self,
|
|
):
|
|
extra = {
|
|
"10": _encoded(parity.POS),
|
|
"11": _encoded(WOLF),
|
|
"12": _image_exists(),
|
|
}
|
|
source = _with_positive(_if_else(["12", 1], ["10", 0], ["11", 0]), extra)
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "ambiguous")
|
|
self.assertEqual(result["sampler_id"], "3")
|
|
self.assertIsNone(result["positive"])
|
|
self.assertIsNone(result["negative"])
|
|
candidates = result["candidates"]
|
|
self.assertEqual([item["state"] for item in candidates], ["resolved"] * 2)
|
|
self.assertEqual(
|
|
[_sides(item) for item in candidates],
|
|
[(parity.POS, parity.NEG), (WOLF, parity.NEG)],
|
|
)
|
|
self.assertEqual(
|
|
[
|
|
item["positive"]["provenance"]["path"][-1]["node_id"]
|
|
for item in candidates
|
|
],
|
|
["10", "11"],
|
|
)
|
|
# One selector read by both roles is one runtime choice, not four pairs.
|
|
source["3"]["inputs"]["negative"] = ["6", 0]
|
|
candidates = _inspect(source)["candidates"]
|
|
self.assertEqual(
|
|
[_sides(item) for item in candidates],
|
|
[(parity.POS, parity.POS), (WOLF, WOLF)],
|
|
)
|
|
# Independent selectors on each role multiply.
|
|
source["7"] = _if_else(["16", 1], ["10", 0], ["14", 0])
|
|
source["14"] = _encoded(parity.NEG)
|
|
source["16"] = _image_exists()
|
|
source["3"]["inputs"]["negative"] = ["7", 0]
|
|
candidates = _inspect(source)["candidates"]
|
|
self.assertEqual(len(candidates), 4)
|
|
self.assertEqual(len({_sides(item) for item in candidates}), 4)
|
|
|
|
def test_selectors_sharing_one_condition_choose_together(self):
|
|
extra = {
|
|
"10": _encoded("pos A"),
|
|
"11": _encoded("pos B"),
|
|
"14": _encoded("neg A"),
|
|
"15": _encoded("neg B"),
|
|
"12": _image_exists(),
|
|
"16": _image_exists(),
|
|
}
|
|
source = _with_positive(_if_else(["12", 1], ["10", 0], ["11", 0]), extra)
|
|
source["7"] = _if_else(["12", 1], ["14", 0], ["15", 0])
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "ambiguous")
|
|
# One runtime value drives both selectors: never "pos A" with "neg B".
|
|
self.assertEqual(
|
|
[_sides(item) for item in result["candidates"]],
|
|
[("pos A", "neg A"), ("pos B", "neg B")],
|
|
)
|
|
# Another output slot of the same node is another value.
|
|
source["7"]["inputs"]["boolean"] = ["12", 0]
|
|
self.assertEqual(len(_inspect(source)["candidates"]), 4)
|
|
# Two independent conditions give the full cross product.
|
|
source["7"]["inputs"]["boolean"] = ["16", 1]
|
|
candidates = _inspect(source)["candidates"]
|
|
self.assertEqual(
|
|
sorted(_sides(item) for item in candidates),
|
|
[
|
|
("pos A", "neg A"),
|
|
("pos A", "neg B"),
|
|
("pos B", "neg A"),
|
|
("pos B", "neg B"),
|
|
],
|
|
)
|
|
|
|
def test_diverging_branch_that_cannot_be_traced_is_never_guessed(self):
|
|
extra = {
|
|
"10": _encoded(parity.POS),
|
|
"11": {"class_type": "FutureEncoder", "inputs": {"text": WOLF}},
|
|
"12": _image_exists(),
|
|
}
|
|
source = _with_positive(_if_else(["12", 1], ["10", 0], ["11", 0]), extra)
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["node"], {"id": "11", "class_type": "FutureEncoder"})
|
|
self.assertIsNone(result["positive"])
|
|
|
|
def test_selector_branches_are_bounded(self):
|
|
extra, previous = {"12": _image_exists()}, None
|
|
for index in range(6):
|
|
extra[f"a{index}"] = _encoded(f"left {index}")
|
|
extra[f"b{index}"] = _encoded(f"right {index}")
|
|
# Each selector reads its own condition: independent choices.
|
|
extra[f"e{index}"] = _image_exists()
|
|
extra[f"s{index}"] = _if_else(
|
|
[f"e{index}", 1], [f"a{index}", 0], [f"b{index}", 0]
|
|
)
|
|
extra[f"c{index}"] = {
|
|
"class_type": "ConditioningCombine",
|
|
"inputs": {
|
|
"conditioning_1": [f"s{index}", 0],
|
|
"conditioning_2": [previous or f"s{index}", 0],
|
|
},
|
|
}
|
|
previous = f"c{index}"
|
|
source = _with_positive(
|
|
{"class_type": "Reroute", "inputs": {"value": [previous, 0]}}, extra
|
|
)
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason_code"], "work-limit")
|
|
self.assertEqual(result["candidates"], [])
|
|
# Branch tracing draws from the per-candidate visit allowance too.
|
|
small = _with_positive(
|
|
_if_else(["12", 1], ["10", 0], ["11", 0]),
|
|
{"10": _encoded(parity.POS), "11": _encoded(WOLF), "12": _image_exists()},
|
|
)
|
|
self.assertEqual(_inspect(small, max_visits=3)["reason_code"], "work-limit")
|
|
|
|
def test_conditioning_combine_distinct_texts_is_composite(self):
|
|
source = parity.build("f06d")["prompt"]
|
|
result = _inspect(source)
|
|
self.assertEqual(
|
|
(result["state"], result["reason_code"]), ("unresolved", "composite")
|
|
)
|
|
self.assertEqual(
|
|
result["node"], {"id": "6", "class_type": "ConditioningCombine"}
|
|
)
|
|
self.assertIsNone(result["positive"])
|
|
for name, inputs in (
|
|
("ConditioningConcat", ("conditioning_to", "conditioning_from")),
|
|
("ConditioningAverage", ("conditioning_to", "conditioning_from")),
|
|
):
|
|
with self.subTest(name=name):
|
|
source["6"] = {
|
|
"class_type": name,
|
|
"inputs": {inputs[0]: ["10", 0], inputs[1]: ["11", 0]},
|
|
}
|
|
result = _inspect(source)
|
|
self.assertEqual(result["reason_code"], "composite")
|
|
self.assertEqual(result["node"], {"id": "6", "class_type": name})
|
|
# A zeroed part is a different text too.
|
|
source["12"] = {
|
|
"class_type": "ConditioningZeroOut",
|
|
"inputs": {"conditioning": ["10", 0]},
|
|
}
|
|
source["6"]["inputs"]["conditioning_from"] = ["12", 0]
|
|
self.assertEqual(_inspect(source)["reason_code"], "composite")
|
|
|
|
def test_conditioning_combine_identical_texts_resolve(self):
|
|
source = parity.build("f06d")["prompt"]
|
|
source["11"]["inputs"]["text"] = parity.POS
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(_sides(result), (parity.POS, parity.NEG))
|
|
source["6"]["inputs"]["conditioning_2"] = ["10", 0]
|
|
self.assertEqual(_inspect(source)["state"], "resolved")
|
|
# A missing part is malformed, never "the other part".
|
|
del source["6"]["inputs"]["conditioning_2"]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason_code"], "malformed")
|
|
|
|
def test_selector_sources_and_string_branches(self):
|
|
# Converging branches with different text sources name no single source.
|
|
source = parity.build("f23")["prompt"]
|
|
source["12"] = _image_exists()
|
|
source["14"] = _encoded("")
|
|
source["15"] = _if_else(["12", 1], ["7", 0], ["14", 0])
|
|
source["3"]["inputs"]["negative"] = ["15", 0]
|
|
negative = _inspect(source)["negative"]
|
|
self.assertEqual(negative["text"], "")
|
|
self.assertEqual(negative["provenance"]["text_source"], "graph-inferred")
|
|
source["3"]["inputs"]["negative"] = ["7", 0]
|
|
source["15"]["inputs"]["on_false"] = ["7", 0]
|
|
source["3"]["inputs"]["negative"] = ["15", 0]
|
|
negative = _inspect(source)["negative"]
|
|
self.assertEqual(negative["provenance"]["text_source"], "zeroed-conditioning")
|
|
# A selector also forwards string inputs, including literal text.
|
|
source = parity.build("f02")["prompt"]
|
|
source["12"] = _image_exists()
|
|
source["20"] = _if_else(["12", 1], "same words", "same words")
|
|
source["6"]["inputs"]["text"] = ["20", 0]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["positive"]["text"], "same words")
|
|
source["20"]["inputs"]["on_false"] = "other words"
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "ambiguous")
|
|
self.assertEqual(
|
|
[item["positive"]["text"] for item in result["candidates"]],
|
|
["same words", "other words"],
|
|
)
|
|
|
|
def test_default_profile_keeps_conditioning_nodes_unsupported(self):
|
|
for name in ("f06c", "f06d", "f06e", "f20", "f23"):
|
|
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 _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"]
|
|
if same_prompt:
|
|
source["50"]["inputs"]["positive"] = ["6", 0]
|
|
return source
|
|
|
|
|
|
def _detailer(image: list, positive: list, negative: list = None) -> dict:
|
|
return {
|
|
"class_type": "FaceDetailer",
|
|
"inputs": {
|
|
"image": image,
|
|
"model": ["4", 0],
|
|
"positive": positive,
|
|
"negative": negative or ["7", 0],
|
|
"seed": 1,
|
|
"steps": 8,
|
|
},
|
|
}
|
|
|
|
|
|
class TestDetailerStages(unittest.TestCase):
|
|
"""D19: a detailer is a sampler stage judged with its upstream sampler."""
|
|
|
|
def test_detailer_with_the_upstream_prompt_resolves(self):
|
|
result = _inspect(_detailed(same_prompt=True))
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(result["sampler_id"], "50")
|
|
self.assertEqual(_sides(result), (parity.POS, parity.NEG))
|
|
self.assertEqual(result["candidates"], [])
|
|
self.assertEqual(
|
|
_inspect(_detailed(same_prompt=True), output_node_id="9"), result
|
|
)
|
|
|
|
def test_detailer_with_a_different_prompt_is_ambiguous_with_both_stages(self):
|
|
from utils.prompt_inspector import inspect_prompt_evidence
|
|
|
|
result = _inspect(_detailed())
|
|
self.assertEqual(
|
|
(result["state"], result["reason_code"]),
|
|
("ambiguous", "multistage-differs"),
|
|
)
|
|
self.assertEqual(result["sampler_id"], "50")
|
|
self.assertIsNone(result["positive"])
|
|
self.assertIsNone(result["negative"])
|
|
self.assertEqual(
|
|
[(item["sampler_id"], _sides(item)) for item in result["candidates"]],
|
|
[
|
|
("50", (parity.DETAILER_POS, parity.NEG)),
|
|
("3", (parity.POS, parity.NEG)),
|
|
],
|
|
)
|
|
self.assertEqual(_inspect(_detailed(), output_node_id="9"), result)
|
|
# Neither the upstream prompt nor the detailer prompt is shown as the answer.
|
|
prompts = inspect_prompt_evidence({"prompt": _detailed()})["prompts"]
|
|
self.assertEqual(
|
|
[prompts[role]["status"] for role in ("positive", "negative")],
|
|
["ambiguous", "ambiguous"],
|
|
)
|
|
self.assertNotIn(parity.POS, str(prompts))
|
|
|
|
def test_detailer_of_a_loaded_image_is_a_single_stage(self):
|
|
source = _detailed()
|
|
source["13"] = {"class_type": "LoadImage", "inputs": {"image": "input.png"}}
|
|
source["50"]["inputs"]["image"] = ["13", 0]
|
|
source["9"]["inputs"]["images"] = ["50", 0]
|
|
source["8"]["inputs"]["samples"] = ["3", 0]
|
|
source["14"] = {"class_type": "SaveImage", "inputs": {"images": ["8", 0]}}
|
|
result = _inspect(source, output_node_id="9")
|
|
self.assertEqual(result["state"], "resolved")
|
|
self.assertEqual(_sides(result), (parity.DETAILER_POS, parity.NEG))
|
|
self.assertEqual(result["sampler_id"], "50")
|
|
|
|
def test_raw_and_detailed_saves_stay_ambiguous_without_an_anchor(self):
|
|
source = _detailed(same_prompt=True)
|
|
source["14"] = {"class_type": "SaveImage", "inputs": {"images": ["8", 0]}}
|
|
result = _inspect(source)
|
|
self.assertEqual(result["state"], "ambiguous")
|
|
self.assertEqual(result["reason"], "Multiple sampler branches")
|
|
self.assertEqual(
|
|
sorted(item["sampler_id"] for item in result["candidates"]), ["3", "50"]
|
|
)
|
|
raw = _inspect(source, output_node_id="14")
|
|
self.assertEqual((raw["state"], raw["sampler_id"]), ("resolved", "3"))
|
|
detailed = _inspect(source, output_node_id="9")
|
|
self.assertEqual(
|
|
(detailed["state"], detailed["sampler_id"]), ("resolved", "50")
|
|
)
|
|
|
|
def test_a_preview_of_the_upstream_image_is_not_another_output(self):
|
|
source = _detailed()
|
|
source["14"] = {"class_type": "PreviewImage", "inputs": {"images": ["8", 0]}}
|
|
result = _inspect(source)
|
|
self.assertEqual(result["reason_code"], "multistage-differs")
|
|
self.assertEqual(len(result["candidates"]), 2)
|
|
|
|
def test_chained_detailers_compare_every_stage(self):
|
|
source = _detailed(same_prompt=True)
|
|
source["51"] = _detailer(["50", 0], ["6", 0])
|
|
source["9"]["inputs"]["images"] = ["51", 0]
|
|
result = _inspect(source)
|
|
self.assertEqual((result["state"], result["sampler_id"]), ("resolved", "51"))
|
|
source["50"]["inputs"]["positive"] = ["10", 0]
|
|
result = _inspect(source)
|
|
self.assertEqual(result["reason_code"], "multistage-differs")
|
|
self.assertEqual(
|
|
[item["sampler_id"] for item in result["candidates"]], ["51", "50", "3"]
|
|
)
|
|
|
|
def test_unresolved_stage_blocks_the_chain(self):
|
|
source = _detailed(same_prompt=True)
|
|
source["30"] = {"class_type": "FutureConditioner", "inputs": {"c": ["6", 0]}}
|
|
source["3"]["inputs"]["positive"] = ["30", 0]
|
|
source["50"]["inputs"]["positive"] = ["10", 0]
|
|
result = _inspect(source)
|
|
self.assertEqual((result["state"], result["sampler_id"]), ("unresolved", "50"))
|
|
self.assertEqual(
|
|
result["node"], {"id": "30", "class_type": "FutureConditioner"}
|
|
)
|
|
self.assertIsNone(result["positive"])
|
|
|
|
def test_malformed_and_cyclic_detailer_ancestry_never_guesses(self):
|
|
source = _detailed()
|
|
del source["50"]["inputs"]["image"]
|
|
result = _inspect(source, output_node_id="9")
|
|
self.assertEqual(
|
|
(result["state"], result["reason_code"]), ("unresolved", "malformed")
|
|
)
|
|
source = _detailed()
|
|
source["8"]["inputs"]["samples"] = ["51", 0]
|
|
source["51"] = _detailer(["50", 0], ["6", 0])
|
|
result = _inspect(source, output_node_id="9")
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason_code"], "malformed")
|
|
|
|
def test_detailer_image_ancestry_is_walked_generically(self):
|
|
source = _detailed(same_prompt=True)
|
|
source["60"] = {
|
|
"class_type": "ImageBlend",
|
|
"inputs": {"image1": ["8", 0], "image2": ["8", 0], "blend_factor": 0.5},
|
|
}
|
|
source["61"] = {
|
|
"class_type": "ImageBlend",
|
|
"inputs": {"image1": ["8", 0], "image2": ["60", 0]},
|
|
}
|
|
source["62"] = {"class_type": "ImageScale", "inputs": {"image": ["60", 0]}}
|
|
source["50"]["inputs"]["image"] = ["61", 0]
|
|
source["63"] = {"class_type": "PreviewImage", "inputs": {"images": ["62", 0]}}
|
|
source["junk"] = "not a node"
|
|
result = _inspect(source)
|
|
self.assertEqual((result["state"], result["sampler_id"]), ("resolved", "50"))
|
|
# A detailer whose image ancestry returns to itself is malformed.
|
|
source = _detailed()
|
|
source["64"] = {"class_type": "ImageScale", "inputs": {"image": ["50", 0]}}
|
|
source["50"]["inputs"]["image"] = ["64", 0]
|
|
result = _inspect(source, output_node_id="9")
|
|
self.assertEqual(result["reason"], "Cyclic sampler stage ancestry")
|
|
self.assertEqual(result["reason_code"], "malformed")
|
|
|
|
def test_detailer_stage_walks_draw_from_the_work_budget(self):
|
|
self.assertEqual(
|
|
_inspect(_detailed(), max_visits=6)["reason_code"], "work-limit"
|
|
)
|
|
# Too little budget to absorb the upstream sampler keeps both as
|
|
# candidates: more ambiguity, never a chosen prompt.
|
|
starved = _inspect(_detailed(), max_visits=3)
|
|
self.assertEqual(starved["state"], "ambiguous")
|
|
self.assertEqual(len(starved["candidates"]), 2)
|
|
self.assertIsNone(starved["positive"])
|
|
source = _detailed()
|
|
previous = "8"
|
|
for index in range(70):
|
|
source[f"u{index}"] = {
|
|
"class_type": "ImageScale",
|
|
"inputs": {"image": [previous, 0]},
|
|
}
|
|
previous = f"u{index}"
|
|
source["50"]["inputs"]["image"] = [previous, 0]
|
|
result = _inspect(source, output_node_id="9")
|
|
self.assertEqual(result["reason_code"], "depth-limit")
|
|
|
|
def test_long_detailer_chains_are_depth_limited(self):
|
|
for count in (70, 600):
|
|
source = _detailed(same_prompt=True)
|
|
previous = "50"
|
|
for index in range(count):
|
|
node_id = f"d{index}"
|
|
source[node_id] = _detailer([previous, 0], ["6", 0])
|
|
previous = node_id
|
|
source["9"]["inputs"]["images"] = [previous, 0]
|
|
for options in ({}, {"output_node_id": "9"}):
|
|
with self.subTest(count=count, options=options):
|
|
result = _inspect(source, **options)
|
|
self.assertEqual(result["state"], "unresolved")
|
|
self.assertEqual(result["reason_code"], "depth-limit")
|
|
self.assertIsNone(result["positive"])
|
|
# A chain within the bound still resolves.
|
|
source = _detailed(same_prompt=True)
|
|
previous = "50"
|
|
for index in range(20):
|
|
source[f"d{index}"] = _detailer([previous, 0], ["6", 0])
|
|
previous = f"d{index}"
|
|
source["9"]["inputs"]["images"] = [previous, 0]
|
|
result = _inspect(source, output_node_id="9")
|
|
self.assertEqual((result["state"], result["sampler_id"]), ("resolved", "d19"))
|
|
|
|
def test_default_profile_keeps_the_upstream_sampler(self):
|
|
# Ingestion is unchanged (D1): the detailer is not a sampler there.
|
|
result = resolve_prompt_pair({"prompt": _detailed()})
|
|
self.assertEqual((result["state"], result["sampler_id"]), ("resolved", "3"))
|
|
self.assertEqual(_sides(result), (parity.POS, parity.NEG))
|
|
|
|
|
|
REPO = Path(__file__).resolve().parents[1]
|
|
SEMANTICS_UNAVAILABLE = "Inspection node semantics are unavailable"
|
|
|
|
|
|
class TestSemanticsIsolation(unittest.TestCase):
|
|
"""Only inspection reads the semantics table; a bad table never breaks ingestion."""
|
|
|
|
def resolve_with_table(self, text: str, *values):
|
|
# Windows: the table is written and closed before it is read or removed.
|
|
with tempfile.TemporaryDirectory() as root:
|
|
path = Path(root) / "semantics.json"
|
|
path.write_text(text, encoding="utf-8")
|
|
inspection_semantics.get_semantics.cache_clear()
|
|
try:
|
|
with (
|
|
mock.patch.object(inspection_semantics, "SEMANTICS_PATH", path),
|
|
mock.patch.object(resolver, "_LOGGER") as logger,
|
|
):
|
|
self.logged = logger.error.call_args_list
|
|
return [
|
|
resolve_prompt_pair(value, profile=profile)
|
|
for value in values
|
|
for profile in (
|
|
resolver.DEFAULT_PROFILE,
|
|
resolver.INSPECTION_PROFILE,
|
|
)
|
|
]
|
|
finally:
|
|
inspection_semantics.get_semantics.cache_clear()
|
|
|
|
def test_corrupt_table_leaves_default_resolution_working(self):
|
|
for text in ("{not json", "[" * 100_000, "{}"):
|
|
with self.subTest(text=text[:10]):
|
|
default, inspected = self.resolve_with_table(text, graph())
|
|
self.assertEqual(default["state"], "resolved")
|
|
self.assertEqual(default["positive"]["text"], "portrait")
|
|
self.assertEqual(inspected["state"], "unresolved")
|
|
self.assertEqual(inspected["reason"], SEMANTICS_UNAVAILABLE)
|
|
self.assertIsNone(inspected["reason_code"])
|
|
self.assertIsNone(inspected["positive"])
|
|
self.assertEqual(len(self.logged), 1)
|
|
|
|
def test_corrupt_table_only_affects_graph_inspection(self):
|
|
parameters = {"parameters": "cat\nNegative prompt: dog\nSteps: 1, Sampler: x"}
|
|
results = self.resolve_with_table("{not json", parameters, {})
|
|
self.assertEqual([item["state"] for item in results[:2]], ["resolved"] * 2)
|
|
self.assertEqual([item["state"] for item in results[2:]], ["absent"] * 2)
|
|
self.assertEqual(self.logged, [])
|
|
|
|
def test_corrupt_table_is_a_clear_inspector_outcome(self):
|
|
from utils.prompt_inspector import inspect_prompt_evidence
|
|
|
|
with tempfile.TemporaryDirectory() as root:
|
|
path = Path(root) / "semantics.json"
|
|
path.write_text("{not json", encoding="utf-8")
|
|
inspection_semantics.get_semantics.cache_clear()
|
|
try:
|
|
with (
|
|
mock.patch.object(inspection_semantics, "SEMANTICS_PATH", path),
|
|
self.assertLogs("prompt_manager.prompt_pair_resolver", "ERROR"),
|
|
):
|
|
evidence = inspect_prompt_evidence({"prompt": graph()})
|
|
finally:
|
|
inspection_semantics.get_semantics.cache_clear()
|
|
self.assertEqual(evidence["resolution"]["state"], "unresolved")
|
|
self.assertEqual(evidence["resolution"]["reason"], SEMANTICS_UNAVAILABLE)
|
|
self.assertEqual(
|
|
[evidence["prompts"][role]["status"] for role in ("positive", "negative")],
|
|
["unresolved", "unresolved"],
|
|
)
|
|
|
|
def test_default_resolution_never_imports_the_semantics_table(self):
|
|
code = (
|
|
"import sys\n"
|
|
"from utils.prompt_pair_resolver import resolve_prompt_pair\n"
|
|
"graph = {'1': {'class_type': 'CLIPTextEncode', 'inputs': {'text': 'a'}},"
|
|
" '3': {'class_type': 'KSampler',"
|
|
" 'inputs': {'positive': ['1', 0], 'negative': ['1', 0]}}}\n"
|
|
"assert resolve_prompt_pair(graph)['state'] == 'resolved'\n"
|
|
"assert 'utils.inspection_semantics' not in sys.modules\n"
|
|
)
|
|
# Fixed interpreter and literal code, no shell.
|
|
completed = subprocess.run( # nosec B603
|
|
[sys.executable, "-c", code],
|
|
cwd=REPO,
|
|
capture_output=True,
|
|
timeout=60,
|
|
check=False,
|
|
)
|
|
self.assertEqual(completed.returncode, 0, completed.stderr.decode("utf-8"))
|
|
|
|
|
|
if __name__ == "__main__":
|
|
unittest.main()
|