diff --git a/comfy_api/latest/_io.py b/comfy_api/latest/_io.py index 98d19d7e0..6eaf7850a 100644 --- a/comfy_api/latest/_io.py +++ b/comfy_api/latest/_io.py @@ -1196,50 +1196,164 @@ class DynamicCombo(ComfyTypeI): @comfytype(io_type="COMFY_DYNAMICSLOT_V3") class DynamicSlot(ComfyTypeI): + """A slot whose revealed inputs depend on the type connected upstream. + + Options dispatch on the type resolved by + :py:class:`comfy_execution.type_resolver.TypeResolver`: + + * ``Option(when=None, ...)`` — nothing connected to the slot. + * ``Option(when=io.AnyType, ...)`` — link present, upstream resolves to ``"*"`` + (e.g. Reroute, generic If/Else, V1 nodes that declare AnyType outputs). + * ``Option(when=io.Image, ...)`` — upstream resolves to ``IMAGE``. + * ``Option(when=[io.Image, io.Mask], ...)`` — share children across types. + * ``Option(when=io.MultiType.Input(...), ...)`` — same as the list form. + + On a connected slot the first option whose ``when`` set intersects the + resolved type set wins; on an unconnected slot the first ``when=None`` + option wins. No implicit "match anything I didn't enumerate" fallback — + declare ``when=io.AnyType`` if you want it. + + Known limitation: when an upstream node declares its output as ``AnyType`` + (Reroute, generic forwarders, many V1 utilities) the resolver can only + report ``"*"`` — it cannot introspect the runtime value to recover a more + specific type. Such links will always select the ``when=io.AnyType`` + branch (or no branch), never a concrete-type branch. + """ Type = dict[str, Any] - class Input(DynamicInput): - def __init__(self, slot: Input, inputs: list[Input], - display_name: str=None, tooltip: str=None, lazy: bool=None, extra_dict=None): - assert(not isinstance(slot, DynamicInput)) - self.slot = copy.copy(slot) - self.slot.display_name = slot.display_name if slot.display_name is not None else display_name - optional = True - self.slot.tooltip = slot.tooltip if slot.tooltip is not None else tooltip - self.slot.lazy = slot.lazy if slot.lazy is not None else lazy - self.slot.extra_dict = slot.extra_dict if slot.extra_dict is not None else extra_dict - super().__init__(slot.id, self.slot.display_name, optional, self.slot.tooltip, self.slot.lazy, self.slot.extra_dict) + class Option: + """One branch of inputs revealed when the slot's resolved type matches ``when``. + + ``when`` accepts: + * ``None`` — no link present + * ``io.AnyType`` — upstream resolved type is literally ``"*"`` + * a single ComfyType class (e.g. ``io.Image``) + * a list of ComfyType classes (shared branch across multiple types) + * a ``MultiType.Input`` instance (parsed via its ``.io_types``) + """ + def __init__(self, when: Any, inputs: list[Input]): + self.when = when self.inputs = inputs - self.force_input = None - # force widget inputs to have no widgets, otherwise this would be awkward - if isinstance(self.slot, WidgetInput): - self.force_input = True - self.slot.force_input = True + self._when_types = self._normalize_when(when) + + @staticmethod + def _normalize_when(when: Any) -> frozenset[str] | None: + """Normalize ``when`` to a ``frozenset[str]`` of io_types, or ``None`` for the unconnected case.""" + if when is None: + return None + if isinstance(when, type) and issubclass(when, _ComfyType): + return frozenset([when.io_type]) + if isinstance(when, MultiType.Input): + return frozenset(t.io_type for t in when.io_types) + if isinstance(when, Iterable) and not isinstance(when, str): + types: list[str] = [] + for t in when: + if not (isinstance(t, type) and issubclass(t, _ComfyType)): + raise ValueError( + f"DynamicSlot.Option: list entries must be ComfyType classes, got {t!r}" + ) + types.append(t.io_type) + if not types: + raise ValueError("DynamicSlot.Option: when=[] is not allowed; use when=None instead") + return frozenset(types) + raise ValueError( + "DynamicSlot.Option: when must be None, a ComfyType class, a list of ComfyType classes, " + f"or a MultiType.Input; got {when!r}" + ) + + def as_dict(self): + return { + "when": None if self._when_types is None else sorted(self._when_types), + "inputs": create_input_dict_v1(self.inputs), + } + + class Input(DynamicInput): + def __init__(self, id: str, options: list[DynamicSlot.Option], + display_name: str=None, tooltip: str=None, lazy: bool=None, extra_dict=None): + if not options: + raise ValueError("DynamicSlot.Input: at least one Option is required") + super().__init__(id, display_name, True, tooltip, lazy, extra_dict) + self.options = options + # Auto-derive the slot's declared connection type as the union of + # every non-None option's `when` set. Order is preserved per option, + # then deduplicated, so authors control the displayed precedence. + connected_types: list[str] = [] + for opt in options: + if opt._when_types is None: + continue + for t in opt._when_types: + if t not in connected_types: + connected_types.append(t) + if not connected_types: + raise ValueError( + "DynamicSlot.Input: at least one Option must have a non-None `when`; " + "a slot with only a `when=None` option can never be connected" + ) + self._slot_io_type = ",".join(connected_types) + + # NOTE: do NOT override get_io_type — parse_class_inputs uses the class + # io_type (COMFY_DYNAMICSLOT_V3) to dispatch into the dynamic expander. + # The auto-derived connection type is published via the `slotType` field + # in as_dict() so the frontend knows what links are accepted. def get_all(self) -> list[Input]: - return [self.slot] + self.inputs + seen_ids: set[str] = set() + out: list[Input] = [] + for opt in self.options: + for inp in opt.inputs: + if inp.id in seen_ids: + continue + seen_ids.add(inp.id) + out.append(inp) + return out def as_dict(self): return super().as_dict() | prune_dict({ - "slotType": str(self.slot.get_io_type()), - "inputs": create_input_dict_v1(self.inputs), - "forceInput": self.force_input, + "slotType": self._slot_io_type, + "options": [o.as_dict() for o in self.options], }) def validate(self): - self.slot.validate() - for input in self.inputs: - input.validate() + for opt in self.options: + for inp in opt.inputs: + inp.validate() + + @staticmethod + def _select_option(options: list[dict[str, Any]], live_input_types: dict[str, str] | None, + finalized_id: str, has_link: bool) -> dict[str, Any] | None: + """Pick the first option whose ``when`` matches the slot's current state. + + Matching is set intersection against the resolved type string split on + commas (so MultiType outputs like ``"IMAGE,MASK"`` work naturally). + """ + if not has_link: + for opt in options: + if opt["when"] is None: + return opt + return None + resolved = (live_input_types or {}).get(finalized_id, "*") + resolved_set = set(t.strip() for t in resolved.split(",")) + for opt in options: + when = opt["when"] + if when is None: + continue + if resolved_set & set(when): + return opt + return None @staticmethod def _expand_schema_for_dynamic(out_dict: dict[str, Any], live_inputs: dict[str, Any], value: tuple[str, dict[str, Any]], input_type: str, curr_prefix: list[str] | None, live_input_types: dict[str, str] | None = None): finalized_id = finalize_prefix(curr_prefix) - if finalized_id in live_inputs: - inputs = value[1]["inputs"] - parse_class_inputs(out_dict, live_inputs, inputs, curr_prefix, live_input_types) - # add self to inputs - out_dict[input_type][finalized_id] = value - out_dict["dynamic_paths"][finalized_id] = finalize_prefix(curr_prefix, curr_prefix[-1]) + options: list[dict[str, Any]] = value[1].get("options", []) + has_link = finalized_id in live_inputs and live_inputs[finalized_id] is not None + selected = DynamicSlot._select_option(options, live_input_types, finalized_id, has_link) + if selected is not None: + parse_class_inputs(out_dict, live_inputs, selected["inputs"], curr_prefix, live_input_types) + # Always advertise the slot itself so the connector renders even when no + # option matched (e.g. resolved type wasn't enumerated and there's no + # AnyType option). Unmatched cases just expand no children. + out_dict[input_type][finalized_id] = value + out_dict["dynamic_paths"][finalized_id] = finalize_prefix(curr_prefix, curr_prefix[-1]) @comfytype(io_type="IMAGECOMPARE") class ImageCompare(ComfyTypeI): diff --git a/comfy_execution/type_resolver.py b/comfy_execution/type_resolver.py index 96df4fc2a..31c3b3059 100644 --- a/comfy_execution/type_resolver.py +++ b/comfy_execution/type_resolver.py @@ -7,6 +7,13 @@ or unresolvable wildcards. Works against either a raw prompt dict or a ``DynamicPrompt``. All resolved values are strings, so resolver state is cross-process serializable. + +Known limitation: when an upstream node declares its output as ``AnyType`` +(``"*"``) — Reroute, generic If/Else, many V1 utility nodes — the resolver +returns ``"*"``. It has no way to introspect the runtime value to recover a +more specific type. Downstream consumers (e.g. :py:class:`DynamicSlot`) will +treat such links as AnyType and select their ``AnyType`` branch (or none), +not a concrete-type branch. """ from __future__ import annotations @@ -328,10 +335,8 @@ class TypeResolver: except Exception: return ANY_TYPE if isinstance(inp, io.DynamicSlot.Input): - try: - return inp.slot.get_io_type() - except Exception: - return ANY_TYPE + # Auto-derived slot type — comma-joined union of all option `when` types. + return getattr(inp, "_slot_io_type", ANY_TYPE) # DynamicCombo's "type" is a key selector, not a connection type. if isinstance(inp, io.DynamicCombo.Input): return ANY_TYPE diff --git a/execution.py b/execution.py index cd58f538e..93e0f185e 100644 --- a/execution.py +++ b/execution.py @@ -926,7 +926,14 @@ async def validate_inputs(prompt_id, prompt, item, validated, visiting=None, typ # frontend-injected type metadata get the same answer as the UI. received_type = type_resolver.resolve_output_type(o_id, val[1]) received_types[x] = received_type - if 'input_types' not in validate_function_inputs and not validate_node_input(received_type, input_type): + # DynamicSlot publishes its accepted connection types via the + # `slotType` field (auto-derived union of option `when` types). + # The declared input_type ("COMFY_DYNAMICSLOT_V3") is just the + # dispatch tag; validate against slotType instead. + effective_input_type = input_type + if input_type == _io.DynamicSlot.io_type and isinstance(extra_info, dict): + effective_input_type = extra_info.get("slotType", input_type) + if 'input_types' not in validate_function_inputs and not validate_node_input(received_type, effective_input_type): details = f"{x}, received_type({received_type}) mismatch input_type({input_type})" error = { "type": "return_type_mismatch", diff --git a/tests-unit/comfy_api_test/test_dynamic_slot.py b/tests-unit/comfy_api_test/test_dynamic_slot.py new file mode 100644 index 000000000..ef0f520c2 --- /dev/null +++ b/tests-unit/comfy_api_test/test_dynamic_slot.py @@ -0,0 +1,266 @@ +"""Unit tests for the redesigned ``DynamicSlot`` with type-keyed options.""" + +import pytest + +from comfy_api.latest import _io as io + + +def _opt(when, ids=None): + """Build an Option whose inputs are placeholder String widgets named after ids.""" + ids = ids or [] + inputs = [io.String.Input(name) for name in ids] + return io.DynamicSlot.Option(when=when, inputs=inputs) + + +# --------------------------------------------------------------------------- +# Option.when normalization +# --------------------------------------------------------------------------- + +def test_option_when_none(): + o = _opt(None, ["a"]) + assert o._when_types is None + assert o.as_dict()["when"] is None + + +def test_option_when_single_type(): + o = _opt(io.Image) + assert o._when_types == frozenset({"IMAGE"}) + assert o.as_dict()["when"] == ["IMAGE"] + + +def test_option_when_anytype(): + o = _opt(io.AnyType) + assert o._when_types == frozenset({"*"}) + assert o.as_dict()["when"] == ["*"] + + +def test_option_when_list(): + o = _opt([io.Image, io.Mask]) + assert o._when_types == frozenset({"IMAGE", "MASK"}) + # list form sorted for stable serialization + assert o.as_dict()["when"] == ["IMAGE", "MASK"] + + +def test_option_when_multitype_input(): + mt = io.MultiType.Input("x", types=[io.Image, io.Latent]) + o = _opt(mt) + assert o._when_types == frozenset({"IMAGE", "LATENT"}) + + +def test_option_when_empty_list_rejected(): + with pytest.raises(ValueError, match="when=\\[\\]"): + io.DynamicSlot.Option(when=[], inputs=[]) + + +def test_option_when_garbage_rejected(): + with pytest.raises(ValueError, match="when must be"): + io.DynamicSlot.Option(when="IMAGE", inputs=[]) + + +def test_option_when_list_with_non_comfytype_rejected(): + with pytest.raises(ValueError, match="list entries"): + io.DynamicSlot.Option(when=[io.Image, "MASK"], inputs=[]) + + +# --------------------------------------------------------------------------- +# DynamicSlot.Input construction and serialization +# --------------------------------------------------------------------------- + +def test_input_requires_at_least_one_option(): + with pytest.raises(ValueError, match="at least one Option"): + io.DynamicSlot.Input("x", options=[]) + + +def test_input_requires_non_none_option(): + with pytest.raises(ValueError, match="non-None `when`"): + io.DynamicSlot.Input("x", options=[_opt(None, ["a"])]) + + +def test_input_auto_derives_slot_type(): + inp = io.DynamicSlot.Input("x", options=[ + _opt(io.Image, ["a"]), + _opt(io.Mask, ["b"]), + _opt(None, ["c"]), + ]) + # Declared order preserved across non-None options; None contributes nothing. + # Note: get_io_type() intentionally still returns the dynamic class io_type + # (COMFY_DYNAMICSLOT_V3) so parse_class_inputs dispatches into the expander. + # The auto-derived slot type is exposed via the `slotType` field of as_dict() + # and via the private `_slot_io_type` attribute (used by the type resolver). + assert inp._slot_io_type == "IMAGE,MASK" + d = inp.as_dict() + assert d["slotType"] == "IMAGE,MASK" + assert len(d["options"]) == 3 + + +def test_input_includes_anytype_in_slot_type(): + inp = io.DynamicSlot.Input("x", options=[ + _opt(io.Image, ["a"]), + _opt(io.AnyType, ["b"]), + ]) + assert inp._slot_io_type == "IMAGE,*" + + +def test_input_get_all_dedups_inputs_by_id(): + inp = io.DynamicSlot.Input("x", options=[ + _opt(io.Image, ["shared", "image_only"]), + _opt(io.Mask, ["shared", "mask_only"]), + ]) + ids = [i.id for i in inp.get_all()] + assert ids == ["shared", "image_only", "mask_only"] + + +# --------------------------------------------------------------------------- +# Option selection +# --------------------------------------------------------------------------- + +def _select(options, live_input_types, has_link, finalized_id="x"): + """Convenience wrapper that runs the dispatch through the dict form (post-as_dict).""" + serialized = [o.as_dict() for o in options] + return io.DynamicSlot._select_option( + serialized, live_input_types, finalized_id, has_link + ) + + +def test_select_unconnected_picks_none_option(): + options = [_opt(io.Image, ["img_widgets"]), _opt(None, ["empty_widgets"])] + sel = _select(options, {}, has_link=False) + assert sel is not None + assert sel["when"] is None + + +def test_select_unconnected_with_no_none_option_returns_none(): + options = [_opt(io.Image, ["x"])] + assert _select(options, {}, has_link=False) is None + + +def test_select_concrete_type_match(): + options = [ + _opt(io.Image, ["a"]), + _opt(io.Mask, ["b"]), + _opt(io.AnyType, ["c"]), + ] + sel = _select(options, {"x": "MASK"}, has_link=True) + assert sel["when"] == ["MASK"] + + +def test_select_anytype_matches_wildcard_resolved(): + options = [_opt(io.Image, ["a"]), _opt(io.AnyType, ["c"])] + sel = _select(options, {"x": "*"}, has_link=True) + assert sel["when"] == ["*"] + + +def test_select_anytype_does_not_match_concrete(): + options = [_opt(io.AnyType, ["c"])] + # MASK isn't in any option's set; AnyType only matches "*". No expansion. + assert _select(options, {"x": "MASK"}, has_link=True) is None + + +def test_select_first_match_wins(): + options = [ + _opt([io.Image, io.Mask], ["both"]), + _opt(io.Image, ["image_only"]), + ] + # Resolved IMAGE matches both; first option wins. + sel = _select(options, {"x": "IMAGE"}, has_link=True) + assert sel["inputs"] + # The "both" option's first input is named "both" + first_input_id = next(iter(sel["inputs"]["required"].keys())) + assert first_input_id == "both" + + +def test_select_multitype_upstream_intersects_option_set(): + """When upstream declares MultiType like 'IMAGE,MASK', any option that + intersects with that set matches (first wins).""" + options = [ + _opt(io.Latent, ["latent_only"]), + _opt(io.Mask, ["mask_only"]), + ] + sel = _select(options, {"x": "IMAGE,MASK"}, has_link=True) + assert sel["when"] == ["MASK"] + + +def test_select_missing_resolved_falls_through_to_anytype(): + """If live_input_types lacks an entry for this slot but a link exists, + we treat it as '*' (resolver default for unresolvable links).""" + options = [_opt(io.Image, ["a"]), _opt(io.AnyType, ["c"])] + sel = _select(options, {}, has_link=True) + assert sel["when"] == ["*"] + + +# --------------------------------------------------------------------------- +# End-to-end expansion via _expand_schema_for_dynamic +# --------------------------------------------------------------------------- + +def test_expand_unconnected_path(): + """An unconnected slot with a `when=None` option expands that option's children.""" + inp = io.DynamicSlot.Input("x", options=[ + _opt(io.Image, ["image_widget"]), + _opt(None, ["empty_widget"]), + ]) + d = inp.as_dict() + value = (io.DynamicSlot.io_type, d) + out_dict = { + "required": {}, "optional": {}, "hidden": {}, + "dynamic_paths": {}, "dynamic_paths_default_value": {}, + } + io.DynamicSlot._expand_schema_for_dynamic( + out_dict=out_dict, + live_inputs={}, # no entry for "x" → unconnected + value=value, + input_type="optional", + curr_prefix=["x"], + live_input_types=None, + ) + # The slot itself is always advertised in the caller's bucket. + assert "x" in out_dict["optional"] + # Children land in their own buckets (required by default) with + # parent-prefixed ids. + assert "x.empty_widget" in out_dict["required"] + assert "x.image_widget" not in out_dict["required"] + + +def test_expand_typed_path(): + """A connected slot expands the matching type's children.""" + inp = io.DynamicSlot.Input("x", options=[ + _opt(io.Image, ["image_widget"]), + _opt(io.Mask, ["mask_widget"]), + ]) + d = inp.as_dict() + value = (io.DynamicSlot.io_type, d) + out_dict = { + "required": {}, "optional": {}, "hidden": {}, + "dynamic_paths": {}, "dynamic_paths_default_value": {}, + } + io.DynamicSlot._expand_schema_for_dynamic( + out_dict=out_dict, + live_inputs={"x": ["src_node", 0]}, # link present + value=value, + input_type="optional", + curr_prefix=["x"], + live_input_types={"x": "MASK"}, + ) + assert "x" in out_dict["optional"] + assert "x.mask_widget" in out_dict["required"] + assert "x.image_widget" not in out_dict["required"] + + +def test_expand_unmatched_concrete_still_advertises_slot(): + """Resolved type not in any option → no children, but the slot itself stays.""" + inp = io.DynamicSlot.Input("x", options=[_opt(io.Image, ["image_widget"])]) + d = inp.as_dict() + value = (io.DynamicSlot.io_type, d) + out_dict = { + "required": {}, "optional": {}, "hidden": {}, + "dynamic_paths": {}, "dynamic_paths_default_value": {}, + } + io.DynamicSlot._expand_schema_for_dynamic( + out_dict=out_dict, + live_inputs={"x": ["src_node", 0]}, + value=value, + input_type="optional", + curr_prefix=["x"], + live_input_types={"x": "LATENT"}, + ) + assert "x" in out_dict["optional"] + assert "x.image_widget" not in out_dict["required"] diff --git a/tests-unit/execution_test/test_type_resolver.py b/tests-unit/execution_test/test_type_resolver.py index 6d23223d0..48ddc4c20 100644 --- a/tests-unit/execution_test/test_type_resolver.py +++ b/tests-unit/execution_test/test_type_resolver.py @@ -358,6 +358,35 @@ def test_effective_slot_type_on_v3_plain_input(fake_nodes_module, TypeResolver): assert r.compute_live_input_types("n") == {"flag": "BOOLEAN"} +def test_effective_slot_type_peels_dynamic_slot(fake_nodes_module, TypeResolver): + """A DynamicSlot input reports its auto-derived slotType (union of `when` types).""" + from comfy_api.latest import _io as io + + class DSNode(io.ComfyNode): + @classmethod + def define_schema(cls): + return io.Schema( + node_id="DSNode", + inputs=[ + io.DynamicSlot.Input("slot", options=[ + io.DynamicSlot.Option(when=io.Image, inputs=[]), + io.DynamicSlot.Option(when=io.Latent, inputs=[]), + ]), + ], + outputs=[io.String.Output()], + ) + + @classmethod + def execute(cls, **kwargs): + return io.NodeOutput("") + + DSNode.GET_SCHEMA() + fake_nodes_module["DSNode"] = DSNode + prompt = {"n": {"class_type": "DSNode", "inputs": {}}} + r = TypeResolver(prompt) + assert r.get_declared_slot_io_type("n", "slot") == "IMAGE,LATENT" + + def test_compute_live_input_types_mixes_links_and_literals(fake_nodes_module, TypeResolver): fake_nodes_module["Src"] = _v1_node(("MODEL",)) fake_nodes_module["Sink"] = _v1_node(