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

fix(inspect): bound inspection work, not graph size (#244)

Browse Source 
* test(inspect): add synthetic inspection parity fixtures and legacy characterization

* test(inspect): pin test-time parity vectors and keep fixtures fully synthetic

* fix(inspect): bound resolver work instead of graph size

* fix(inspect): bound shared-node and stripped-text work in the inspection profile

Keep one memo per resolution whose entries carry their unfolded visit cost;
a memo hit charges that cost to the current candidate's allowance only.
Cache node validation and charge input keys, stripped characters and copied
provenance entries to a whole-resolution work budget before doing the work.
The default profile is unchanged.

* fix(inspect): keep memoized depth honest across inspection candidates

A memo entry now records its height below the reusing depth; a hit raises
depth-limit exactly where a fresh walk would, so each candidate's outcome
matches its anchored result and does not depend on trace or node order.
The default profile is unchanged.

* test(inspect): pin memo height propagation

Three samplers share one chain, each later one hanging below a memo hit.
The third candidate must report depth-limit in both trace orders and when
anchored; removing the memo-hit height propagation fails this test. The
provenance test now asserts a plain subset.

* fix(inspect): tolerate ComfyUI node metadata quirks in projections

Projections now take a kind. Each prompt node keeps exactly class_type and
inputs; _meta, is_changed and other node keys are validated and dropped, so
"is_changed": [NaN] no longer marks the whole prompt graph malformed. A
nonfinite value inside inputs, anywhere in a workflow, or in a retained
carrier member projects to null and is never reported as a number. The
carrier applies the prompt/workflow kinds to its members. Exact numbers are
unchanged (Decimal spelling, unsafe integers as strings) and duplicate keys
are still malformed. The 500-node gate is replaced by a 20,000-node backstop
(workflow count includes subgraph definitions). Raw records are untouched.

The f14 inspection vector's facts become known; default vectors are unchanged.

* fix(inspect): budget metadata per record so large workflows never hide prompts

PNG inspection no longer counts every text chunk against one shared 2 MiB
budget. Role records (prompt, parameters, the pair execution record) keep a
2 MiB budget together, encoded and decoded, and exceeding it still fails the
whole read. A workflow has its own 4 MiB budget: a chunk over it is skipped
by seek, and a compressed one stops inflating at budget + 1. Only the
workflow is then reported as limited (record format "limited", no raw text,
workflow_graph limited); prompts, facts and the prompt graph are unaffected.

Each text chunk's keyword is read first (at most 80 bytes). Chunks that are
not retained are skipped by length without being read or checksummed, so
memory stays bounded. Byte-identical duplicates of a retained record are
accepted and do not consume budget; differing duplicates still raise
"Duplicate prompt metadata", and a duplicate of a limited workflow stays
limited. The chunk-count limit is unchanged.

InspectionRead gains limited_records (empty by default). The reader is only
used by the inspection routes, so ingestion, scan and import are unchanged.

Vectors: f04-inspection (2.5 MB workflow, within the new workflow budget)
and f15-inspection (identical duplicate chunks) now read and resolve, with
the same outcome and facts as f05-inspection. Default vectors are unchanged.

* fix(inspect): keep prompt evidence longest under the response budget

* fix(inspect): bound identical duplicate chunk reads

* fix(inspect): fail closed when role duplicates exceed their budget

* fix(inspect): use the bounded inspection profile end to end

Inspection routes resolve with INSPECTION_PROFILE and expose reason_code
and node on the resolution; scan, live monitor and admin import keep the
default profile (pinned by spies over the real resolver on f03).

JS facts drop the entries and workflow node-count gates; per-lookup visit
sets and the server's node ceiling bound the work. The validator accepts
the textless `limited` workflow format, follows the workflow-first
response omission order, and checks the closed reason-code vocabulary.
The file inspector labels limited records and disables their workflow
download.

vectors.json: 12 inspection vectors gain their reason codes; f03 facts
become known and f01 facts ambiguous. Default vectors are unchanged. A
new parity test asserts vectors.json facts through the workspace JS.

* chore(tools): add read-only inspection corpus tool

tools/inspection_corpus.py runs the inspection route pipeline and the
workspace JS facts of a chosen code tree over a media folder, read-only,
and reports the #240 invariants H1 to H8 with aggregate counts. Rows hold
statuses and SHA-256 digests of resolved role texts, never file names or
prompt text unless --with-paths is given. Outputs inside the root or a code
tree are refused.

The facts-over-vectors test now fails instead of skipping when CI is set
and Node.js is missing.

* docs(testing): document bounded inspection work

* fix(tools): fail the corpus run when facts or inspection did not run

Rows whose facts were never computed (no or exiting Node.js) now fail H3,
and a new harness line fails on any tool exception or missing facts, so a
run where nothing was measured can no longer exit 0. Unreadable files stay
inspector outcomes. A missing --node binary is refused, an exited harness
is reaped, H8 has a 50 ms floor for tiny baselines, and JSONL rows are
sorted by id.

* docs(inspect): correct ceiling wording and bandit annotations

Describe the two 20,000-node ceilings (prompt entries, workflow UI plus
subgraph nodes) in TESTING.md, and use bare nosec ids so Bandit stops
warning about unparsed test names.

* test(inspect): make parity observation tuples explicit
This commit is contained in:
Vito
2026-10-19 09:54:29 -07:00
committed by GitHub
parent 44b6ed2420
commit 048d599386
46 changed files with 6186 additions and 192 deletions
Download Patch File Download Diff File Expand all files Collapse all files
utils/inspection_response.py
+118 -40
View File
@@ -3,7 +3,7 @@
import json
import math
from decimal import Decimal, InvalidOperation
from typing import TYPE_CHECKING
from typing import TYPE_CHECKING, Literal
if TYPE_CHECKING:
from .prompt_inspector import InspectionRead
@@ -14,7 +14,13 @@ _MAX_SAFE_INTEGER = 2**53 - 1
_RECORD_KEYS = frozenset(
{"prompt", "workflow", "parameters", "comfyassets_prompt_pairs"}
)
_FORMATS = frozenset({"original-text", "reserialized-json", "invalid"})
_FORMATS = frozenset({"original-text", "reserialized-json", "invalid", "limited"})
_PROMPT_NODE_KEYS = frozenset({"class_type", "inputs"})
MAX_PROJECTION_DEPTH = 64
# Backstop, not an expected outcome: executable API nodes for a prompt; UI nodes
# including subgraph definitions for a workflow.
GRAPH_NODE_CEILING = 20_000
ProjectionKind = Literal["prompt", "workflow", "carrier"]
class _ProjectionLimit(ValueError):
@@ -30,23 +36,27 @@ def _pairs(items: list[tuple]) -> dict:
return result
def _nonfinite(value: str) -> None:
raise ValueError("Nonfinite JSON number")
class _Nonfinite:
"""A parsed ``NaN``/``Infinity`` literal: never a fact, projected as null."""
def _normalize(
value: object, depth: int = 1, *, retain: bool = True, carrier: bool = False
) -> object:
"""Walk bounded source once; unrelated carrier fields are only validated."""
_NONFINITE = _Nonfinite()
def _nonfinite(value: str) -> _Nonfinite:
return _NONFINITE
def _normalize(value: object, depth: int = 1, *, retain: bool = True) -> object:
"""Walk bounded source once; unretained values are only validated as JSON."""
if isinstance(value, (dict, list)):
if depth > 64:
if depth > MAX_PROJECTION_DEPTH:
raise _ProjectionLimit()
if isinstance(value, dict):
result = {} if retain else None
for key, child in value.items():
keep = retain and (not carrier or key in ("prompt", "workflow"))
normalized = _normalize(child, depth + 1, retain=keep)
if keep:
normalized = _normalize(child, depth + 1, retain=retain)
if retain:
result[key] = normalized
else:
result = [] if retain else None
@@ -55,17 +65,65 @@ def _normalize(
if retain:
result.append(normalized)
return result
if not retain or value is _NONFINITE:
return None
if isinstance(value, Decimal):
if not value.is_finite():
raise ValueError("Nonfinite JSON number")
# str keeps large exponents compact and never rounds through a float.
return str(value) if retain else None
return str(value)
if type(value) is int and abs(value) > _MAX_SAFE_INTEGER:
return str(value) if retain else None
return value if retain else None
return str(value)
return value
def _strict_projection(text: str, *, carrier: bool = False) -> object:
def _project_prompt(tree: object, depth: int = 1) -> dict:
"""Each API node keeps exactly ``class_type`` and ``inputs``.
Other node keys (``_meta``, ``is_changed``, future keys) are validated and
dropped, so a nonfinite value there never reaches the projection.
"""
if not isinstance(tree, dict):
raise ValueError("Prompt graph must be an object of nodes")
if depth + 1 > MAX_PROJECTION_DEPTH: # the root and its node objects
raise _ProjectionLimit()
result = {}
for node_id, node in tree.items():
if not isinstance(node, dict):
result[node_id] = _normalize(node, depth + 1)
continue
projected = {}
for key, child in node.items():
keep = key in _PROMPT_NODE_KEYS
normalized = _normalize(child, depth + 2, retain=keep)
if keep:
projected[key] = normalized
result[node_id] = projected
return result
def _project(tree: object, kind: str, depth: int = 1) -> object:
if kind == "prompt":
return _project_prompt(tree, depth)
if kind == "workflow":
return _normalize(tree, depth)
if not isinstance(tree, dict):
raise ValueError("Malformed carrier")
result = {}
for key, member in tree.items():
if key not in ("prompt", "workflow"):
_normalize(member, depth + 1, retain=False)
elif isinstance(member, str):
result[key] = member # projected by its own kind when read
elif key == "prompt" and not isinstance(member, dict):
result[key] = _normalize(member, depth + 1)
else:
result[key] = _project(member, key, depth + 1)
return result
def _strict_projection(text: str, *, kind: ProjectionKind) -> object:
"""Parse exactly (Decimal, no duplicate keys) and project by record kind."""
try:
tree = json.loads(
text,
@@ -73,34 +131,35 @@ def _strict_projection(text: str, *, carrier: bool = False) -> object:
object_pairs_hook=_pairs,
parse_constant=_nonfinite,
)
if carrier and not isinstance(tree, dict):
raise ValueError("Malformed carrier")
return _normalize(tree, carrier=carrier)
return _project(tree, kind)
except RecursionError:
raise _ProjectionLimit() from None
except InvalidOperation:
raise ValueError("JSON decimal exceeds supported range") from None
def _node_count(value: dict, *, workflow: bool) -> int:
if not workflow:
return len(value)
nodes = value.get("nodes")
count = len(nodes) if isinstance(nodes, list) else 0
definitions = value.get("definitions")
subgraphs = definitions.get("subgraphs") if isinstance(definitions, dict) else None
for subgraph in subgraphs if isinstance(subgraphs, list) else ():
inner = subgraph.get("nodes") if isinstance(subgraph, dict) else None
count += len(inner) if isinstance(inner, list) else 0
return count
def _graph(value: object, *, workflow: bool = False) -> dict:
if not isinstance(value, dict):
return {"status": "malformed", "value": None}
nodes = value.get("nodes") if workflow else value
if isinstance(nodes, (dict, list)) and len(nodes) > 500:
if _node_count(value, workflow=workflow) > GRAPH_NODE_CEILING:
return {"status": "limited", "value": None}
return {"status": "parsed", "value": value}
def _metadata(inspection: "InspectionRead | None") -> dict:
raw = (
{}
if inspection is None
else {
key: value
for key, value in inspection.metadata.items()
if key in _RECORD_KEYS and isinstance(value, str)
}
)
formats = (
{}
if inspection is None
@@ -110,6 +169,22 @@ def _metadata(inspection: "InspectionRead | None") -> dict:
if key in _RECORD_KEYS and value in _FORMATS
}
)
if inspection is not None:
# A record over its read budget has no text: like ``invalid``, raw omits
# it, and its graph is ``limited`` rather than parsed from nothing.
for key in inspection.limited_records & _RECORD_KEYS:
formats[key] = "limited"
raw = (
{}
if inspection is None
else {
key: value
for key, value in inspection.metadata.items()
if key in _RECORD_KEYS
and isinstance(value, str)
and formats.get(key) != "limited"
}
)
result = {
"status": (
"unreadable"
@@ -124,7 +199,7 @@ def _metadata(inspection: "InspectionRead | None") -> dict:
carrier, failed = None, None
if inspection is not None and inspection.video_carrier_text is not None:
try:
carrier = _strict_projection(inspection.video_carrier_text, carrier=True)
carrier = _strict_projection(inspection.video_carrier_text, kind="carrier")
except _ProjectionLimit:
failed = "limited"
except (ValueError, TypeError, OverflowError):
@@ -132,8 +207,8 @@ def _metadata(inspection: "InspectionRead | None") -> dict:
for key, field in (("prompt", "prompt_graph"), ("workflow", "workflow_graph")):
graph = {"status": "absent", "value": None}
if key in raw or key in formats:
if failed:
graph["status"] = failed
if failed or formats.get(key) == "limited":
graph["status"] = failed or "limited"
elif formats.get(key) == "invalid":
graph["status"] = "malformed"
else:
@@ -141,11 +216,11 @@ def _metadata(inspection: "InspectionRead | None") -> dict:
if carrier is not None:
value = carrier.get(key)
if isinstance(value, str):
value = _strict_projection(value)
value = _strict_projection(value, kind=key)
elif formats.get(key) == "reserialized-json":
raise _ProjectionLimit()
else:
value = _strict_projection(raw[key])
value = _strict_projection(raw[key], kind=key)
graph = _graph(value, workflow=key == "workflow")
except _ProjectionLimit:
graph["status"] = "limited"
@@ -259,11 +334,14 @@ def _encode_inspection(
"library": _library_projection(library),
}
body = _bytes(response)
if len(body) > MAX_RESPONSE_BYTES:
for field in ("prompt_graph", "workflow_graph"):
if metadata[field]["status"] != "absent":
metadata[field] = {"status": "limited", "value": None}
body = _bytes(response)
# Omission order keeps prompt evidence longest: the workflow graph first,
# then the prompt graph, then all raw records atomically.
for field in ("workflow_graph", "prompt_graph"):
if len(body) <= MAX_RESPONSE_BYTES:
break
if metadata[field]["status"] != "absent":
metadata[field] = {"status": "limited", "value": None}
body = _bytes(response)
if len(body) > MAX_RESPONSE_BYTES:
metadata.update(
raw={},
utils/prompt_inspector.py
+186 -51
View File
@@ -1,26 +1,44 @@
"""Read prompt evidence without inference, indexing, or decoding image pixels."""
import hashlib
import io
import json
import struct
import subprocess # nosec B404
import threading
import zlib
from dataclasses import dataclass
from dataclasses import dataclass, field as dataclass_field
from pathlib import Path
from typing import Any
from PIL import Image
from .pair_execution import EXECUTION_KEY
from .prompt_pair_resolver import resolve_prompt_pair
from .prompt_pair_resolver import (
INSPECTION_PROFILE,
ResolverProfile,
resolve_prompt_pair,
)
from .video_metadata import ffprobe_path, is_video_path
MAX_METADATA_BYTES = 2 * 1024 * 1024
# Role records (prompt, parameters, the pair execution record) share one
# budget, encoded and decoded: roles cannot be trusted from a partial record,
# so exceeding it fails the whole read. A workflow has its own budget; past it
# only the workflow is ``limited``. Other text chunks are never retained.
ROLE_METADATA_BYTES = 2 * 1024 * 1024
WORKFLOW_METADATA_BYTES = 4 * 1024 * 1024
# Total bytes read for byte-identical duplicate chunks of one file, per class
# so workflow repeats can never starve role records: two budgets each, making
# repeated chunks cost bounded I/O instead of file size.
DUPLICATE_WORKFLOW_BYTES = 2 * WORKFLOW_METADATA_BYTES
DUPLICATE_ROLE_BYTES = 2 * ROLE_METADATA_BYTES
MAX_METADATA_BYTES = ROLE_METADATA_BYTES # video stdout bound, unchanged here
MAX_KEYWORD_BYTES = 80 # a 1-79 byte PNG keyword plus its NUL separator
MAX_HEADER_BYTES = 8 * 1024 * 1024
MAX_PNG_CHUNKS = 10_000
_ROLE_METADATA_KEYS = frozenset({"prompt", "parameters", EXECUTION_KEY})
_METADATA_KEYS = _ROLE_METADATA_KEYS | {"workflow"}
_TEXT_CHUNKS = (b"tEXt", b"zTXt", b"iTXt")
_SIGNATURE = b"\x89PNG\r\n\x1a\n"
@@ -38,6 +56,8 @@ class InspectionRead:
format: str | None
width: int | None
height: int | None
# Records present but over their budget: no text, format ``limited``.
limited_records: frozenset[str] = dataclass_field(default_factory=frozenset)
@dataclass(frozen=True)
@@ -47,24 +67,33 @@ class _VideoInspection:
carrier_text: str | None
def _inflate(data: bytes) -> bytes:
def _inflate(data: bytes, limit: int) -> bytes | None:
"""Inflate at most ``limit + 1`` bytes; ``None`` when the output is over limit."""
inflater = zlib.decompressobj()
result = inflater.decompress(data, MAX_METADATA_BYTES + 1)
if len(result) > MAX_METADATA_BYTES or not inflater.eof or inflater.unused_data:
result = inflater.decompress(data, limit + 1)
if len(result) > limit:
return None
if not inflater.eof or inflater.unused_data:
raise MetadataReadError("Compressed metadata is invalid or exceeds limits")
return result
def _png_text(kind: bytes, data: bytes) -> tuple[str, str]:
keyword, separator, body = data.partition(b"\0")
def _keyword(prefix: bytes) -> str:
keyword, separator, _ = prefix.partition(b"\0")
if not separator or not 1 <= len(keyword) <= 79:
raise MetadataReadError("Malformed PNG text metadata")
key = keyword.decode("latin-1")
return keyword.decode("latin-1")
def _png_text(kind: bytes, data: bytes, limit: int) -> tuple[str, str | None]:
"""Decode one retained text chunk; ``None`` text when it inflates past limit."""
_, _, body = data.partition(b"\0")
key = _keyword(data[:MAX_KEYWORD_BYTES])
encoding = "latin-1"
if kind == b"zTXt":
if not body or body[0] != 0:
raise MetadataReadError("Unsupported PNG text compression")
body = _inflate(body[1:])
body = _inflate(body[1:], limit)
elif kind == b"iTXt":
if len(body) < 2 or body[0] not in (0, 1):
raise MetadataReadError("Malformed international PNG metadata")
@@ -74,21 +103,91 @@ def _png_text(kind: bytes, data: bytes) -> tuple[str, str]:
raise MetadataReadError("Malformed international PNG metadata")
fields[0].decode("ascii")
fields[1].decode("utf-8")
body = _inflate(fields[2]) if flag else fields[2]
body = _inflate(fields[2], limit) if flag else fields[2]
encoding = "utf-8"
if body is None:
return key, None
if b"\0" in body:
raise MetadataReadError("Malformed PNG text metadata")
return key, body.decode(encoding)
def _png_metadata(stream: Any, size: int) -> tuple[dict, int, int]:
"""CRC-check text and structural chunks, seeking past compressed pixels.
class _PngRecords:
"""Retained records with per-class budgets and byte-identical duplicates."""
def __init__(self) -> None:
self.found: dict[str, str] = {}
self.limited: set[str] = set()
self.digests: dict[str, tuple[int, bytes]] = {}
self.role_encoded = self.role_decoded = 0
self.duplicate_workflow = self.duplicate_role = 0
def skip(self, key: str, length: int) -> bool:
"""Decide from the keyword and length alone whether to seek past a chunk."""
if key not in _METADATA_KEYS or key in self.limited:
return True # never retained, or a duplicate of a limited workflow
if key in self.digests:
if self.digests[key][0] != length:
raise MetadataReadError("Duplicate prompt metadata")
if key == "workflow":
if self.duplicate_workflow + length > DUPLICATE_WORKFLOW_BYTES:
self.limited.add(key)
self.found.pop(key, None)
return True
self.duplicate_workflow += length
else:
if self.duplicate_role + length > DUPLICATE_ROLE_BYTES:
raise MetadataReadError("Metadata exceeds inspection limits")
self.duplicate_role += length
return False # same length: read and compare bytes
if key == "workflow":
if length > WORKFLOW_METADATA_BYTES:
self.limited.add(key)
return True
return False
self.role_encoded += length
if self.role_encoded > ROLE_METADATA_BYTES:
raise MetadataReadError("Metadata exceeds inspection limits")
return False
def add(self, kind: bytes, data: bytes) -> None:
key = _keyword(data[:MAX_KEYWORD_BYTES])
digest = hashlib.sha256(kind + data).digest()
if key in self.digests:
if self.digests[key][1] != digest:
raise MetadataReadError("Duplicate prompt metadata")
return
self.digests[key] = (len(data), digest)
workflow = key == "workflow"
limit = (
WORKFLOW_METADATA_BYTES
if workflow
else ROLE_METADATA_BYTES - self.role_decoded
)
_, value = _png_text(kind, data, limit)
size = None if value is None else len(value.encode("utf-8"))
if workflow:
if size is None or size > WORKFLOW_METADATA_BYTES:
self.limited.add(key)
return
else:
self.role_decoded += limit + 1 if size is None else size
if self.role_decoded > ROLE_METADATA_BYTES:
raise MetadataReadError("Metadata exceeds inspection limits")
self.found[key] = value
def _png_metadata(stream: Any, size: int) -> tuple[dict, frozenset, int, int]:
"""CRC-check retained text and structural chunks, seeking past everything else.
PNG text chunks may follow IDAT. Pillow's ``.text`` loads pixels to find
them; this bounded reader instead follows the W3C PNG chunk layout.
IDAT checksums/pixel validity are deliberately outside metadata inspection.
them; this bounded reader instead follows the W3C PNG chunk layout. A text
chunk's keyword is read first (at most ``MAX_KEYWORD_BYTES``); chunks that
are not retained, or are over their budget, are skipped by length, so their
bytes are never held in memory or checksummed. IDAT checksums/pixel
validity are deliberately outside metadata inspection.
"""
found, compressed_size, expanded_size = {}, 0, 0
records = _PngRecords()
saw_header = saw_pixels = False
for index in range(MAX_PNG_CHUNKS):
header = stream.read(8)
@@ -99,34 +198,32 @@ def _png_metadata(stream: Any, size: int) -> tuple[dict, int, int]:
raise MetadataReadError("Incomplete PNG chunk")
if index == 0 and (kind != b"IHDR" or length != 13):
raise MetadataReadError("Missing PNG header")
if kind in (b"tEXt", b"zTXt", b"iTXt", b"IHDR", b"IEND"):
compressed_size += length
if compressed_size > MAX_METADATA_BYTES:
raise MetadataReadError("Metadata exceeds inspection limits")
data, checksum = stream.read(length), stream.read(4)
if zlib.crc32(kind + data) != int.from_bytes(checksum, "big"):
raise MetadataReadError("PNG metadata checksum mismatch")
if kind == b"IHDR":
if saw_header or not all(struct.unpack(">II", data[:8])):
raise MetadataReadError("Malformed PNG header")
saw_header = True
width, height = struct.unpack(">II", data[:8])
elif kind == b"IEND":
if length or not saw_pixels:
raise MetadataReadError("Incomplete PNG image")
return found, width, height
else:
key, value = _png_text(kind, data)
expanded_size += len(value.encode("utf-8"))
if expanded_size > MAX_METADATA_BYTES:
raise MetadataReadError("Metadata exceeds inspection limits")
if key in _METADATA_KEYS:
if key in found:
raise MetadataReadError("Duplicate prompt metadata")
found[key] = value
if kind in _TEXT_CHUNKS:
prefix = stream.read(min(length, MAX_KEYWORD_BYTES))
if records.skip(_keyword(prefix), length):
stream.seek(length - len(prefix) + 4, 1)
continue
data = prefix + stream.read(length - len(prefix))
elif kind in (b"IHDR", b"IEND"):
data = stream.read(length)
else:
saw_pixels |= kind == b"IDAT"
stream.seek(length + 4, 1)
continue
checksum = stream.read(4)
if zlib.crc32(kind + data) != int.from_bytes(checksum, "big"):
raise MetadataReadError("PNG metadata checksum mismatch")
if kind == b"IHDR":
if saw_header or not all(struct.unpack(">II", data[:8])):
raise MetadataReadError("Malformed PNG header")
saw_header = True
width, height = struct.unpack(">II", data[:8])
elif kind == b"IEND":
if length or not saw_pixels:
raise MetadataReadError("Incomplete PNG image")
return records.found, frozenset(records.limited), width, height
else:
records.add(kind, data)
raise MetadataReadError("PNG chunk count exceeds inspection limits")
@@ -222,7 +319,7 @@ def _video_metadata(path: Path) -> _VideoInspection:
def read_inspection_file(path: Path) -> InspectionRead:
"""Read bounded metadata and header facts without decoding media pixels."""
try:
formats, carrier = {}, None
formats, carrier, limited = {}, None, frozenset()
format_name = width = height = None
if is_video_path(path):
video = _video_metadata(path)
@@ -235,7 +332,9 @@ def read_inspection_file(path: Path) -> InspectionRead:
with path.open("rb") as stream:
signature = stream.read(8)
if signature == _SIGNATURE:
metadata, width, height = _png_metadata(stream, path.stat().st_size)
metadata, limited, width, height = _png_metadata(
stream, path.stat().st_size
)
format_name = "PNG"
else:
stream.seek(0)
@@ -243,7 +342,7 @@ def read_inspection_file(path: Path) -> InspectionRead:
with Image.open(io.BytesIO(header)) as image:
metadata = dict(image.info)
format_name, (width, height) = image.format, image.size
result, size = {}, 0
result, role_size, limited = {}, 0, set(limited)
for key in _METADATA_KEYS:
if key not in metadata:
continue
@@ -253,12 +352,31 @@ def read_inspection_file(path: Path) -> InspectionRead:
continue
if not isinstance(value, str):
raise MetadataReadError("Malformed prompt metadata")
size += len(value.encode("utf-8", "surrogatepass"))
if size > MAX_METADATA_BYTES:
raise MetadataReadError("Metadata exceeds inspection limits")
size = len(value.encode("utf-8", "surrogatepass"))
if key == "workflow":
if size > WORKFLOW_METADATA_BYTES:
limited.add(key)
continue
else:
role_size += size
if role_size > ROLE_METADATA_BYTES:
raise MetadataReadError("Metadata exceeds inspection limits")
result[key] = value
formats.setdefault(key, "original-text")
return InspectionRead(result, formats, carrier, format_name, width, height)
if _ROLE_METADATA_KEYS.intersection(limited):
# A limited role record would vanish and let fallback evidence win.
raise MetadataReadError("Metadata exceeds inspection limits")
for key in limited:
formats[key] = "limited"
return InspectionRead(
result,
formats,
carrier,
format_name,
width,
height,
frozenset(limited),
)
except MetadataReadError:
raise
except (
@@ -282,8 +400,19 @@ def _side(status: str, text: str | None = None, source: str | None = None) -> di
return {"status": status, "text": text, "source": source}
def inspect_prompt_evidence(metadata: dict, saved: dict | None = None) -> dict:
"""Keep uncertainty visible; a saved association never replaces file evidence."""
def inspect_prompt_evidence(
metadata: dict,
saved: dict | None = None,
*,
profile: ResolverProfile = INSPECTION_PROFILE,
) -> dict:
"""Keep uncertainty visible; a saved association never replaces file evidence.
Inspection routes resolve with ``INSPECTION_PROFILE``: bounded work, not
graph size. Ingestion (scan, live monitor, admin import) calls the resolver
directly with its default profile. ``resolution`` carries the closed
``reason_code`` and, for a blocking node, ``node: {id, class_type}``.
"""
metadata = {
key: value for key, value in metadata.items() if key in _ROLE_METADATA_KEYS
}
@@ -291,6 +420,8 @@ def inspect_prompt_evidence(metadata: dict, saved: dict | None = None) -> dict:
"state": "unresolved",
"sampler_id": None,
"reason": "Prompt evidence is malformed",
"reason_code": "malformed",
"node": None,
}
if not metadata:
result = {role: _side("absent") for role in ("positive", "negative")}
@@ -311,6 +442,8 @@ def inspect_prompt_evidence(metadata: dict, saved: dict | None = None) -> dict:
"state": "absent",
"sampler_id": None,
"reason": "No parsed prompt metadata",
"reason_code": None,
"node": None,
},
}
canonical = dict(metadata)
@@ -329,7 +462,7 @@ def inspect_prompt_evidence(metadata: dict, saved: dict | None = None) -> dict:
},
"resolution": resolution,
}
resolved = resolve_prompt_pair(canonical)
resolved = resolve_prompt_pair(canonical, profile=profile)
state = resolved["state"]
result = {}
for role in ("positive", "negative"):
@@ -350,6 +483,8 @@ def inspect_prompt_evidence(metadata: dict, saved: dict | None = None) -> dict:
"state": state,
"sampler_id": resolved.get("sampler_id"),
"reason": resolved.get("reason", ""),
"reason_code": resolved.get("reason_code"),
"node": resolved.get("node"),
},
}
utils/prompt_pair_resolver.py
+418 -62
View File
@@ -14,10 +14,19 @@ Effective text uses a unique bounded execution record for traced paired nodes
when available; cached graph-only results remain explicitly inferred.
The resolver has no
ComfyUI, database, configuration or filesystem dependencies.
Two ``ResolverProfile`` values exist. ``DEFAULT_PROFILE`` (ingestion: scan, live
monitor, admin import) keeps the historical behavior exactly, including the
500-node gate, and only adds ``reason_code: None``. ``INSPECTION_PROFILE`` bounds
work instead of graph size: a per-candidate visit allowance, a total visit budget
shared by every candidate and attribution walk, a node ceiling backstop, and
bounds on built provenance and text. It also names why a result is not resolved
(``reason_code``, and ``node`` for the first unsupported node), and a prompt graph
without a supported sampler is ``unresolved`` (``no-sampler``), never ``absent``.
"""
import re
from dataclasses import dataclass
from dataclasses import dataclass, replace
from typing import Any, Dict, Mapping, Optional, Tuple
from .pair_execution import ExecutionEvidenceError, read_pair_execution
@@ -62,8 +71,109 @@ for _name in ("Text Multiline", "String Literal", "ShowText|pysssss"):
_DEFAULT_ADAPTERS[(_name, 0)] = OutputAdapter("text", "string")
@dataclass(frozen=True)
class ResolverProfile:
"""Resolution bounds. ``legacy_node_gate`` set means the historical code path.
``max_path_entries`` and ``max_built_chars`` bound memory across the whole
resolution (provenance entries and newly built text). ``max_scanned_units``
bounds linear work across the whole resolution: input keys validated or
scanned, characters stripped or joined, and provenance entries copied, each
charged before the work is done. ``None`` disables each bound.
``semantics`` and ``selection`` are reserved for later slices and stay off.
"""
name: str
per_candidate_visits: int
total_visits: int
max_depth: int
max_text_length: int
node_ceiling: Optional[int]
legacy_node_gate: Optional[int]
semantics: bool = False
selection: bool = False
max_path_entries: Optional[int] = None
max_built_chars: Optional[int] = None
max_scanned_units: Optional[int] = None
@property
def legacy(self) -> bool:
return self.legacy_node_gate is not None
DEFAULT_PROFILE = ResolverProfile(
name="default",
per_candidate_visits=500,
total_visits=500,
max_depth=64,
max_text_length=100_000,
node_ceiling=None,
legacy_node_gate=500,
)
INSPECTION_PROFILE = ResolverProfile(
name="inspection",
per_candidate_visits=2_000,
total_visits=20_000,
max_depth=64,
max_text_length=100_000,
node_ceiling=20_000,
legacy_node_gate=None,
max_path_entries=64 * 20_000,
max_built_chars=16 * 100_000,
# About four passes over the 2 MiB role-record budget.
max_scanned_units=8_000_000,
)
_SAMPLER_LIKE_INPUTS = ("latent_image", "guider", "positive", "negative")
_NO_SAMPLER_NAMES = 3
_CLASS_NAME_LENGTH = 64
class _Unresolved(Exception):
pass
"""A non-resolved outcome; ``code`` is the closed machine vocabulary."""
def __init__(
self, message: str, code: Optional[str] = None, node: Optional[dict] = None
) -> None:
super().__init__(message)
self.code = code
self.node = node
def _node_ref(node_id: str, class_type: Any) -> dict:
name = class_type[:_CLASS_NAME_LENGTH] if isinstance(class_type, str) else None
return {"id": node_id[:_CLASS_NAME_LENGTH], "class_type": name}
class _Budget:
"""Visits shared by every candidate, plus a per-candidate allowance.
With ``total`` equal to the allowance (the default profile) this is exactly
the historical single counter, so the reset per sampler is a no-op there.
"""
def __init__(self, total: int) -> None:
self.total = total
self.allowance = total
def reset(self, allowance: int) -> None:
self.allowance = allowance
@property
def remaining(self) -> int:
return min(self.total, self.allowance)
def spend(self) -> None:
# Checked before decrementing, so an exhausted allowance costs no total.
if self.total <= 0 or self.allowance <= 0:
raise _Unresolved("Resolution visit limit reached", "work-limit")
self.total -= 1
self.allowance -= 1
def charge(self, cost: int) -> None:
"""Charge a memoized subtree's visit cost to this candidate only."""
if cost > self.allowance:
raise _Unresolved("Resolution visit limit reached", "work-limit")
self.allowance -= cost
def _link(value: Any) -> bool:
@@ -85,6 +195,8 @@ def _result(
sampler_id: Optional[str] = None,
reason: str = "",
candidates: Optional[list] = None,
reason_code: Optional[str] = None,
node: Optional[dict] = None,
) -> Dict[str, Any]:
return {
"state": state,
@@ -93,9 +205,19 @@ def _result(
"sampler_id": sampler_id,
"reason": reason,
"candidates": candidates or [],
"reason_code": reason_code,
"node": node,
}
def _legacy_result(result: Dict[str, Any]) -> Dict[str, Any]:
"""The default profile's wire: today's keys plus ``reason_code: None``."""
legacy = {key: value for key, value in result.items() if key != "node"}
legacy["reason_code"] = None
legacy["candidates"] = [_legacy_result(item) for item in result["candidates"]]
return legacy
class _GraphResolver:
def __init__(
self,
@@ -105,10 +227,32 @@ class _GraphResolver:
max_depth: int,
max_text_length: int,
metadata: Optional[dict] = None,
*,
budget: Optional[_Budget] = None,
profile: ResolverProfile = DEFAULT_PROFILE,
) -> None:
# ``max_visits`` is the per-candidate allowance; ``profile`` supplies only
# the path selection and whole-resolution bounds. ``resolve_text_input``
# relies on the legacy default profile here.
self.graph = graph
self.adapters = {**_DEFAULT_ADAPTERS, **adapters}
self.budget = max_visits
self.per_candidate_visits = max_visits
self.budget = budget if budget is not None else _Budget(max_visits)
self.budget.reset(max_visits)
self.legacy = profile.legacy
self.path_entries = profile.max_path_entries
self.built_chars = profile.max_built_chars
self.scanned = profile.max_scanned_units
self.checked: dict = {}
# Unfolded visit cost of each memo entry (inspection): a memo hit charges
# it to the current candidate, so each candidate's outcome is independent
# of tracing order while the shared work is done only once.
self.costs: dict = {}
# Height of each memo entry (deepest depth reached below it, relative to
# its own depth): a hit raises ``depth-limit`` exactly where a fresh walk
# from the reusing depth would, so depth too is independent of order.
self.heights: dict = {}
self.deepest = 0
self.max_depth = max_depth
self.max_text_length = max_text_length
self.memo: dict = {}
@@ -117,46 +261,99 @@ class _GraphResolver:
self.text_sources: dict = {}
def node(self, node_id: str) -> Tuple[str, dict]:
if self.legacy:
return self.checked_node(node_id)
if node_id not in self.checked:
try:
self.checked[node_id] = self.checked_node(node_id)
except _Unresolved as error:
if error.code == "work-limit":
raise
self.checked[node_id] = error
checked = self.checked[node_id]
if isinstance(checked, _Unresolved):
raise _Unresolved(str(checked), checked.code, checked.node)
return checked
def checked_node(self, node_id: str) -> Tuple[str, dict]:
node = self.graph.get(node_id)
if not isinstance(node, dict) or not isinstance(node.get("inputs"), dict):
raise _Unresolved("Missing or malformed graph node")
raise _Unresolved("Missing or malformed graph node", "malformed")
self.scan(len(node["inputs"]))
if any(not isinstance(key, str) for key in node["inputs"]):
raise _Unresolved("Malformed node input names")
raise _Unresolved("Malformed node input names", "malformed")
return str(node.get("class_type", "")), node["inputs"]
def scan(self, units: int) -> None:
"""Charge linear work before doing it (inspection; no-op when legacy)."""
if self.scanned is not None:
if units > self.scanned:
self.scanned = -1
raise _Unresolved("Resolution work limit reached", "work-limit")
self.scanned -= units
def unsupported(self, message: str, node_id: str) -> _Unresolved:
node = self.graph.get(node_id)
class_type = node.get("class_type") if isinstance(node, dict) else None
if isinstance(class_type, str):
class_type = class_type[:_CLASS_NAME_LENGTH]
return _Unresolved(message, "unsupported-node", _node_ref(node_id, class_type))
def bounded(self, text: str) -> str:
if len(text) > self.max_text_length:
raise _Unresolved("Resolved text exceeds the length limit")
raise _Unresolved("Resolved text exceeds the length limit", "text-limit")
return text
def built(self, text: str) -> str:
"""Charge newly built text against the profile's whole-resolution bound."""
if self.built_chars is not None:
self.built_chars -= len(text)
if self.built_chars < 0:
raise _Unresolved("Resolution work limit reached", "work-limit")
return text
def text(self, value: Any, kind: str, depth: int = 0) -> Tuple[str, list, str]:
if not self.legacy:
self.deepest = max(self.deepest, depth)
if depth >= self.max_depth:
raise _Unresolved("Resolution depth limit reached")
raise _Unresolved("Resolution depth limit reached", "depth-limit")
if isinstance(value, str) and kind == "string":
return self.bounded(value), [], value
if not _link(value):
raise _Unresolved("Expected a typed graph link or text input")
raise _Unresolved("Expected a typed graph link or text input", "malformed")
node_id, slot = str(value[0]), value[1]
key = (node_id, slot, kind)
if key in self.active:
raise _Unresolved("Cyclic prompt graph")
raise _Unresolved("Cyclic prompt graph", "malformed")
if key in self.memo:
if not self.legacy:
height = self.heights[key]
if depth + height >= self.max_depth:
raise _Unresolved("Resolution depth limit reached", "depth-limit")
self.budget.charge(self.costs[key])
self.deepest = max(self.deepest, depth + height)
return self.memo[key]
self.budget -= 1
if self.budget < 0:
raise _Unresolved("Resolution visit limit reached")
name, inputs = self.node(node_id)
self.active.add(key)
allowance = self.budget.allowance
outer_deepest = self.deepest
self.deepest = depth
try:
self.budget.spend()
name, inputs = self.node(node_id)
self.scan(len(inputs))
self.active.add(key)
if name == "Reroute" and slot == 0:
keys = [key for key in ("value", "input", "in") if key in inputs]
if len(keys) != 1:
raise _Unresolved("Reroute input is not uniquely identifiable")
raise _Unresolved(
"Reroute input is not uniquely identifiable", "malformed"
)
text, path, storage_text = self.text(inputs[keys[0]], kind, depth + 1)
elif (name, slot) in self.adapters:
adapter = self.adapters[(name, slot)]
if not isinstance(adapter, OutputAdapter) or adapter.kind != kind:
raise _Unresolved("Output slot has an incompatible type")
raise _Unresolved(
"Output slot has an incompatible type", "malformed"
)
evidence = (
self.execution_record(node_id, inputs)
if name == "PromptManagerPair"
@@ -186,11 +383,20 @@ class _GraphResolver:
after, after_path, _ = self.text(
inputs.get("append_text", ""), "string", depth + 1
)
text = self.bounded(
" ".join(
part
for part in (before.strip(), text, after.strip())
if part
self.scan(
len(before)
+ len(after)
+ len(before_path)
+ len(path)
+ len(after_path)
)
text = self.built(
self.bounded(
" ".join(
part
for part in (before.strip(), text, after.strip())
if part
)
)
)
path = before_path + path + after_path
@@ -212,38 +418,56 @@ class _GraphResolver:
str(inputs.get("clean_whitespace", "true")).lower() == "true"
)
if not keys or not isinstance(delimiter, str):
raise _Unresolved("Malformed concatenation")
raise _Unresolved("Malformed concatenation", "malformed")
parts, path, size = [], [], 0
for input_name in keys:
part, followed, _ = self.text(
inputs.get(input_name, ""), "string", depth + 1
)
# Charged before stripping: stripped-away text is work too.
self.scan(len(part) + len(followed))
if clean:
part = part.strip()
path.extend(followed)
if part or name == "StringConcatenate":
size += len(part) + (len(delimiter) if parts else 0)
if size > self.max_text_length:
raise _Unresolved("Resolved text exceeds the length limit")
raise _Unresolved(
"Resolved text exceeds the length limit", "text-limit"
)
parts.append(part)
text = delimiter.join(parts)
text = self.built(delimiter.join(parts))
storage_text = text
else:
raise _Unresolved("Unsupported node semantics or output slot")
raise self.unsupported(
"Unsupported node semantics or output slot", node_id
)
# Memoized paths stay bounded even for diamond-shaped string graphs.
self.scan(len(path) + 1)
path = [{"node_id": node_id, "output_slot": slot}] + path
if len(path) > self.max_depth * max(1, len(self.graph)):
raise _Unresolved("Provenance path limit reached")
if self.legacy:
path_limit = self.max_depth * max(1, len(self.graph))
else:
path_limit = self.max_depth * max(1, self.per_candidate_visits)
if self.path_entries is not None:
self.path_entries -= len(path)
if len(path) > path_limit or (
self.path_entries is not None and self.path_entries < 0
):
raise _Unresolved("Provenance path limit reached", "work-limit")
self.memo[key] = (text, path, storage_text)
self.costs[key] = allowance - self.budget.allowance
self.heights[key] = self.deepest - depth
return text, path, storage_text
finally:
self.deepest = max(outer_deepest, self.deepest)
self.active.discard(key)
def execution_record(self, node_id: str, inputs: dict) -> Optional[dict]:
try:
record = read_pair_execution(self.metadata, node_id)
except ExecutionEvidenceError as error:
raise _Unresolved(str(error)) from error
raise _Unresolved(str(error), "malformed") from error
if record is None:
return None
for name in ("text", "negative_text", "prepend_text", "append_text"):
@@ -273,7 +497,8 @@ class _GraphResolver:
expected = expected.strip()
if expected != record[name]:
raise _Unresolved(
"Execution evidence does not match recorded node inputs"
"Execution evidence does not match recorded node inputs",
"malformed",
)
return record
@@ -296,6 +521,14 @@ class _GraphResolver:
def pair(self, sampler_id: str) -> dict:
positive = negative = None
self.budget.reset(self.per_candidate_visits)
if not self.legacy and self.budget.total <= 0:
return _result(
"unresolved",
sampler_id=sampler_id,
reason="Resolution visit limit reached",
reason_code="work-limit",
)
try:
_, inputs = self.node(sampler_id)
# A second sampler in the latent ancestry denotes a multistage run;
@@ -315,31 +548,50 @@ class _GraphResolver:
negative = self.side(inputs["negative"], sampler_id)
return _result("resolved", positive, negative, sampler_id)
except _Unresolved as error:
return _result("unresolved", positive, negative, sampler_id, str(error))
return _result(
"unresolved",
positive,
negative,
sampler_id,
str(error),
reason_code=error.code,
node=error.node,
)
def latent_source(self, value: Any, seen: set, depth: int) -> None:
if not _link(value) or depth >= self.max_depth:
raise _Unresolved("Unknown latent ancestry")
code = "malformed" if not _link(value) else "depth-limit"
raise _Unresolved("Unknown latent ancestry", code)
node_id, slot = str(value[0]), value[1]
if node_id in seen or len(seen) >= max(0, self.budget):
raise _Unresolved("Cyclic or excessive latent ancestry")
if self.legacy:
if node_id in seen or len(seen) >= max(0, self.budget.remaining):
raise _Unresolved("Cyclic or excessive latent ancestry")
elif node_id in seen:
raise _Unresolved("Cyclic or excessive latent ancestry", "malformed")
else:
self.budget.spend()
seen.add(node_id)
name, inputs = self.node(node_id)
if name in _SAMPLERS:
# ``multistage-differs`` is reserved for when stage texts are compared.
raise _Unresolved("Multistage sampler ancestry is ambiguous")
if name in ("EmptyLatentImage", "EmptySD3LatentImage") and slot == 0:
return
if name == "Reroute" and slot == 0 and "value" in inputs:
self.latent_source(inputs["value"], seen, depth + 1)
return
raise _Unresolved("Unsupported latent ancestry")
raise self.unsupported("Unsupported latent ancestry", node_id)
def anchor(self, node_id: str) -> str:
seen = set()
slot = 0
while len(seen) < self.max_depth and len(seen) < self.budget:
while len(seen) < self.max_depth and (
not self.legacy or len(seen) < self.budget.remaining
):
if node_id in seen:
raise _Unresolved("Cyclic output ancestry")
raise _Unresolved("Cyclic output ancestry", "malformed")
if not self.legacy:
self.budget.spend()
seen.add(node_id)
name, inputs = self.node(node_id)
if name in _SAMPLERS and slot == 0:
@@ -353,19 +605,28 @@ class _GraphResolver:
}.get(name)
value = inputs.get(key) if key and slot == 0 else None
if not _link(value):
raise _Unresolved("Unsupported output ancestry")
raise self.unsupported("Unsupported output ancestry", node_id)
node_id, slot = str(value[0]), value[1]
raise _Unresolved("Output ancestry limit reached")
code = "depth-limit" if len(seen) >= self.max_depth else "work-limit"
raise _Unresolved("Output ancestry limit reached", code)
def _parameters(value: Any, max_text_length: int) -> dict:
if not isinstance(value, str) or not value:
return _result("absent", reason="No A1111 parameters")
if len(value) > max_text_length:
return _result("unresolved", reason="Parameters exceed the length limit")
return _result(
"unresolved",
reason="Parameters exceed the length limit",
reason_code="text-limit",
)
footers = list(re.finditer(r"(?m)^Steps:\s*\d+\s*,[^\n]*$", value))
if len(footers) != 1 or value[footers[0].end() :].strip():
return _result("unresolved", reason="A1111 settings footer is not unique")
return _result(
"unresolved",
reason="A1111 settings footer is not unique",
reason_code="malformed",
)
body = value[: footers[0].start()].rstrip("\r\n")
delimiter = "\nNegative prompt:"
if body.count(delimiter) > 1:
@@ -396,49 +657,136 @@ def _parameters(value: Any, max_text_length: int) -> dict:
return _result("resolved", positive_side, negative_side)
def _profile_with(
profile: ResolverProfile,
max_visits: Optional[int],
max_depth: Optional[int],
max_text_length: Optional[int],
) -> ResolverProfile:
"""Explicit ``max_*`` keywords override the profile they are given with."""
changes: Dict[str, Any] = {}
if max_visits is not None:
changes["per_candidate_visits"] = max_visits
if profile.legacy: # one historical number: gate, total and allowance
changes.update(total_visits=max_visits, legacy_node_gate=max_visits)
if max_depth is not None:
changes["max_depth"] = max_depth
if max_text_length is not None:
changes["max_text_length"] = max_text_length
return replace(profile, **changes) if changes else profile
def _sampler_like_classes(graph: dict) -> list:
"""Distinct classes, in graph order, of nodes shaped like samplers/guiders."""
names: list = []
for node in graph.values():
if not isinstance(node, dict) or not isinstance(node.get("inputs"), dict):
continue
name = node.get("class_type")
if not isinstance(name, str) or not name or name in names:
continue
if any(_link(node["inputs"].get(key)) for key in _SAMPLER_LIKE_INPUTS):
names.append(name[:_CLASS_NAME_LENGTH])
if len(names) == _NO_SAMPLER_NAMES:
break
return names
def resolve_prompt_pair(
metadata: Any,
*,
output_node_id: Optional[str] = None,
adapters: Optional[Mapping[Tuple[str, int], OutputAdapter]] = None,
max_visits: int = 500,
max_depth: int = 64,
max_text_length: int = 100_000,
profile: ResolverProfile = DEFAULT_PROFILE,
max_visits: Optional[int] = None,
max_depth: Optional[int] = None,
max_text_length: Optional[int] = None,
) -> Dict[str, Any]:
"""Resolve parsed metadata; a present API graph is authoritative over text.
Accepts ``{"prompt": api_graph, "parameters": optional_a1111_text}`` or a
direct API graph. An optional saver/sampler node ID anchors branch selection.
Unanchored multiple samplers are ambiguous, even when their texts are equal.
``profile`` selects the bounds; explicit ``max_*`` keywords override it.
"""
profile = _profile_with(profile, max_visits, max_depth, max_text_length)
result = _resolve(metadata, output_node_id, adapters, profile)
return _legacy_result(result) if profile.legacy else result
def _limit_result(profile: ResolverProfile, graph: dict) -> Optional[Dict[str, Any]]:
if profile.legacy:
if (
len(graph) > profile.legacy_node_gate
or profile.max_depth <= 0
or profile.max_text_length <= 0
):
return _result("unresolved", reason="Resolution limit reached")
return None
if profile.node_ceiling is not None and len(graph) > profile.node_ceiling:
return _result(
"unresolved",
reason="Prompt graph exceeds the inspection node ceiling",
reason_code="graph-ceiling",
)
if profile.max_depth <= 0:
return _result(
"unresolved", reason="Resolution limit reached", reason_code="depth-limit"
)
if profile.max_text_length <= 0:
return _result(
"unresolved", reason="Resolution limit reached", reason_code="text-limit"
)
return None
def _resolve(
metadata: Any,
output_node_id: Optional[str],
adapters: Optional[Mapping[Tuple[str, int], OutputAdapter]],
profile: ResolverProfile,
) -> Dict[str, Any]:
if not isinstance(metadata, dict) or not metadata:
return _result("absent", reason="No parsed prompt metadata")
if "prompt" in metadata:
graph = metadata["prompt"]
elif "parameters" in metadata:
return _parameters(metadata["parameters"], max_text_length)
return _parameters(metadata["parameters"], profile.max_text_length)
else:
graph = metadata
if not isinstance(graph, dict) or not graph:
return _result("unresolved", reason="API graph is missing or malformed")
if len(graph) > max_visits or max_depth <= 0 or max_text_length <= 0:
return _result("unresolved", reason="Resolution limit reached")
return _result(
"unresolved",
reason="API graph is missing or malformed",
reason_code="malformed",
)
limited = _limit_result(profile, graph)
if limited is not None:
return limited
normalized = {str(node_id): node for node_id, node in graph.items()}
if len(normalized) != len(graph):
return _result("unresolved", reason="Graph node IDs are not unique")
return _result(
"unresolved",
reason="Graph node IDs are not unique",
reason_code="malformed",
)
resolver = _GraphResolver(
normalized,
adapters or {},
max_visits,
max_depth,
max_text_length,
profile.per_candidate_visits,
profile.max_depth,
profile.max_text_length,
metadata if "prompt" in metadata else None,
budget=_Budget(profile.total_visits),
profile=profile,
)
if output_node_id is not None:
try:
sampler_ids = [resolver.anchor(str(output_node_id))]
except _Unresolved as error:
return _result("unresolved", reason=str(error))
return _result(
"unresolved", reason=str(error), reason_code=error.code, node=error.node
)
else:
sampler_ids = [
node_id
@@ -448,16 +796,7 @@ def resolve_prompt_pair(
and node["class_type"] in _SAMPLERS
]
if not sampler_ids:
has_pair_inputs = any(
isinstance(node, dict)
and isinstance(node.get("inputs"), dict)
and ("positive" in node["inputs"] or "negative" in node["inputs"])
for node in normalized.values()
)
return _result(
"unresolved" if has_pair_inputs else "absent",
reason="No supported sampler pair found",
)
return _no_sampler(normalized, profile)
if len(sampler_ids) > 1:
return _result(
"ambiguous",
@@ -467,6 +806,23 @@ def resolve_prompt_pair(
return resolver.pair(sampler_ids[0])
def _no_sampler(graph: dict, profile: ResolverProfile) -> Dict[str, Any]:
reason = "No supported sampler pair found"
if profile.legacy:
has_pair_inputs = any(
isinstance(node, dict)
and isinstance(node.get("inputs"), dict)
and ("positive" in node["inputs"] or "negative" in node["inputs"])
for node in graph.values()
)
return _result("unresolved" if has_pair_inputs else "absent", reason=reason)
# A prompt graph is a record: its prompt is unresolved, never "not recorded".
seen = _sampler_like_classes(graph)
if seen:
reason = f"{reason} (saw {', '.join(seen)})"
return _result("unresolved", reason=reason, reason_code="no-sampler")
def resolve_text_input(
graph: Any,
node_id: str,