feat(inspect): add a declarative inspection node semantics table

This commit is contained in:
Vito Sansevero
2026-10-19 15:00:33 -07:00
parent cbf46a7573
commit 5a6c0bdbc1
5 changed files with 1173 additions and 0 deletions
@@ -0,0 +1,35 @@
// Run with: node --test tests/js/
// comfy-metadata.js stays a classic script for the legacy pages, so its facts
// tables are inline literals. This test keeps them equal to the one semantics
// table (web/js/inspection-semantics.json) that Python also loads.
const test = require("node:test");
const assert = require("node:assert/strict");
const fs = require("node:fs");
const path = require("node:path");
const metadata = require("../../web/js/comfy-metadata.js");
const tablePath = path.join(__dirname, "..", "..", "web", "js", "inspection-semantics.json");
const semantics = JSON.parse(fs.readFileSync(tablePath, "utf8"));
function plain(value) {
if (value instanceof Set) return [...value];
if (Array.isArray(value)) return value.map(plain);
if (value && typeof value === "object") return Object.fromEntries(Object.entries(value).map(([key, item]) => [key, plain(item)]));
return value;
}
test("facts tables equal semantics.facts", () => {
assert.equal(typeof metadata._factsTables, "object", "comfy-metadata exposes its facts tables for this test");
assert.deepStrictEqual(plain(metadata._factsTables), semantics.facts);
});
test("facts guider providers are guiders in the semantics table", () => {
const providers = Object.entries(semantics.facts.configuration_outputs).filter(([, adapter]) => adapter.kind === "guider");
assert.ok(providers.length > 0);
for (const [name] of providers) assert.ok(Object.hasOwn(semantics.guiders, name), name);
});
test("the facts tables are not part of the legacy page contract", () => {
assert.deepStrictEqual(Object.keys(metadata).filter(key => !key.startsWith("_")).sort(), ["PLACEHOLDERS", "extractGenerationFacts", "extractGenerationParams"]);
assert.ok(Object.isFrozen(metadata._factsTables));
});
+553
View File
@@ -0,0 +1,553 @@
"""The declarative inspection node semantics table and its loader (#240, spec 7).
``web/js/inspection-semantics.json`` is the single table. Python validates it at
import and fails loudly; the browser facts tables are checked against its
``facts`` section by ``tests/js/inspection-semantics-sync.test.js``.
"""
import copy
import json
import tempfile
import unittest
from pathlib import Path
from types import MappingProxyType
from utils import inspection_semantics as semantics
from utils.inspection_semantics import SemanticsError, load_semantics
REPO = Path(__file__).resolve().parents[1]
TABLE = REPO / "web" / "js" / "inspection-semantics.json"
def _minimal():
"""A small valid table; mutation tests break exactly one thing in a copy."""
return {
"schema": 1,
"samplers": {
"KSampler": {
"strategy": "ksampler",
"conditioning": {"positive": "positive", "negative": "negative"},
"latent": "latent_image",
"facts": "ksampler",
}
},
"guiders": {"BasicGuider": {"strategy": "basic", "positive": "conditioning"}},
"conditioning": [
{
"class": "FluxGuidance",
"slot": 0,
"strategy": "passthrough",
"input": "conditioning",
}
],
"selectors": {},
"composites": {},
"encoders": [
{
"class": "CLIPTextEncode",
"slot": 0,
"strategy": "single",
"inputs": ["text"],
}
],
"strings": [],
"runtime": [],
"savers": {},
"non_file_outputs": [],
"facts": {
"configuration_outputs": {},
"model_outputs": {},
"model_passthroughs": [],
"scalar_outputs": {},
},
}
class _TableCase(unittest.TestCase):
def load_text(self, text):
# Windows: the file handle is closed before the directory is removed.
with tempfile.TemporaryDirectory() as root:
path = Path(root) / "semantics.json"
path.write_text(text, encoding="utf-8")
return load_semantics(path)
def load_doc(self, doc):
return self.load_text(json.dumps(doc))
def assertRejected(self, doc, fragment):
with self.assertRaises(SemanticsError) as caught:
self.load_doc(doc)
self.assertIn(fragment, str(caught.exception))
class SpecEntriesTest(unittest.TestCase):
def test_semantics_file_validates_and_lists_spec_entries(self):
table = load_semantics()
self.assertEqual(semantics.SEMANTICS_PATH, TABLE)
self.assertEqual(
semantics.SEMANTICS_PATH,
Path(semantics.__file__).resolve().parents[1]
/ "web"
/ "js"
/ "inspection-semantics.json",
)
pair = {"positive": "positive", "negative": "negative"}
for name in ("KSampler", "KSamplerAdvanced"):
self.assertEqual(
table.samplers[name],
{
"strategy": "ksampler",
"conditioning": pair,
"latent": "latent_image",
"facts": "ksampler",
},
)
self.assertEqual(
table.samplers["SamplerCustom"],
{
"strategy": "ksampler",
"conditioning": pair,
"latent": "latent_image",
"facts": "custom",
},
)
self.assertEqual(
table.samplers["SamplerCustomAdvanced"],
{
"strategy": "custom-advanced",
"conditioning": {"guider": "guider"},
"latent": "latent_image",
"facts": "custom-advanced",
},
)
# D19: FaceDetailer is a stage over its upstream image, ksampler-shaped facts.
self.assertEqual(
table.samplers["FaceDetailer"],
{
"strategy": "detailer-stage",
"conditioning": pair,
"image_stage": "image",
"facts": "ksampler",
},
)
self.assertEqual(
table.guiders["CFGGuider"],
{"strategy": "cfg", "positive": "positive", "negative": "negative"},
)
self.assertEqual(
table.guiders["BasicGuider"],
{
"strategy": "basic",
"positive": "conditioning",
"absent_reason": "Guider has no negative input",
},
)
# D17: DualCFGGuider has two positives and stays an unsupported node.
self.assertNotIn("DualCFGGuider", table.guiders)
self.assertNotIn("DualCFGGuider", table.all_classes)
single_input = (
"FluxGuidance",
"ReferenceLatent",
"ConditioningSetArea",
"ConditioningSetAreaPercentage",
"ConditioningSetAreaStrength",
"ConditioningSetMask",
"ConditioningSetTimestepRange",
"ControlNetApply",
)
for name in single_input:
self.assertEqual(
table.conditioning[(name, 0)],
{"strategy": "passthrough", "input": "conditioning"},
)
self.assertNotIn((name, 1), table.conditioning)
for name in (
"ControlNetApplyAdvanced",
"ControlNetApplySD3",
"InpaintModelConditioning",
"WanImageToVideo",
):
self.assertEqual(
table.conditioning[(name, 0)],
{"strategy": "passthrough", "input": "positive"},
)
self.assertEqual(
table.conditioning[(name, 1)],
{"strategy": "passthrough", "input": "negative"},
)
# Slot 2 (a latent) is not conditioning; spec 7.4 handles origins.
self.assertNotIn((name, 2), table.conditioning)
self.assertEqual(
table.conditioning[("ConditioningZeroOut", 0)],
{"strategy": "zeroed", "input": "conditioning"},
)
self.assertEqual(
table.selectors["easy ifElse"],
{
"strategy": "boolean-select",
"condition": "boolean",
"branches": {"true": "on_true", "false": "on_false"},
},
)
self.assertEqual(
table.composites["ConditioningCombine"],
{"strategy": "combine", "inputs": ("conditioning_1", "conditioning_2")},
)
for name in ("ConditioningConcat", "ConditioningAverage"):
self.assertEqual(
table.composites[name],
{
"strategy": "combine",
"inputs": ("conditioning_to", "conditioning_from"),
},
)
encoders = {
("CLIPTextEncode", 0): ("single", ("text",)),
("CLIPTextEncodeSDXL", 0): ("parts-equal", ("text_g", "text_l")),
("CLIPTextEncodeFlux", 0): ("parts-equal", ("clip_l", "t5xxl")),
("CLIPTextEncodeSD3", 0): ("parts-equal", ("clip_l", "clip_g", "t5xxl")),
("TextEncodeQwenImageEdit", 0): ("single", ("prompt",)),
("TextEncodeQwenImageEditPlus", 0): ("single", ("prompt",)),
("TextEncodeQwenImage21", 0): ("single", ("prompt",)),
("TextEncodeQwenImage21", 1): ("single", ("negative_prompt",)),
# D11: a plain text encoder, no composition.
("KikoTextEncode", 0): ("single", ("text",)),
}
for key, (strategy, inputs) in encoders.items():
self.assertEqual(
table.encoders[key], {"strategy": strategy, "inputs": inputs}, key
)
self.assertNotIn(("TextEncodeQwenImage21", 2), table.encoders)
self.assertEqual(
table.strings[("StringConstant", 0)],
{"strategy": "literal", "inputs": ("string",)},
)
self.assertEqual(
table.strings[("StringConstantMultiline", 0)],
{
"strategy": "strip-newlines",
"inputs": ("string",),
"params": {"flag": "strip_newlines"},
},
)
self.assertEqual(
table.strings[("JoinStringMulti", 0)],
{
"strategy": "join-multi",
"inputs": ("string_1",),
"params": {
"count": "inputcount",
"delimiter": "delimiter",
"list_flag": "return_list",
"input_prefix": "string_",
},
},
)
# D18: decided at run time.
self.assertEqual(table.runtime, frozenset({"PromptSearchList"}))
comfy = "{name}_{counter:05}_.{ext}"
for name, image in (
("SaveImage", "images"),
("SaveAnimatedWEBP", "images"),
("SaveAnimatedPNG", "images"),
("SaveVideo", "video"),
("SaveWEBM", "images"),
):
self.assertEqual(
table.savers[name],
{
"strategy": "comfy-counter",
"prefix_input": "filename_prefix",
"image_input": image,
"pattern": comfy,
},
)
self.assertEqual(
table.savers["VHS_VideoCombine"],
{
"strategy": "vhs-counter",
"prefix_input": "filename_prefix",
"image_input": "images",
"pattern": "{name}_{counter:05}.{ext}",
},
)
# Until its naming is verified in pinned source it stays terminal-unknown.
self.assertNotIn("KikoSaveImage", table.all_classes)
self.assertEqual(
table.non_file_outputs,
frozenset(
{
"PreviewImage",
"PreviewAny",
"SaveImageWebsocket",
"Image Comparer (rgthree)",
}
),
)
# Not listed this slice: they stay unsupported-node with the class named.
for name in (
"ImpactWildcardProcessor",
"RegexReplace",
"LoRAfromText",
"TextBoxMira",
"XYZPrompt",
):
self.assertNotIn(name, table.all_classes)
# Existing code adapters stay in code, unchanged.
for name in ("PromptManager", "PromptManagerPair", "PromptManagerText"):
self.assertNotIn(name, table.all_classes)
def test_facts_section_mirrors_the_browser_tables(self):
facts = load_semantics().facts
self.assertEqual(
facts["configuration_outputs"],
{
"RandomNoise": {"kind": "noise", "fields": ("noise_seed",)},
"KSamplerSelect": {"kind": "sampler", "fields": ("sampler_name",)},
"BasicScheduler": {"kind": "sigmas", "fields": ("steps", "scheduler")},
"CFGGuider": {"kind": "guider", "fields": ("cfg", "model")},
"BasicGuider": {"kind": "guider", "fields": ("model",)},
},
)
self.assertEqual(
facts["model_outputs"],
{
"CheckpointLoader": "ckpt_name",
"CheckpointLoaderSimple": "ckpt_name",
"unCLIPCheckpointLoader": "ckpt_name",
"UNETLoader": "unet_name",
},
)
self.assertEqual(
facts["model_passthroughs"], ("LoraLoader", "LoraLoaderModelOnly")
)
self.assertEqual(
facts["scalar_outputs"],
{
"PrimitiveInt": "number",
"PrimitiveFloat": "number",
"PrimitiveString": "string",
"PrimitiveStringMultiline": "string",
},
)
def test_module_loads_the_table_once_at_import(self):
self.assertEqual(semantics.SEMANTICS, load_semantics())
def test_loaded_table_is_frozen(self):
table = load_semantics()
self.assertIsInstance(table.samplers, MappingProxyType)
self.assertIsInstance(table.samplers["KSampler"], MappingProxyType)
with self.assertRaises(TypeError):
table.samplers["Other"] = {}
with self.assertRaises(TypeError):
table.samplers["KSampler"]["latent"] = "x"
with self.assertRaises(AttributeError):
table.samplers = {}
self.assertIsInstance(table.composites["ConditioningCombine"]["inputs"], tuple)
class ValidationTest(_TableCase):
def test_minimal_table_loads(self):
table = self.load_doc(_minimal())
self.assertEqual(table.encoders[("CLIPTextEncode", 0)]["inputs"], ("text",))
self.assertEqual(table.runtime, frozenset())
def test_unknown_strategy_rejected(self):
for category, mutate in (
("samplers", lambda d: d["samplers"]["KSampler"].update(strategy="dual")),
# A known strategy of another category is still unknown here.
("conditioning", lambda d: d["conditioning"][0].update(strategy="single")),
("encoders", lambda d: d["encoders"][0].update(strategy="guess")),
("guiders", lambda d: d["guiders"]["BasicGuider"].update(strategy=None)),
):
with self.subTest(category=category):
doc = _minimal()
mutate(doc)
self.assertRejected(doc, "unknown strategy")
def test_strategies_are_the_closed_spec_set(self):
self.assertEqual(
dict(semantics.STRATEGIES),
{
"samplers": frozenset(
{"ksampler", "custom-advanced", "detailer-stage"}
),
"guiders": frozenset({"cfg", "basic"}),
"conditioning": frozenset({"passthrough", "zeroed"}),
"selectors": frozenset({"boolean-select"}),
"composites": frozenset({"combine"}),
"encoders": frozenset({"single", "parts-equal", "pm-compose"}),
"strings": frozenset(
{"literal", "strip-newlines", "join-multi", "concat"}
),
"savers": frozenset({"comfy-counter", "vhs-counter"}),
},
)
with self.assertRaises(TypeError):
semantics.STRATEGIES["samplers"] = frozenset()
def test_duplicate_output_key_rejected(self):
doc = _minimal()
doc["encoders"].append(copy.deepcopy(doc["encoders"][0]))
self.assertRejected(doc, "duplicate")
doc = _minimal()
doc["conditioning"].append(
{
"class": "CLIPTextEncode",
"slot": 0,
"strategy": "passthrough",
"input": "conditioning",
}
)
self.assertRejected(doc, "duplicate")
doc = _minimal()
doc["runtime"] = ["KSampler"]
self.assertRejected(doc, "duplicate")
doc = _minimal()
doc["non_file_outputs"] = ["PreviewImage", "PreviewImage"]
self.assertRejected(doc, "duplicate")
# JSON objects with a repeated key would silently keep the last one.
text = json.dumps(_minimal())
entry = json.dumps(_minimal()["samplers"]["KSampler"])
text = text.replace(
'"samplers": {', '"samplers": {"KSampler": ' + entry + ", ", 1
)
with self.assertRaises(SemanticsError) as caught:
self.load_text(text)
self.assertIn("duplicate", str(caught.exception))
def test_slot_must_be_nonnegative_int(self):
for slot in (-1, 1.5, True, "0", None, 2**31):
with self.subTest(slot=slot):
doc = _minimal()
doc["encoders"][0]["slot"] = slot
self.assertRejected(doc, "slot")
doc = _minimal()
del doc["conditioning"][0]["slot"]
self.assertRejected(doc, "slot")
def test_missing_input_name_rejected(self):
cases = (
lambda d: d["samplers"]["KSampler"]["conditioning"].pop("negative"),
lambda d: d["samplers"]["KSampler"].pop("latent"),
lambda d: d["guiders"]["BasicGuider"].update(positive=""),
lambda d: d["conditioning"][0].pop("input"),
lambda d: d["encoders"][0].update(inputs=[]),
lambda d: d["encoders"][0].update(inputs=["text", "text"]),
lambda d: d["encoders"][0].update(inputs=[3]),
lambda d: d["encoders"][0].update(inputs="text"),
)
for index, mutate in enumerate(cases):
with self.subTest(case=index):
doc = _minimal()
mutate(doc)
self.assertRejected(doc, "input")
def test_entry_shape_is_strict(self):
cases = (
(lambda d: d.update(extra={}), "category"),
(lambda d: d.pop("guiders"), "category"),
(lambda d: d.update(schema=2), "schema"),
(lambda d: d["samplers"]["KSampler"].update(note="x"), "field"),
(lambda d: d["samplers"]["KSampler"].update(facts="flux"), "facts"),
(lambda d: d["guiders"]["BasicGuider"].update(negative="n"), "field"),
(lambda d: d["encoders"][0].update({"class": ""}), "class"),
(lambda d: d["facts"].update(extra={}), "facts"),
(
lambda d: d["facts"]["scalar_outputs"].update(PrimitiveInt="int"),
"facts",
),
)
for index, (mutate, fragment) in enumerate(cases):
with self.subTest(case=index):
doc = _minimal()
mutate(doc)
self.assertRejected(doc, fragment)
def test_strategy_specific_requirements(self):
doc = _minimal()
doc["samplers"]["X"] = {
"strategy": "custom-advanced",
"conditioning": {"positive": "positive", "negative": "negative"},
"latent": "latent_image",
"facts": "custom-advanced",
}
self.assertRejected(doc, "input")
doc = _minimal()
doc["samplers"]["D"] = {
"strategy": "detailer-stage",
"conditioning": {"positive": "positive", "negative": "negative"},
"facts": "ksampler",
}
self.assertRejected(doc, "input")
doc = _minimal()
doc["guiders"]["G"] = {"strategy": "cfg", "positive": "positive"}
self.assertRejected(doc, "input")
doc = _minimal()
doc["encoders"][0].update(strategy="parts-equal", inputs=["text_g"])
self.assertRejected(doc, "input")
doc = _minimal()
doc["savers"]["S"] = {
"strategy": "comfy-counter",
"prefix_input": "filename_prefix",
"image_input": "images",
"pattern": "{name}_{counter:05}.{ext}",
}
self.assertRejected(doc, "pattern")
doc = _minimal()
doc["selectors"]["easy ifElse"] = {
"strategy": "boolean-select",
"condition": "boolean",
"branches": {"true": "on_true"},
}
self.assertRejected(doc, "input")
doc = _minimal()
doc["strings"].append(
{
"class": "JoinStringMulti",
"slot": 0,
"strategy": "join-multi",
"inputs": ["string_1"],
"params": {"count": "inputcount"},
}
)
self.assertRejected(doc, "params")
def test_unreadable_table_fails_loudly(self):
with self.assertRaises(SemanticsError):
self.load_text("{not json")
with self.assertRaises(SemanticsError):
self.load_text("[]")
with self.assertRaises(SemanticsError):
self.load_text(
json.dumps(_minimal()).replace('"schema": 1', '"schema": NaN')
)
with tempfile.TemporaryDirectory() as root:
with self.assertRaises(SemanticsError):
load_semantics(Path(root) / "missing.json")
if __name__ == "__main__":
unittest.main()
+404
View File
@@ -0,0 +1,404 @@
"""Load and validate the declarative inspection node semantics table (#240).
The table lives once in ``web/js/inspection-semantics.json`` so a browser-local
resolver can load the same data. It maps node classes (or ``(class, output
slot)`` pairs) to a named strategy from the closed ``STRATEGIES`` set plus the
exact input names that strategy reads. Code implements each strategy; adding a
node that reuses one is a data-only change.
Only the inspection resolver profile consumes this table. The default
(ingestion) profile never reads it. The table is validated once at import and an
invalid table raises ``SemanticsError`` so a bad edit fails loudly instead of
silently changing what the inspector resolves. Every class appears in exactly
one category; classes that are not listed stay ``unsupported-node``.
"""
import json
from dataclasses import dataclass
from pathlib import Path
from types import MappingProxyType
from typing import Any, Callable, Dict, FrozenSet, Mapping, Optional, Tuple, Union
SEMANTICS_PATH = (
Path(__file__).resolve().parents[1] / "web" / "js" / "inspection-semantics.json"
)
SCHEMA_VERSION = 1
_NAME_LENGTH = 128
# Comfy output slots are small; this only rejects nonsense, never a real node.
_MAX_SLOT = 255
STRATEGIES: Mapping[str, FrozenSet[str]] = MappingProxyType(
{
"samplers": frozenset({"ksampler", "custom-advanced", "detailer-stage"}),
"guiders": frozenset({"cfg", "basic"}),
"conditioning": frozenset({"passthrough", "zeroed"}),
"selectors": frozenset({"boolean-select"}),
"composites": frozenset({"combine"}),
"encoders": frozenset({"single", "parts-equal", "pm-compose"}),
"strings": frozenset({"literal", "strip-newlines", "join-multi", "concat"}),
"savers": frozenset({"comfy-counter", "vhs-counter"}),
}
)
# Shapes of the facts each sampler class provides (consumed by comfy-metadata.js).
FACT_SHAPES = frozenset({"ksampler", "custom", "custom-advanced"})
SAVER_PATTERNS = MappingProxyType(
{
"comfy-counter": "{name}_{counter:05}_.{ext}",
"vhs-counter": "{name}_{counter:05}.{ext}",
}
)
_CONFIGURATION_KINDS = frozenset({"noise", "sampler", "sigmas", "guider"})
_SCALAR_KINDS = frozenset({"number", "string"})
_CLASS_CATEGORIES = ("samplers", "guiders", "selectors", "composites", "savers")
_SLOT_CATEGORIES = ("conditioning", "encoders", "strings")
_SET_CATEGORIES = ("runtime", "non_file_outputs")
_FACTS_TABLES = (
"configuration_outputs",
"model_outputs",
"model_passthroughs",
"scalar_outputs",
)
_CATEGORIES = frozenset(
("schema",) + _CLASS_CATEGORIES + _SLOT_CATEGORIES + _SET_CATEGORIES + ("facts",)
)
_STRING_PARAMS = MappingProxyType(
{
"literal": frozenset(),
"concat": frozenset(),
"strip-newlines": frozenset({"flag"}),
"join-multi": frozenset({"count", "delimiter", "list_flag", "input_prefix"}),
}
)
Frozen = Union[Mapping[str, Any], Tuple[Any, ...], str, int, None]
class SemanticsError(ValueError):
"""The semantics table is unreadable or violates its schema."""
@dataclass(frozen=True)
class Semantics:
"""Validated, deeply frozen semantics. Slot categories key by ``(class, slot)``."""
samplers: Mapping[str, Mapping[str, Any]]
guiders: Mapping[str, Mapping[str, Any]]
conditioning: Mapping[Tuple[str, int], Mapping[str, Any]]
selectors: Mapping[str, Mapping[str, Any]]
composites: Mapping[str, Mapping[str, Any]]
encoders: Mapping[Tuple[str, int], Mapping[str, Any]]
strings: Mapping[Tuple[str, int], Mapping[str, Any]]
runtime: FrozenSet[str]
savers: Mapping[str, Mapping[str, Any]]
non_file_outputs: FrozenSet[str]
facts: Mapping[str, Any]
all_classes: FrozenSet[str]
def _freeze(value: Any) -> Frozen:
if isinstance(value, dict):
return MappingProxyType({key: _freeze(item) for key, item in value.items()})
if isinstance(value, list):
return tuple(_freeze(item) for item in value)
return value
def _reject_duplicate_keys(pairs: list) -> dict:
result: Dict[str, Any] = {}
for key, value in pairs:
if key in result:
raise SemanticsError(f"duplicate key {key!r}")
result[key] = value
return result
def _reject_constant(name: str) -> None:
raise SemanticsError(f"non-finite number {name} is not allowed")
def _name(value: Any, where: str, what: str = "input") -> str:
if not isinstance(value, str) or not value or len(value) > _NAME_LENGTH:
raise SemanticsError(f"{where}: missing or invalid {what} name {value!r}")
return value
def _names(value: Any, where: str, minimum: int = 1) -> None:
if not isinstance(value, list) or len(value) < minimum:
raise SemanticsError(f"{where}: needs at least {minimum} input name(s)")
for item in value:
_name(item, where)
if len(set(value)) != len(value):
raise SemanticsError(f"{where}: repeated input name")
def _fields(
entry: Any, where: str, required: FrozenSet[str], optional: FrozenSet[str]
) -> None:
if not isinstance(entry, dict):
raise SemanticsError(f"{where}: entry must be an object")
missing = sorted(required - entry.keys())
if missing:
raise SemanticsError(f"{where}: missing input field(s) {missing}")
unknown = sorted(entry.keys() - required - optional)
if unknown:
raise SemanticsError(f"{where}: unknown field(s) {unknown}")
def _strategy(entry: Any, category: str, where: str) -> str:
strategy = entry.get("strategy") if isinstance(entry, dict) else None
if not isinstance(strategy, str) or strategy not in STRATEGIES[category]:
raise SemanticsError(f"{where}: unknown strategy {strategy!r}")
return strategy
def _check_sampler(entry: dict, strategy: str, where: str) -> None:
stage = "image_stage" if strategy == "detailer-stage" else "latent"
_fields(
entry,
where,
frozenset({"strategy", "conditioning", stage, "facts"}),
frozenset(),
)
_name(entry[stage], where)
keys = {"guider"} if strategy == "custom-advanced" else {"positive", "negative"}
conditioning = entry["conditioning"]
_fields(conditioning, where + ".conditioning", frozenset(keys), frozenset())
for key in keys:
_name(conditioning[key], where + ".conditioning")
if entry["facts"] not in FACT_SHAPES:
raise SemanticsError(f"{where}: unknown facts shape {entry['facts']!r}")
def _check_guider(entry: dict, strategy: str, where: str) -> None:
if strategy == "cfg":
_fields(
entry, where, frozenset({"strategy", "positive", "negative"}), frozenset()
)
_name(entry["negative"], where)
else:
# Negative absent by design; its reason is fixed text, not an input.
_fields(
entry,
where,
frozenset({"strategy", "positive"}),
frozenset({"absent_reason"}),
)
if "absent_reason" in entry:
_name(entry["absent_reason"], where, "absent reason")
_name(entry["positive"], where)
def _check_selector(entry: dict, strategy: str, where: str) -> None:
_fields(entry, where, frozenset({"strategy", "condition", "branches"}), frozenset())
_name(entry["condition"], where)
branches = entry["branches"]
_fields(branches, where + ".branches", frozenset({"true", "false"}), frozenset())
_name(branches["true"], where)
_name(branches["false"], where)
def _check_composite(entry: dict, strategy: str, where: str) -> None:
_fields(entry, where, frozenset({"strategy", "inputs"}), frozenset())
_names(entry["inputs"], where, minimum=2)
def _check_saver(entry: dict, strategy: str, where: str) -> None:
_fields(
entry,
where,
frozenset({"strategy", "prefix_input", "image_input", "pattern"}),
frozenset(),
)
_name(entry["prefix_input"], where)
_name(entry["image_input"], where)
if entry["pattern"] != SAVER_PATTERNS[strategy]:
raise SemanticsError(f"{where}: pattern does not match {strategy}")
def _check_conditioning(entry: dict, strategy: str, where: str) -> None:
_fields(entry, where, frozenset({"strategy", "input"}), frozenset())
_name(entry["input"], where)
def _check_encoder(entry: dict, strategy: str, where: str) -> None:
_fields(entry, where, frozenset({"strategy", "inputs"}), frozenset({"compose"}))
_names(entry["inputs"], where, minimum=2 if strategy == "parts-equal" else 1)
if "compose" in entry and not isinstance(entry["compose"], bool):
raise SemanticsError(f"{where}: compose must be a boolean")
def _check_string(entry: dict, strategy: str, where: str) -> None:
required = _STRING_PARAMS[strategy]
if required:
_fields(entry, where, frozenset({"strategy", "inputs", "params"}), frozenset())
params = entry["params"]
if not isinstance(params, dict) or set(params) != required:
raise SemanticsError(f"{where}: params must be exactly {sorted(required)}")
for value in params.values():
_name(value, where + ".params")
else:
_fields(entry, where, frozenset({"strategy", "inputs"}), frozenset())
_names(entry["inputs"], where)
_CHECKS: Mapping[str, Callable[[dict, str, str], None]] = MappingProxyType(
{
"samplers": _check_sampler,
"guiders": _check_guider,
"selectors": _check_selector,
"composites": _check_composite,
"savers": _check_saver,
"conditioning": _check_conditioning,
"encoders": _check_encoder,
"strings": _check_string,
}
)
class _Owners:
"""Each class belongs to one category; each (class, slot) is listed once."""
def __init__(self) -> None:
self._category: Dict[str, str] = {}
self._outputs: set = set()
def claim(self, name: str, category: str, slot: Optional[int] = None) -> None:
owner = self._category.setdefault(name, category)
if owner != category:
raise SemanticsError(f"duplicate class {name!r} in {owner} and {category}")
key = (name, slot)
if slot is not None and key in self._outputs:
raise SemanticsError(f"{category}: duplicate output key {key!r}")
self._outputs.add(key)
def classes(self) -> FrozenSet[str]:
return frozenset(self._category)
def _class_category(doc: dict, category: str, owners: _Owners) -> Mapping:
table = doc[category]
if not isinstance(table, dict):
raise SemanticsError(f"category {category} must be an object")
result = {}
for name, entry in table.items():
where = f"{category}[{name!r}]"
_name(name, where, "class")
strategy = _strategy(entry, category, where)
_CHECKS[category](entry, strategy, where)
owners.claim(name, category)
result[name] = _freeze(entry)
return MappingProxyType(result)
def _slot(entry: dict, where: str) -> int:
slot = entry.get("slot")
# bool is an int subclass; a JSON true is not an output slot.
if type(slot) is not int or not 0 <= slot <= _MAX_SLOT:
raise SemanticsError(f"{where}: slot must be a non-negative int, not {slot!r}")
return slot
def _slot_category(doc: dict, category: str, owners: _Owners) -> Mapping:
rows = doc[category]
if not isinstance(rows, list):
raise SemanticsError(f"category {category} must be a list")
result = {}
for index, row in enumerate(rows):
where = f"{category}[{index}]"
if not isinstance(row, dict):
raise SemanticsError(f"{where}: entry must be an object")
name = _name(row.get("class"), where, "class")
slot = _slot(row, where)
entry = {
key: value for key, value in row.items() if key not in ("class", "slot")
}
strategy = _strategy(entry, category, where)
_CHECKS[category](entry, strategy, where)
owners.claim(name, category, slot)
result[(name, slot)] = _freeze(entry)
return MappingProxyType(result)
def _set_category(doc: dict, category: str, owners: _Owners) -> FrozenSet[str]:
rows = doc[category]
if not isinstance(rows, list):
raise SemanticsError(f"category {category} must be a list")
seen: set = set()
for name in rows:
_name(name, category, "class")
if name in seen:
raise SemanticsError(f"{category}: duplicate class {name!r}")
seen.add(name)
owners.claim(name, category)
return frozenset(seen)
def _string_map(table: Any, where: str, allowed: Optional[FrozenSet[str]]) -> None:
if not isinstance(table, dict):
raise SemanticsError(f"{where} must be an object")
for name, value in table.items():
_name(name, where, "class")
_name(value, where)
if allowed is not None and value not in allowed:
raise SemanticsError(f"{where}[{name!r}]: unknown kind {value!r}")
def _facts(doc: dict) -> Mapping[str, Any]:
facts = doc["facts"]
if not isinstance(facts, dict) or set(facts) != set(_FACTS_TABLES):
raise SemanticsError(f"facts must have exactly {list(_FACTS_TABLES)}")
configuration = facts["configuration_outputs"]
if not isinstance(configuration, dict):
raise SemanticsError("facts.configuration_outputs must be an object")
for name, adapter in configuration.items():
where = f"facts.configuration_outputs[{name!r}]"
_name(name, where, "class")
_fields(adapter, where, frozenset({"kind", "fields"}), frozenset())
if adapter["kind"] not in _CONFIGURATION_KINDS:
raise SemanticsError(f"{where}: unknown kind {adapter['kind']!r}")
_names(adapter["fields"], where)
_string_map(facts["model_outputs"], "facts.model_outputs", None)
_string_map(facts["scalar_outputs"], "facts.scalar_outputs", _SCALAR_KINDS)
passthroughs = facts["model_passthroughs"]
_names(passthroughs, "facts.model_passthroughs", minimum=0)
return _freeze(facts)
def _validate(doc: Any) -> Semantics:
if not isinstance(doc, dict):
raise SemanticsError("semantics table must be an object")
if set(doc) != _CATEGORIES:
missing = sorted(_CATEGORIES - doc.keys())
unknown = sorted(doc.keys() - _CATEGORIES)
raise SemanticsError(f"category mismatch: missing {missing}, unknown {unknown}")
if type(doc["schema"]) is not int or doc["schema"] != SCHEMA_VERSION:
raise SemanticsError(f"unsupported schema {doc['schema']!r}")
owners = _Owners()
tables: Dict[str, Any] = {}
for category in _CLASS_CATEGORIES:
tables[category] = _class_category(doc, category, owners)
for category in _SLOT_CATEGORIES:
tables[category] = _slot_category(doc, category, owners)
for category in _SET_CATEGORIES:
tables[category] = _set_category(doc, category, owners)
return Semantics(facts=_facts(doc), all_classes=owners.classes(), **tables)
def load_semantics(path: Optional[Union[str, Path]] = None) -> Semantics:
"""Read and validate a semantics table; ``SemanticsError`` on any problem."""
source = Path(path) if path is not None else SEMANTICS_PATH
try:
text = source.read_text(encoding="utf-8")
doc = json.loads(
text,
object_pairs_hook=_reject_duplicate_keys,
parse_constant=_reject_constant,
)
except SemanticsError:
raise
except (OSError, UnicodeDecodeError, ValueError) as error:
raise SemanticsError(f"cannot read semantics table {source.name}: {error}")
return _validate(doc)
SEMANTICS = load_semantics()
+12
View File
@@ -405,6 +405,9 @@
const FACT_FIELDS = ["seed", "steps", "cfg", "sampler", "scheduler", "model"];
const GENERATION_SAMPLERS = new Set(["KSampler", "KSamplerAdvanced", "SamplerCustom", "SamplerCustomAdvanced"]);
const WORKFLOW_PASSTHROUGH = Symbol("inspection workflow passthrough");
// These four tables must equal the `facts` section of inspection-semantics.json
// (tests/js/inspection-semantics-sync.test.js). They stay inline literals because the
// legacy pages load this file as a classic script.
// Fixed output0 contracts from the pinned core node schemas, not input-name inference.
const SCALAR_OUTPUTS = Object.freeze({
PrimitiveInt: "number", PrimitiveFloat: "number",
@@ -575,6 +578,15 @@
}
const api = { extractGenerationParams, extractGenerationFacts, PLACEHOLDERS };
// For the semantics sync test only; not part of the legacy page contract.
Object.defineProperty(api, "_factsTables", {
value: Object.freeze({
configuration_outputs: CONFIGURATION_OUTPUTS,
model_outputs: MODEL_OUTPUTS,
model_passthroughs: MODEL_PASSTHROUGHS,
scalar_outputs: SCALAR_OUTPUTS,
}),
});
if (typeof module !== "undefined" && module.exports) module.exports = api;
else root.ComfyMetadata = api;
})(typeof window !== "undefined" ? window : globalThis);
+169
View File
@@ -0,0 +1,169 @@
{
"schema": 1,
"samplers": {
"KSampler": {
"strategy": "ksampler",
"conditioning": { "positive": "positive", "negative": "negative" },
"latent": "latent_image",
"facts": "ksampler"
},
"KSamplerAdvanced": {
"strategy": "ksampler",
"conditioning": { "positive": "positive", "negative": "negative" },
"latent": "latent_image",
"facts": "ksampler"
},
"SamplerCustom": {
"strategy": "ksampler",
"conditioning": { "positive": "positive", "negative": "negative" },
"latent": "latent_image",
"facts": "custom"
},
"SamplerCustomAdvanced": {
"strategy": "custom-advanced",
"conditioning": { "guider": "guider" },
"latent": "latent_image",
"facts": "custom-advanced"
},
"FaceDetailer": {
"strategy": "detailer-stage",
"conditioning": { "positive": "positive", "negative": "negative" },
"image_stage": "image",
"facts": "ksampler"
}
},
"guiders": {
"CFGGuider": { "strategy": "cfg", "positive": "positive", "negative": "negative" },
"BasicGuider": {
"strategy": "basic",
"positive": "conditioning",
"absent_reason": "Guider has no negative input"
}
},
"conditioning": [
{ "class": "FluxGuidance", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
{ "class": "ReferenceLatent", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
{ "class": "ConditioningSetArea", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
{ "class": "ConditioningSetAreaPercentage", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
{ "class": "ConditioningSetAreaStrength", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
{ "class": "ConditioningSetMask", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
{ "class": "ConditioningSetTimestepRange", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
{ "class": "ControlNetApply", "slot": 0, "strategy": "passthrough", "input": "conditioning" },
{ "class": "ControlNetApplyAdvanced", "slot": 0, "strategy": "passthrough", "input": "positive" },
{ "class": "ControlNetApplyAdvanced", "slot": 1, "strategy": "passthrough", "input": "negative" },
{ "class": "ControlNetApplySD3", "slot": 0, "strategy": "passthrough", "input": "positive" },
{ "class": "ControlNetApplySD3", "slot": 1, "strategy": "passthrough", "input": "negative" },
{ "class": "InpaintModelConditioning", "slot": 0, "strategy": "passthrough", "input": "positive" },
{ "class": "InpaintModelConditioning", "slot": 1, "strategy": "passthrough", "input": "negative" },
{ "class": "WanImageToVideo", "slot": 0, "strategy": "passthrough", "input": "positive" },
{ "class": "WanImageToVideo", "slot": 1, "strategy": "passthrough", "input": "negative" },
{ "class": "ConditioningZeroOut", "slot": 0, "strategy": "zeroed", "input": "conditioning" }
],
"selectors": {
"easy ifElse": {
"strategy": "boolean-select",
"condition": "boolean",
"branches": { "true": "on_true", "false": "on_false" }
}
},
"composites": {
"ConditioningCombine": { "strategy": "combine", "inputs": ["conditioning_1", "conditioning_2"] },
"ConditioningConcat": { "strategy": "combine", "inputs": ["conditioning_to", "conditioning_from"] },
"ConditioningAverage": { "strategy": "combine", "inputs": ["conditioning_to", "conditioning_from"] }
},
"encoders": [
{ "class": "CLIPTextEncode", "slot": 0, "strategy": "single", "inputs": ["text"] },
{ "class": "CLIPTextEncodeSDXL", "slot": 0, "strategy": "parts-equal", "inputs": ["text_g", "text_l"] },
{ "class": "CLIPTextEncodeFlux", "slot": 0, "strategy": "parts-equal", "inputs": ["clip_l", "t5xxl"] },
{ "class": "CLIPTextEncodeSD3", "slot": 0, "strategy": "parts-equal", "inputs": ["clip_l", "clip_g", "t5xxl"] },
{ "class": "TextEncodeQwenImageEdit", "slot": 0, "strategy": "single", "inputs": ["prompt"] },
{ "class": "TextEncodeQwenImageEditPlus", "slot": 0, "strategy": "single", "inputs": ["prompt"] },
{ "class": "TextEncodeQwenImage21", "slot": 0, "strategy": "single", "inputs": ["prompt"] },
{ "class": "TextEncodeQwenImage21", "slot": 1, "strategy": "single", "inputs": ["negative_prompt"] },
{ "class": "KikoTextEncode", "slot": 0, "strategy": "single", "inputs": ["text"] }
],
"strings": [
{ "class": "StringConstant", "slot": 0, "strategy": "literal", "inputs": ["string"] },
{
"class": "StringConstantMultiline",
"slot": 0,
"strategy": "strip-newlines",
"inputs": ["string"],
"params": { "flag": "strip_newlines" }
},
{
"class": "JoinStringMulti",
"slot": 0,
"strategy": "join-multi",
"inputs": ["string_1"],
"params": {
"count": "inputcount",
"delimiter": "delimiter",
"list_flag": "return_list",
"input_prefix": "string_"
}
}
],
"runtime": ["PromptSearchList"],
"savers": {
"SaveImage": {
"strategy": "comfy-counter",
"prefix_input": "filename_prefix",
"image_input": "images",
"pattern": "{name}_{counter:05}_.{ext}"
},
"SaveAnimatedWEBP": {
"strategy": "comfy-counter",
"prefix_input": "filename_prefix",
"image_input": "images",
"pattern": "{name}_{counter:05}_.{ext}"
},
"SaveAnimatedPNG": {
"strategy": "comfy-counter",
"prefix_input": "filename_prefix",
"image_input": "images",
"pattern": "{name}_{counter:05}_.{ext}"
},
"SaveVideo": {
"strategy": "comfy-counter",
"prefix_input": "filename_prefix",
"image_input": "video",
"pattern": "{name}_{counter:05}_.{ext}"
},
"SaveWEBM": {
"strategy": "comfy-counter",
"prefix_input": "filename_prefix",
"image_input": "images",
"pattern": "{name}_{counter:05}_.{ext}"
},
"VHS_VideoCombine": {
"strategy": "vhs-counter",
"prefix_input": "filename_prefix",
"image_input": "images",
"pattern": "{name}_{counter:05}.{ext}"
}
},
"non_file_outputs": ["PreviewImage", "PreviewAny", "SaveImageWebsocket", "Image Comparer (rgthree)"],
"facts": {
"configuration_outputs": {
"RandomNoise": { "kind": "noise", "fields": ["noise_seed"] },
"KSamplerSelect": { "kind": "sampler", "fields": ["sampler_name"] },
"BasicScheduler": { "kind": "sigmas", "fields": ["steps", "scheduler"] },
"CFGGuider": { "kind": "guider", "fields": ["cfg", "model"] },
"BasicGuider": { "kind": "guider", "fields": ["model"] }
},
"model_outputs": {
"CheckpointLoader": "ckpt_name",
"CheckpointLoaderSimple": "ckpt_name",
"unCLIPCheckpointLoader": "ckpt_name",
"UNETLoader": "unet_name"
},
"model_passthroughs": ["LoraLoader", "LoraLoaderModelOnly"],
"scalar_outputs": {
"PrimitiveInt": "number",
"PrimitiveFloat": "number",
"PrimitiveString": "string",
"PrimitiveStringMultiline": "string"
}
}
}