feat(inspect): explain why a prompt could not be resolved

This commit is contained in:
Vito Sansevero
2026-10-19 15:00:33 -07:00
parent 72a2d0c800
commit 2bda944581
12 changed files with 660 additions and 46 deletions
+3 -13
View File
@@ -9,6 +9,7 @@ try:
MetadataReadError,
inspect_prompt_evidence,
read_inspection_file,
unreadable_evidence,
)
from ...utils.video_metadata import is_video_path
except ImportError:
@@ -16,6 +17,7 @@ except ImportError:
MetadataReadError,
inspect_prompt_evidence,
read_inspection_file,
unreadable_evidence,
)
from utils.video_metadata import is_video_path
@@ -73,19 +75,7 @@ def inspect_current_file(
if any(getattr(before, key) != getattr(after, key) for key in fields):
raise CurrentFileChanged
if inspection is None:
evidence = {
"prompts": {
role: {"status": "unresolved", "text": None, "source": None}
for role in ("positive", "negative")
},
"resolution": {
"state": "unresolved",
"sampler_id": None,
"reason": "Media metadata could not be read",
"reason_code": None,
"node": None,
},
}
evidence = unreadable_evidence()
else:
role_saved = (
None if saved and saved.get("association_status") == "ambiguous" else saved
+11
View File
@@ -232,6 +232,17 @@ test('unverified saved associations explain uncertainty and cannot be sent',asyn
await expect(details(page).getByRole('button',{name:/^Send /})).toHaveCount(0);
});
test('unresolved roles name the blocking node as text and offer no prompt actions',async({page})=>{
const line='Prompt uses a node the inspector does not support (<img src=x onerror="window.injected=1">).';
await page.route('**/images/inspect/uncertain.png?root=0*',route=>route.fulfill({json:{success:true,root_scope:new URL(route.request().url()).searchParams.get('root_scope'),image:{root_index:0,path:'uncertain.png',filename:'uncertain.png'},prompts:{
positive:{status:'unresolved',text:null,source:null,message:line},negative:{status:'unresolved',text:null,source:null,message:line},
}}}));
await page.goto(workspace);await select(page,'0:uncertain.png');
for (const name of ['Positive','Negative']) await expect(role(details(page),name)).toContainText(line);
await expect(details(page).locator('img[onerror]')).toHaveCount(0);
expect(await page.evaluate(()=>window.injected)).toBeUndefined();
await expect(details(page).getByRole('button',{name:/^(Send|Copy) /})).toHaveCount(0);
});
test('prompt role headings contain compact real actions and honest source evidence',async({page})=>{
await page.goto(workspace);await select(page,'0:shared/same.png');
+49
View File
@@ -81,3 +81,52 @@ test('a zeroed conditioning negative shows as an explicit empty negative with it
assert.deepEqual(sent,[{positive:'positive',negative:''}]);
}finally{scope.dispose();f.restore();}
});
const unresolved=message=>({status:'unresolved',text:null,source:null,...(message===undefined?{}:{message})});
const statusLines=(f,container)=>f.walk(container,node=>node.tagName==='p'&&node.className==='muted').map(node=>node.textContent);
test('unsupported node status names the class',()=>{
const f=dom(),scope=createScope(),container=new f.Element('div');
const line='Prompt uses a node the inspector does not support (DualCFGGuider).';
try{
module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(line),negative:unresolved(line)}});
assert.deepEqual(statusLines(f,container),[line,line]);
assert.equal(f.walk(container,node=>node.dataset.sendRole!==undefined||node.dataset.copyRole!==undefined).length,0,'no action implies a prompt exists');
}finally{scope.dispose();f.restore();}
});
test('runtime and composite statuses',()=>{
const f=dom(),scope=createScope(),container=new f.Element('div');
const runtime='Prompt is chosen at run time (PromptSearchList).',composite='Prompt combines several texts (ConditioningCombine).';
try{
const view=module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(runtime),negative:unresolved(runtime)}});
assert.deepEqual(statusLines(f,container),[runtime,runtime]);
view.update({positive:unresolved(composite),negative:{status:'resolved',text:'kept',source:'execution'}});
assert.deepEqual(statusLines(f,container),[composite]);
assert.equal(f.walk(container,node=>node.tagName==='pre'&&node.textContent==='kept').length,1);
}finally{scope.dispose();f.restore();}
});
test('limit status',()=>{
const f=dom(),scope=createScope(),container=new f.Element('div');
const line='Prompt exceeds inspection limits.';
try{
module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(line),negative:unresolved(line)}});
assert.deepEqual(statusLines(f,container),[line,line]);
}finally{scope.dispose();f.restore();}
});
test('roles without a usable line keep the status copy, and a candidate is never shown as a prompt',()=>{
const f=dom(),scope=createScope(),container=new f.Element('div');
try{
module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(''),negative:{status:'ambiguous',text:'candidate text',source:null}}});
assert.deepEqual(statusLines(f,container),['Prompt could not be resolved.','Multiple possible prompts. Source could not be determined.']);
assert.equal(f.walk(container,node=>node.textContent==='candidate text').length,0);
}finally{scope.dispose();f.restore();}
});
test('class names in status lines render as text, never markup',()=>{
const f=dom(),scope=createScope(),container=new f.Element('div');
const line='Prompt uses a node the inspector does not support (<img src=x onerror=alert(1)>).';
try{
module.mountPromptBlocks(container,{parentScope:scope,sender:{open:()=>{}},prompts:{positive:unresolved(line),negative:unresolved(line)}});
assert.deepEqual(statusLines(f,container),[line,line]);
assert.equal(f.walk(container,node=>node.tagName==='img').length,0);
assert.equal(f.walk(container,node=>Object.hasOwn(node,'innerHTML')).length,0);
}finally{scope.dispose();f.restore();}
});
@@ -225,6 +225,48 @@ test('record accepts closed reason codes and rejects contradictory ones', () =>
}
});
test('resolution validator accepts closed reason codes and a bounded node record, rejects others', () => {
const toRecord = contract('toInspectionRecord');
for (const reason_code of ['unsupported-node', 'runtime-value', 'composite']) {
for (const node of [{ id: '7', class_type: 'N'.repeat(200) }, { id: '7'.repeat(200), class_type: null }, null]) {
const data = envelope(); Object.assign(data.resolution, { state: 'unresolved', reason_code, node });
assert.doesNotThrow(() => toRecord(data, image(), extractGenerationFacts), JSON.stringify({ reason_code, node }));
}
}
for (const change of [
{ reason_code: 'unsupported-node', node: { id: '7', class_type: 'N'.repeat(201) } },
{ reason_code: 'unsupported-node', node: { id: '7'.repeat(201), class_type: 'X' } },
{ reason_code: 'unsupported-node', node: { id: '', class_type: 'X' } },
{ reason_code: 'unsupported-node', node: { id: '7', class_type: 'X', extra: true } },
{ reason_code: 'unsupported-node', node: [] },
{ reason_code: 'malformed', node: { id: '7', class_type: 'X' } },
{ reason_code: 'work-limit', node: { id: '7', class_type: 'X' } },
{ reason_code: 'no-sampler', node: { id: '7', class_type: 'X' } },
{ reason_code: 'Unsupported-Node', node: null },
{ reason_code: '', node: null },
]) {
const data = envelope(); Object.assign(data.resolution, { state: 'unresolved', ...change });
assert.throws(() => toRecord(data, image(), extractGenerationFacts), TypeError, JSON.stringify(change));
}
});
test('role outcome lines are bounded text on non-resolved roles only', () => {
const toRecord = contract('toInspectionRecord');
const line = 'Prompt uses a node the inspector does not support (<img src=x onerror=alert(1)>).';
const data = envelope(); data.prompts.negative = { status: 'unresolved', text: null, source: null, message: line };
assert.equal(toRecord(data, image(), extractGenerationFacts).prompts.negative.message, line);
for (const [role, record] of [
['positive', { status: 'resolved', text: 'exact', source: 'execution', message: 'Hidden reason.' }],
['negative', { status: 'unresolved', text: null, source: null, message: '' }],
['negative', { status: 'unresolved', text: null, source: null, message: 7 }],
['negative', { status: 'unresolved', text: null, source: null, message: null }],
['negative', { status: 'unresolved', text: null, source: null, message: 'x'.repeat(513) }],
]) {
const bad = envelope(); bad.prompts[role] = record;
assert.throws(() => toRecord(bad, image(), extractGenerationFacts), TypeError, JSON.stringify(record));
}
});
test('record rejects root path media and invalid versioned fields with retryable TypeError', () => {
const toRecord = contract('toInspectionRecord');
for (const mutate of [
+16 -4
View File
@@ -244,10 +244,10 @@ class TestPromptInspector(ImageAPITestCase):
async def test_corrupt_malformed_and_no_metadata_do_not_invent_text(self):
path = self.output_dir / "image.png"
for content, expected in (
(b"broken", "unresolved"),
(None, "absent"),
("bad-json", "unresolved"),
for content, expected, message in (
(b"broken", "unresolved", "Metadata in this file could not be read."),
(None, "absent", None),
("bad-json", "unresolved", "Prompt metadata in this file is malformed."),
):
if isinstance(content, bytes):
path.write_bytes(content)
@@ -256,6 +256,7 @@ class TestPromptInspector(ImageAPITestCase):
_, result = await self.inspect()
self.assertEqual(result["prompts"]["positive"]["status"], expected)
self.assertIsNone(result["prompts"]["positive"]["text"])
self.assertEqual(result["prompts"]["negative"].get("message"), message)
async def test_compressed_international_and_post_pixel_metadata(self):
import struct
@@ -790,6 +791,9 @@ class TestPromptInspector(ImageAPITestCase):
result["resolution"]["reason"],
"No supported sampler pair found (saw FutureSampler)",
)
self.assertEqual(
result["prompts"]["positive"]["message"], "No supported sampler was found."
)
async def test_inspect_route_exposes_reason_code_and_blocking_node(self):
# f06a's encoder is supported since Task 2.4; an unknown encoder is not.
@@ -803,6 +807,14 @@ class TestPromptInspector(ImageAPITestCase):
self.assertEqual(
resolution["node"], {"id": "6", "class_type": "FutureTextEncode"}
)
self.assertEqual(
{role: side.get("message") for role, side in result["prompts"].items()},
dict.fromkeys(
("positive", "negative"),
"Prompt uses a node the inspector does not support "
"(FutureTextEncode).",
),
)
png(self.output_dir / "image.png", prompt="{broken")
_, result = await self.inspect()
self.assertEqual(result["resolution"]["reason_code"], "malformed")
+355
View File
@@ -0,0 +1,355 @@
"""Non-resolved prompt outcomes carry one short, truthful line per role."""
import json
import unittest
from dataclasses import replace
from unittest import mock
from utils.inspection_outcomes import (
CLASS_NAME_LENGTH,
REASON_CODES,
outcome_message,
project_resolution,
)
from utils.inspection_response import encode_inspection_response
from utils.prompt_inspector import inspect_prompt_evidence
from utils.prompt_pair_resolver import (
EMPTY_POSITIVE_REASON,
INSPECTION_PROFILE,
MULTISTAGE_UNATTRIBUTED_REASON,
SEMANTICS_UNAVAILABLE_REASON,
)
LIMITS = "Prompt exceeds inspection limits."
def _encode(text: str) -> dict:
return {"class_type": "CLIPTextEncode", "inputs": {"text": text, "clip": ["9", 0]}}
def _latent() -> dict:
return {
"class_type": "EmptyLatentImage",
"inputs": {"width": 8, "height": 8, "batch_size": 1},
}
def _ksampler(positive: str = "7") -> dict:
return {
"class_type": "KSampler",
"inputs": {
"positive": [positive, 0],
"negative": ["6", 0],
"latent_image": ["4", 0],
"model": ["9", 0],
"seed": 1,
"steps": 2,
"cfg": 1.0,
"sampler_name": "euler",
"scheduler": "normal",
"denoise": 1.0,
},
}
def _dual_cfg() -> dict:
return {
"1": _encode("positive"),
"6": _encode("negative"),
"2": {
"class_type": "DualCFGGuider",
"inputs": {
"cond1": ["1", 0],
"cond2": ["1", 0],
"negative": ["6", 0],
"model": ["9", 0],
},
},
"3": {
"class_type": "SamplerCustomAdvanced",
"inputs": {
"guider": ["2", 0],
"noise": ["9", 0],
"sampler": ["9", 0],
"sigmas": ["9", 0],
"latent_image": ["4", 0],
},
},
"4": _latent(),
}
def _runtime() -> dict:
return {
"1": {"class_type": "PromptSearchList", "inputs": {"search": "cat"}},
"7": {
"class_type": "CLIPTextEncode",
"inputs": {"text": ["1", 0], "clip": ["9", 0]},
},
"6": _encode("negative"),
"3": _ksampler(),
"4": _latent(),
}
def _composite() -> dict:
graph = _runtime()
graph["7"] = {
"class_type": "ConditioningCombine",
"inputs": {"conditioning_1": ["6", 0], "conditioning_2": ["8", 0]},
}
graph["8"] = _encode("other")
return graph
def _messages(evidence: dict) -> dict:
return {role: side.get("message") for role, side in evidence["prompts"].items()}
def _statuses(evidence: dict) -> dict:
return {role: side["status"] for role, side in evidence["prompts"].items()}
class TestOutcomeMessages(unittest.TestCase):
def assert_unresolved_with(self, evidence: dict, message: str) -> None:
self.assertEqual(
_statuses(evidence), {"positive": "unresolved", "negative": "unresolved"}
)
self.assertEqual(
_messages(evidence), {"positive": message, "negative": message}
)
def test_unsupported_node_status_names_the_class(self):
evidence = inspect_prompt_evidence({"prompt": _dual_cfg()})
self.assertEqual(evidence["resolution"]["reason_code"], "unsupported-node")
self.assert_unresolved_with(
evidence,
"Prompt uses a node the inspector does not support (DualCFGGuider).",
)
def test_runtime_and_composite_statuses(self):
self.assert_unresolved_with(
inspect_prompt_evidence({"prompt": _runtime()}),
"Prompt is chosen at run time (PromptSearchList).",
)
self.assert_unresolved_with(
inspect_prompt_evidence({"prompt": _composite()}),
"Prompt combines several texts (ConditioningCombine).",
)
def test_limit_status(self):
tight = replace(INSPECTION_PROFILE, per_candidate_visits=1, total_visits=1)
evidence = inspect_prompt_evidence({"prompt": _runtime()}, profile=tight)
self.assertEqual(evidence["resolution"]["reason_code"], "work-limit")
self.assert_unresolved_with(evidence, LIMITS)
for code in ("work-limit", "depth-limit", "text-limit", "graph-ceiling"):
self.assertEqual(
outcome_message({"state": "unresolved", "reason_code": code}), LIMITS
)
def test_every_closed_code_and_known_uncoded_reason_has_a_line(self):
for code in REASON_CODES:
message = outcome_message({"state": "unresolved", "reason_code": code})
self.assertIsInstance(message, str, code)
self.assertTrue(message.endswith("."), code)
self.assertNotIn("could not be resolved", message)
for reason, expected in (
(EMPTY_POSITIVE_REASON, "Positive prompt is empty."),
(
SEMANTICS_UNAVAILABLE_REASON,
"The inspector's node table is unavailable.",
),
(
MULTISTAGE_UNATTRIBUTED_REASON,
"Several sampler stages were found; their prompts were not compared.",
),
(
"Media metadata could not be read",
"Metadata in this file could not be read.",
),
):
resolution = {"state": "unresolved", "reason": reason, "reason_code": None}
self.assertEqual(outcome_message(resolution), expected)
def test_named_codes_without_a_class_omit_the_name(self):
for code, line in (
("unsupported-node", "Prompt uses a node the inspector does not support."),
("runtime-value", "Prompt is chosen at run time."),
("composite", "Prompt combines several texts."),
):
for node in (None, {"id": "7", "class_type": None}):
resolution = {"state": "unresolved", "reason_code": code, "node": node}
self.assertEqual(outcome_message(resolution), line)
def test_class_name_is_bounded(self):
resolution = {
"state": "unresolved",
"reason_code": "unsupported-node",
"node": {"id": "7", "class_type": "N" * 500},
}
self.assertEqual(
outcome_message(resolution),
"Prompt uses a node the inspector does not support "
f"({'N' * CLASS_NAME_LENGTH}).",
)
self.assertEqual(CLASS_NAME_LENGTH, 200)
def test_resolved_ambiguous_absent_and_unknown_outcomes_carry_no_line(self):
for resolution in (
{"state": "resolved", "reason_code": None},
{"state": "ambiguous", "reason_code": "multistage-differs"},
{"state": "ambiguous", "reason_code": None, "reason": "Multiple"},
{"state": "absent", "reason_code": None, "reason": "No parsed"},
{"state": "unresolved", "reason_code": "guess"},
{"state": "unresolved", "reason_code": None, "reason": "Anything else"},
None,
):
self.assertIsNone(outcome_message(resolution), resolution)
def test_inspector_lines_for_no_sampler_malformed_and_empty_positive(self):
evidence = inspect_prompt_evidence({"prompt": {"1": _encode("x")}})
self.assert_unresolved_with(evidence, "No supported sampler was found.")
evidence = inspect_prompt_evidence({"prompt": "not-json"})
self.assert_unresolved_with(
evidence, "Prompt metadata in this file is malformed."
)
graph = _runtime()
graph["7"] = _encode(" ")
evidence = inspect_prompt_evidence({"prompt": graph})
self.assertEqual(evidence["resolution"]["reason"], EMPTY_POSITIVE_REASON)
self.assert_unresolved_with(evidence, "Positive prompt is empty.")
def test_a_resolved_pair_with_an_empty_positive_explains_only_that_role(self):
side = {"text": " ", "provenance": {"text_source": "graph-inferred"}}
negative = {"text": "kept", "provenance": {"text_source": "graph-inferred"}}
resolved = {
"state": "resolved",
"positive": side,
"negative": negative,
"sampler_id": "3",
"reason": "",
"reason_code": None,
"node": None,
}
with mock.patch(
"utils.prompt_inspector.resolve_prompt_pair", return_value=resolved
):
evidence = inspect_prompt_evidence({"prompt": {}})
self.assertEqual(
evidence["prompts"]["positive"],
{
"status": "unresolved",
"text": None,
"source": None,
"message": "Positive prompt is empty.",
},
)
self.assertEqual(evidence["prompts"]["negative"]["text"], "kept")
self.assertNotIn("message", evidence["prompts"]["negative"])
def test_candidates_and_absence_keep_their_status_lines(self):
graph = _runtime()
graph["7"] = _encode("positive")
graph["5"] = _ksampler()
evidence = inspect_prompt_evidence({"prompt": graph})
self.assertEqual(evidence["resolution"]["state"], "ambiguous")
self.assertEqual(_messages(evidence), {"positive": None, "negative": None})
evidence = inspect_prompt_evidence({})
self.assertEqual(_messages(evidence), {"positive": None, "negative": None})
graph = _runtime()
graph["7"] = _encode("positive")
evidence = inspect_prompt_evidence({"prompt": graph})
self.assertEqual(evidence["resolution"]["state"], "resolved")
self.assertEqual(_messages(evidence), {"positive": None, "negative": None})
class TestResolutionProjection(unittest.TestCase):
def test_closed_codes_and_named_nodes_pass_unchanged(self):
for code in REASON_CODES:
resolution = {
"state": "unresolved",
"sampler_id": "3",
"reason": "r",
"reason_code": code,
"node": None,
}
self.assertEqual(project_resolution(resolution), resolution)
named = {
"state": "unresolved",
"sampler_id": None,
"reason": "r",
"reason_code": "composite",
"node": {"id": "7", "class_type": None},
}
self.assertEqual(project_resolution(named), named)
def test_unknown_codes_and_unbounded_nodes_fail_closed(self):
base = {"state": "unresolved", "sampler_id": None, "reason": "r"}
for code, node in (
("guess", None),
(1, None),
("malformed", {"id": "7", "class_type": "X"}),
(None, {"id": "7", "class_type": "X"}),
("unsupported-node", {"id": 7, "class_type": "X"}),
("unsupported-node", {"id": "", "class_type": "X"}),
("unsupported-node", {"id": "7"}),
("unsupported-node", {"id": "7", "class_type": 7}),
("unsupported-node", {"id": "7", "class_type": "X", "extra": 1}),
("unsupported-node", "X"),
):
projected = project_resolution({**base, "reason_code": code, "node": node})
self.assertIsNone(projected["node"], (code, node))
if code not in REASON_CODES:
self.assertIsNone(projected["reason_code"], code)
for state in ("resolved", "absent"):
projected = project_resolution(
{**base, "state": state, "reason_code": "work-limit", "node": None}
)
self.assertIsNone(projected["reason_code"])
long_name = project_resolution(
{
**base,
"reason_code": "runtime-value",
"node": {"id": "7" * 300, "class_type": "N" * 300},
}
)["node"]
self.assertEqual(
long_name, {"id": "7" * CLASS_NAME_LENGTH, "class_type": "N" * 200}
)
unnamed = project_resolution(
{
**base,
"reason_code": "composite",
"node": {"id": "7", "class_type": ""},
}
)["node"]
self.assertEqual(unnamed, {"id": "7", "class_type": None})
def test_wire_carries_role_lines_and_projected_resolution(self):
evidence = inspect_prompt_evidence({"prompt": _dual_cfg()})
resolution = dict(evidence["resolution"], reason_code="guess")
body = json.loads(
encode_inspection_response(
image={"root_index": 0, "path": "a.png", "filename": "a.png"},
root_scope="a" * 64,
file={
"observation": "current-file",
"source_label": "output",
"media_type": "image",
"extension": ".png",
"size": 1,
"modified_time": 1,
},
prompts=evidence["prompts"],
resolution=resolution,
inspection=None,
library={"status": "none", "tags": []},
)
)
self.assertEqual(body["prompts"], evidence["prompts"])
self.assertIsNone(body["resolution"]["reason_code"])
self.assertIsNone(body["resolution"]["node"])
if __name__ == "__main__":
unittest.main()
+109
View File
@@ -0,0 +1,109 @@
"""User-facing lines for non-resolved prompt outcomes, from the closed vocabulary.
The inspector shows one line per unresolved role in place of the generic
"Prompt could not be resolved." Ambiguous and absent roles keep their status
copy, so a candidate is never presented as a confirmed prompt. The wire
projection accepts exactly the closed ``reason_code`` vocabulary and names a
blocking node only beside a code that names one.
"""
from typing import Any, Optional
from .prompt_pair_resolver import (
EMPTY_POSITIVE_REASON,
MULTISTAGE_UNATTRIBUTED_REASON,
SEMANTICS_UNAVAILABLE_REASON,
)
REASON_CODES = frozenset(
{
"work-limit",
"depth-limit",
"text-limit",
"graph-ceiling",
"unsupported-node",
"runtime-value",
"composite",
"no-sampler",
"malformed",
"multistage-differs",
}
)
LIMIT_CODES = frozenset({"work-limit", "depth-limit", "text-limit", "graph-ceiling"})
# Codes whose outcome names the blocking node: ``node: {id, class_type}``.
NAMED_CODES = frozenset({"unsupported-node", "runtime-value", "composite"})
# Bound for a node ID or class name on the wire and in a line.
CLASS_NAME_LENGTH = 200
UNREADABLE_METADATA_REASON = "Media metadata could not be read"
_NODE_KEYS = frozenset({"id", "class_type"})
_NAMED_LINES = {
"unsupported-node": "Prompt uses a node the inspector does not support",
"runtime-value": "Prompt is chosen at run time",
"composite": "Prompt combines several texts",
}
_CODED_LINES = {
**dict.fromkeys(LIMIT_CODES, "Prompt exceeds inspection limits."),
"no-sampler": "No supported sampler was found.",
"malformed": "Prompt metadata in this file is malformed.",
"multistage-differs": "Sampler stages have different prompts.",
}
# Uncoded outcomes, keyed by the resolver's exact reason.
_UNCODED_LINES = {
EMPTY_POSITIVE_REASON: "Positive prompt is empty.",
MULTISTAGE_UNATTRIBUTED_REASON: (
"Several sampler stages were found; their prompts were not compared."
),
SEMANTICS_UNAVAILABLE_REASON: "The inspector's node table is unavailable.",
UNREADABLE_METADATA_REASON: "Metadata in this file could not be read.",
}
def _text(value: Any) -> Optional[str]:
return value if isinstance(value, str) else None
def _class_name(node: Any) -> Optional[str]:
if not isinstance(node, dict):
return None
name = node.get("class_type")
return name[:CLASS_NAME_LENGTH] if isinstance(name, str) and name else None
def outcome_message(resolution: Any) -> Optional[str]:
"""One line for an unresolved outcome, or ``None`` to keep the status copy."""
if not isinstance(resolution, dict) or resolution.get("state") != "unresolved":
return None
code = _text(resolution.get("reason_code"))
if code in _NAMED_LINES:
name = _class_name(resolution.get("node"))
return f"{_NAMED_LINES[code]} ({name})." if name else f"{_NAMED_LINES[code]}."
if code is None:
return _UNCODED_LINES.get(_text(resolution.get("reason")))
return _CODED_LINES.get(code)
def _node(code: Optional[str], node: Any) -> Optional[dict]:
if code not in NAMED_CODES or not isinstance(node, dict) or set(node) != _NODE_KEYS:
return None
node_id, name = node["id"], node["class_type"]
if not isinstance(node_id, str) or not node_id:
return None
if name is not None and not isinstance(name, str):
return None
return {
"id": node_id[:CLASS_NAME_LENGTH],
"class_type": name[:CLASS_NAME_LENGTH] if name else None,
}
def project_resolution(resolution: dict) -> dict:
"""Fail closed: an unknown or contradictory code and its node are dropped."""
code = _text(resolution.get("reason_code"))
if code not in REASON_CODES or resolution.get("state") in ("resolved", "absent"):
code = None
return {
**resolution,
"reason_code": code,
"node": _node(code, resolution.get("node")),
}
+3 -1
View File
@@ -5,6 +5,8 @@ import math
from decimal import Decimal, InvalidOperation
from typing import TYPE_CHECKING, Literal
from .inspection_outcomes import project_resolution
if TYPE_CHECKING:
from .prompt_inspector import InspectionRead
@@ -329,7 +331,7 @@ def _encode_inspection(
**identity,
"file": _file_projection(file, inspection),
"prompts": prompts,
"resolution": resolution,
"resolution": project_resolution(resolution),
"metadata": metadata,
"library": _library_projection(library),
}
+54 -19
View File
@@ -13,8 +13,10 @@ from typing import Any
from PIL import Image
from .inspection_outcomes import UNREADABLE_METADATA_REASON, outcome_message
from .pair_execution import EXECUTION_KEY
from .prompt_pair_resolver import (
EMPTY_POSITIVE_REASON,
INSPECTION_PROFILE,
ResolverProfile,
resolve_prompt_pair,
@@ -400,6 +402,39 @@ def _side(status: str, text: str | None = None, source: str | None = None) -> di
return {"status": status, "text": text, "source": source}
_EMPTY_POSITIVE = outcome_message(
{"state": "unresolved", "reason": EMPTY_POSITIVE_REASON}
)
def _explained(prompts: dict, resolution: dict) -> dict:
"""Each unresolved role carries the outcome's line; other roles are kept."""
message = outcome_message(resolution)
if message is None:
return prompts
return {
role: (
{**side, "message": message}
if side["status"] == "unresolved" and "message" not in side
else side
)
for role, side in prompts.items()
}
def unreadable_evidence() -> dict:
"""Both roles when the file's metadata could not be read at all."""
resolution = {
"state": "unresolved",
"sampler_id": None,
"reason": UNREADABLE_METADATA_REASON,
"reason_code": None,
"node": None,
}
prompts = {role: _side("unresolved") for role in ("positive", "negative")}
return {"prompts": _explained(prompts, resolution), "resolution": resolution}
def inspect_prompt_evidence(
metadata: dict,
saved: dict | None = None,
@@ -447,21 +482,20 @@ def inspect_prompt_evidence(
},
}
canonical = dict(metadata)
malformed = {
"prompts": _explained(
{role: _side("unresolved") for role in ("positive", "negative")},
resolution,
),
"resolution": resolution,
}
if "prompt" not in canonical and "parameters" not in canonical:
return {
"prompts": {role: _side("unresolved") for role in ("positive", "negative")},
"resolution": resolution,
}
return malformed
if isinstance(canonical.get("prompt"), str):
try:
canonical["prompt"] = json.loads(canonical["prompt"])
except (ValueError, RecursionError):
return {
"prompts": {
role: _side("unresolved") for role in ("positive", "negative")
},
"resolution": resolution,
}
return malformed
resolved = resolve_prompt_pair(canonical, profile=profile)
state = resolved["state"]
result = {}
@@ -475,18 +509,19 @@ def inspect_prompt_evidence(
result[role] = _side(
"resolved", side["text"], side["provenance"]["text_source"]
)
elif state == "resolved" and isinstance(side, dict):
# A resolved pair withholds only an empty positive.
result[role] = {**_side("unresolved"), "message": _EMPTY_POSITIVE}
else:
result[role] = _side(state if state != "resolved" else "unresolved")
return {
"prompts": result,
"resolution": {
"state": state,
"sampler_id": resolved.get("sampler_id"),
"reason": resolved.get("reason", ""),
"reason_code": resolved.get("reason_code"),
"node": resolved.get("node"),
},
resolution = {
"state": state,
"sampler_id": resolved.get("sampler_id"),
"reason": resolved.get("reason", ""),
"reason_code": resolved.get("reason_code"),
"node": resolved.get("node"),
}
return {"prompts": _explained(result, resolution), "resolution": resolution}
def inspect_prompt_roles(metadata: dict, saved: dict | None = None) -> dict:
+9 -6
View File
@@ -151,6 +151,9 @@ _ZEROED_SOURCE = "zeroed-conditioning"
_MAX_BRANCH_CANDIDATES = 32
# Statically undecidable text (a search, a list output): never pick an item.
_RUNTIME_REASON = "Text is decided at run time"
# Uncoded outcomes the inspector explains by their exact reason.
EMPTY_POSITIVE_REASON = "Positive prompt is empty"
MULTISTAGE_UNATTRIBUTED_REASON = "Multistage sampler ancestry is ambiguous"
_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"})
@@ -1189,7 +1192,7 @@ class _GraphResolver:
inputs.get(conditioning["positive"]), sampler_id, guider_id, depth
)
if not positive["text"].strip():
raise _Unresolved("Positive prompt is empty")
raise _Unresolved(EMPTY_POSITIVE_REASON)
negative_name = conditioning["negative"]
if negative_name is None or negative_name not in inputs:
return _result(
@@ -1227,7 +1230,7 @@ class _GraphResolver:
name, inputs = self.node(node_id)
if name in self.stage_classes:
# ``multistage-differs`` is reserved for when stage texts are compared.
raise _Unresolved("Multistage sampler ancestry is ambiguous")
raise _Unresolved(MULTISTAGE_UNATTRIBUTED_REASON)
if name in ("EmptyLatentImage", "EmptySD3LatentImage") and slot == 0:
return
if name == "Reroute" and slot == 0 and "value" in inputs:
@@ -1287,7 +1290,7 @@ class _GraphResolver:
if name in self.stage_classes:
# ``multistage-differs`` is reserved for when stage texts are
# compared (PR3); until then no stage's prompt is chosen.
raise _Unresolved("Multistage sampler ancestry is ambiguous")
raise _Unresolved(MULTISTAGE_UNATTRIBUTED_REASON)
if self.stage_shaped(name, inputs):
raise self.unsupported("Unsupported latent ancestry", node_id)
self.enqueue(queue, queued, inputs, depth + 1)
@@ -1344,7 +1347,7 @@ def _parameters(value: Any, max_text_length: int) -> dict:
positive, found, negative = body.partition(delimiter)
positive = positive.strip()
if not positive:
return _result("unresolved", reason="Positive prompt is empty")
return _result("unresolved", reason=EMPTY_POSITIVE_REASON)
provenance = {
"format": "a1111",
"field": "parameters",
@@ -1424,7 +1427,7 @@ def resolve_prompt_pair(
return _legacy_result(result) if profile.legacy else result
_SEMANTICS_UNAVAILABLE = "Inspection node semantics are unavailable"
SEMANTICS_UNAVAILABLE_REASON = "Inspection node semantics are unavailable"
def _inspection_semantics() -> Optional[Any]:
@@ -1495,7 +1498,7 @@ def _resolve(
if profile.semantics:
semantics = _inspection_semantics()
if semantics is None:
return _result("unresolved", reason=_SEMANTICS_UNAVAILABLE)
return _result("unresolved", reason=SEMANTICS_UNAVAILABLE_REASON)
normalized = {str(node_id): node for node_id, node in graph.items()}
if len(normalized) != len(graph):
return _result(
+8 -2
View File
@@ -9,6 +9,10 @@ const FACT_FIELDS = ['seed', 'steps', 'cfg', 'sampler', 'scheduler', 'model'];
const RECORD_FORMATS = ['original-text', 'reserialized-json', 'invalid', 'limited'];
const TEXTLESS_FORMATS = ['invalid', 'limited'];
const REASON_CODES = ['work-limit', 'depth-limit', 'text-limit', 'graph-ceiling', 'unsupported-node', 'runtime-value', 'composite', 'no-sampler', 'malformed', 'multistage-differs'];
// Only these codes name the blocking node; node IDs and class names are bounded.
const NAMED_CODES = ['unsupported-node', 'runtime-value', 'composite'];
const NAME_LENGTH = 200, MESSAGE_LENGTH = 512;
const boundedText = (value, limit) => typeof value === 'string' && value.length > 0 && value.length <= limit;
const object = value => value !== null && typeof value === 'object' && !Array.isArray(value);
const validScope = value => typeof value === 'string' && /^[a-f0-9]{64}$/.test(value) && value.length === 64;
@@ -77,7 +81,8 @@ function promptsFor(value) {
if (record === undefined) continue;
if (!object(record) || typeof record.status !== 'string' ||
(record.text !== undefined && record.text !== null && typeof record.text !== 'string') ||
(record.status === 'resolved' && !Object.hasOwn(confirmed, role))) throw new TypeError('Invalid inspection role');
(record.status === 'resolved' && !Object.hasOwn(confirmed, role)) ||
(record.message !== undefined && (record.status === 'resolved' || !boundedText(record.message, MESSAGE_LENGTH)))) throw new TypeError('Invalid inspection role');
}
return value;
}
@@ -141,7 +146,8 @@ function resolutionFor(value) {
const code = value.reason_code ?? null, node = value.node ?? null;
// Codes explain a non-resolved outcome; a blocking node is named only with its code.
if ((code !== null && (!REASON_CODES.includes(code) || ['resolved', 'absent'].includes(value.state))) ||
(node !== null && (code === null || !object(node) || typeof node.id !== 'string' || (node.class_type !== null && typeof node.class_type !== 'string')))) throw new TypeError('Invalid inspection resolution reason');
(node !== null && (!NAMED_CODES.includes(code) || !object(node) || Object.keys(node).sort().join() !== 'class_type,id' ||
!boundedText(node.id, NAME_LENGTH) || (node.class_type !== null && !boundedText(node.class_type, NAME_LENGTH))))) throw new TypeError('Invalid inspection resolution reason');
return value;
}
+1 -1
View File
@@ -114,7 +114,7 @@ export function mountPromptBlocks(container, {prompts, sender, parentScope, role
heading.append(actions);
const status = el('p', '', 'prompt-copy-status muted'); status.setAttribute('role', 'status');
block.append(status);
} else block.append(el('p', typeof record?.message === 'string' ? record.message : statuses[record?.status] || statuses.unresolved, 'muted'));
} else block.append(el('p', typeof record?.message === 'string' && record.message ? record.message : statuses[record?.status] || statuses.unresolved, 'muted'));
container.append(block);
}
if (includePairAction && Object.hasOwn(roles, 'positive') && Object.hasOwn(roles, 'negative')) {