From 5a6c0bdbc11e85c696e58801663db3aad320ba45 Mon Sep 17 00:00:00 2001 From: Vito Sansevero Date: Mon, 19 Oct 2026 09:20:37 -0700 Subject: [PATCH] feat(inspect): add a declarative inspection node semantics table --- tests/js/inspection-semantics-sync.test.js | 35 ++ tests/test_inspection_semantics.py | 553 +++++++++++++++++++++ utils/inspection_semantics.py | 404 +++++++++++++++ web/js/comfy-metadata.js | 12 + web/js/inspection-semantics.json | 169 +++++++ 5 files changed, 1173 insertions(+) create mode 100644 tests/js/inspection-semantics-sync.test.js create mode 100644 tests/test_inspection_semantics.py create mode 100644 utils/inspection_semantics.py create mode 100644 web/js/inspection-semantics.json diff --git a/tests/js/inspection-semantics-sync.test.js b/tests/js/inspection-semantics-sync.test.js new file mode 100644 index 0000000..0a6ee52 --- /dev/null +++ b/tests/js/inspection-semantics-sync.test.js @@ -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)); +}); diff --git a/tests/test_inspection_semantics.py b/tests/test_inspection_semantics.py new file mode 100644 index 0000000..fc35c89 --- /dev/null +++ b/tests/test_inspection_semantics.py @@ -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() diff --git a/utils/inspection_semantics.py b/utils/inspection_semantics.py new file mode 100644 index 0000000..23603cd --- /dev/null +++ b/utils/inspection_semantics.py @@ -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() diff --git a/web/js/comfy-metadata.js b/web/js/comfy-metadata.js index 044dc34..2166e5e 100644 --- a/web/js/comfy-metadata.js +++ b/web/js/comfy-metadata.js @@ -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); diff --git a/web/js/inspection-semantics.json b/web/js/inspection-semantics.json new file mode 100644 index 0000000..b8f538a --- /dev/null +++ b/web/js/inspection-semantics.json @@ -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" + } + } +}