nodes/ComfyAssets-ComfyUI_PromptManager
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

fix(inspect): keep pair execution evidence behind unreadable string sources

Browse Source
This commit is contained in:
Vito Sansevero
2026-10-19 15:00:33 -07:00
parent fbd9fba7e3
commit 80b68f659e
2 changed files with 109 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
tests/test_prompt_pair_resolver.py
+89 -1
View File
@@ -1865,7 +1865,7 @@ class TestEncoderAndStringSemantics(unittest.TestCase):
# 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.
# A linked text is followed; a linked flag is unsupported (not guessed).
linked = _string_positive(
_multiline(["11", 0], True), {"11": _primitive("x\ny")}
)
@@ -1923,6 +1923,15 @@ class TestEncoderAndStringSemantics(unittest.TestCase):
{"node_id": "14", "output_slot": 0},
],
)
# One link read by several parts is listed once in the provenance path.
twice = _join(3, " ", string_1=["11", 0], string_2=["11", 0], string_3="c")
result = _inspect(_string_positive(twice, {"11": _primitive("w")}))
self.assertEqual(result["positive"]["text"], "w w c")
self.assertEqual(
result["positive"]["provenance"]["path"],
[{"node_id": "6", "output_slot": 0}, {"node_id": "10", "output_slot": 0}]
+ [{"node_id": "11", "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"
@@ -2081,6 +2090,85 @@ class TestEncoderAndStringSemantics(unittest.TestCase):
self.assertEqual(result["reason_code"], "unsupported-node")
self.assertEqual(result["node"]["class_type"], name)
def test_unreadable_string_sources_keep_pair_execution_evidence(self):
from utils.pair_execution import EXECUTION_KEY
record = {
"text": "recorded words",
"negative_text": "neg",
"positive_text": "recorded words",
"negative_effective_text": "neg",
"prepend_text": "",
"append_text": "",
}
evidence = {
EXECUTION_KEY: {"version": 1, "nodes": {"8": {"records": [record]}}}
}
def paired(source: dict, extra: dict) -> dict:
graph = parity.build("f02")["prompt"]
graph["8"] = {
"class_type": "PromptManagerPair",
"inputs": {
"text": ["10", 0],
"negative_text": "neg",
"prepend_text": "",
"append_text": "",
},
}
graph["3"]["inputs"]["positive"] = ["8", 0]
graph["3"]["inputs"]["negative"] = ["8", 1]
graph.update({"10": source, **extra})
return graph
flag = {"11": {"class_type": "PrimitiveBoolean", "inputs": {"value": True}}}
unknown = {"11": {"class_type": "FutureNode", "inputs": {}}}
# Text that cannot be read statically is unverifiable, not a mismatch:
# the pair's unique execution record still decides (as it did before
# these nodes were listed).
cases = {
"linked strip flag": (_multiline("recorded words", ["11", 0]), flag),
"linked delimiter": (
_join(2, ["11", 0], string_1="recorded", string_2="words"),
unknown,
),
"linked count": (
_join(["11", 0], " ", string_1="recorded", string_2="words"),
unknown,
),
"list output": (
_join(2, " ", True, string_1="recorded", string_2="words"),
{},
),
"search list": ({"class_type": "PromptSearchList", "inputs": {}}, {}),
}
for label, (source, extra) in cases.items():
with self.subTest(source=label):
result = resolve_prompt_pair(
{"prompt": paired(source, extra), **evidence},
profile=resolver.INSPECTION_PROFILE,
)
self.assertEqual(result["state"], "resolved")
self.assertEqual(_sides(result), ("recorded words", "neg"))
provenance = result["positive"]["provenance"]
self.assertEqual(provenance["text_source"], "execution")
# A source that does resolve is still verified against the record.
readable = _join(2, " ", string_1="other", string_2="words")
result = resolve_prompt_pair(
{"prompt": paired(readable, {}), **evidence},
profile=resolver.INSPECTION_PROFILE,
)
self.assertEqual(result["reason_code"], "malformed")
self.assertEqual(
result["reason"], "Execution evidence does not match recorded node inputs"
)
matching = _join(2, " ", string_1="recorded", string_2="words")
result = resolve_prompt_pair(
{"prompt": paired(matching, {}), **evidence},
profile=resolver.INSPECTION_PROFILE,
)
self.assertEqual(result["positive"]["provenance"]["text_source"], "execution")
def test_default_profile_keeps_encoder_and_string_nodes_unsupported(self):
for name in ("f06a", "f06b", "f21", "f22", "f24"):
with self.subTest(fixture=name):
utils/prompt_pair_resolver.py
+20 -2
View File
@@ -151,6 +151,8 @@ _MAX_BRANCH_CANDIDATES = 32
# Statically undecidable text (a search, a list output): never pick an item.
_RUNTIME_REASON = "Text is decided at run time"
_INCOMPATIBLE_SLOT = "Output slot has an incompatible type"
# Trace outcomes that leave an input unverifiable against execution evidence.
_UNVERIFIABLE_CODES = frozenset({"unsupported-node", "runtime-value"})
# JoinStringMulti's own ``inputcount`` range; anything else never executed.
_MAX_JOIN_INPUTS = 1000
_SAMPLER_LIKE_INPUTS = ("latent_image", "guider", "positive", "negative")
@@ -753,13 +755,18 @@ class _GraphResolver:
]
parts: list = []
path: list = []
seen: set = set()
size = 0
for position, input_name in enumerate(names):
if position and input_name not in inputs:
continue
part, followed, _ = self.text(inputs.get(input_name), "string", depth + 1)
self.scan(len(part) + len(followed))
path.extend(followed)
for step in followed: # one link read by several parts is listed once
identity = (step["node_id"], step["output_slot"])
if identity not in seen:
seen.add(identity)
path.append(step)
if position and not part:
continue
size += len(part) + (len(delimiter) if parts else 0)
@@ -787,6 +794,17 @@ class _GraphResolver:
else:
self.text_sources.pop((node_id, slot), None)
def unverifiable(self, error: _Unresolved) -> bool:
"""Whether a failed input trace only means "cannot be read statically".
Such an input is not checked against a pair's execution record. The
default profile keeps its historical message test; inspection decides
by code, so every unsupported or run-time source qualifies.
"""
if self.semantics is None:
return str(error) == "Unsupported node semantics or output slot"
return error.code in _UNVERIFIABLE_CODES
def execution_record(self, node_id: str, inputs: dict) -> Optional[dict]:
try:
record = read_pair_execution(self.metadata, node_id)
@@ -798,7 +816,7 @@ class _GraphResolver:
try:
expected = self.text(inputs.get(name, ""), "string")[0]
except _Unresolved as error:
if str(error) == "Unsupported node semantics or output slot":
if self.unverifiable(error):
value = inputs.get(name)
if _link(value):
source_name, _ = self.node(str(value[0]))