DynamicOutputs: allow selectors to reach nested DynamicCombo/DynamicSlot

Schema.validate() now walks DynamicCombo.Input / DynamicSlot.Input options
recursively when building the selector-target map, so DynamicOutputs.ByKey /
BySlot can address a dynamic input nested inside another dynamic option via
its dotted path (e.g. selector="outer.inner_slot"). The runtime dispatch /
type resolver already handled dotted-path nested inputs; this just removes
the schema-level rejection.

Amp-Thread-ID: https://ampcode.com/threads/T-019e8568-f382-743d-a97f-0de3ff29d501
Co-authored-by: Amp <amp@ampcode.com>

comfy_api/latest/_io.py

@@ -1090,7 +1090,7 @@ class Autogrow(ComfyTypeI):
             self.template.validate()
 
     @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):
+    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):
         # NOTE: purposely do not include self in out_dict; instead use only the template inputs
         # need to figure out names based on template type
         is_names = ("names" in value[1]["template"])
@@ -1139,7 +1139,7 @@ class Autogrow(ComfyTypeI):
             finalized_prefix = finalize_prefix(curr_prefix)
             out_dict["dynamic_paths"][finalized_prefix] = finalized_prefix
             out_dict["dynamic_paths_default_value"][finalized_prefix] = DynamicPathsDefaultValue.EMPTY_DICT
-        parse_class_inputs(out_dict, live_inputs, new_dict, curr_prefix)
+        parse_class_inputs(out_dict, live_inputs, new_dict, curr_prefix, live_input_types)
 
 @comfytype(io_type="COMFY_DYNAMICCOMBO_V3")
 class DynamicCombo(ComfyTypeI):
@@ -1177,7 +1177,7 @@ class DynamicCombo(ComfyTypeI):
                     input.validate()
 
     @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):
+    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:
             key = live_inputs[finalized_id]
@@ -1189,57 +1189,488 @@ class DynamicCombo(ComfyTypeI):
                     selected_option = option
                     break
             if selected_option is not None:
-                parse_class_inputs(out_dict, live_inputs, selected_option["inputs"], curr_prefix)
+                parse_class_inputs(out_dict, live_inputs, selected_option["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])
 
 @comfytype(io_type="COMFY_DYNAMICSLOT_V3")
 class DynamicSlot(ComfyTypeI):
+    """A slot whose revealed inputs depend on the type connected upstream.
+
+    Each ``Option`` declares a ``when`` condition; the first option whose
+    condition matches the slot's resolved upstream type (or whose
+    ``when=None`` matches an empty slot) decides which child inputs are
+    exposed.
+
+    Each concrete type may appear in at most one option's ``when``, so the
+    matching branch is unambiguous. The unconnected case (``when=None``) is
+    its own bucket and may also appear at most once.
+
+    The AnyType limitation documented in
+    :py:mod:`comfy_execution.type_resolver` applies: an upstream output
+    declared as ``AnyType`` resolves to ``"*"`` and will only match a
+    ``when=io.AnyType`` option, never a concrete-type one.
+    """
     Type = dict[str, Any]
 
+    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] = None):
+            self.when = when
+            self.inputs = inputs or []
+            # ``_when_types`` is the ordered tuple of io_types (deterministic);
+            # ``_when_set`` is the same content as a set for fast matching.
+            self._when_types = self._normalize_when(when)
+            self._when_set: frozenset[str] | None = (
+                None if self._when_types is None else frozenset(self._when_types)
+            )
+
+        @staticmethod
+        def _normalize_when(when: Any) -> tuple[str, ...] | None:
+            """Normalize ``when`` to an ordered, deduplicated tuple of io_types, or ``None`` for the unconnected case."""
+            if when is None:
+                return None
+            if isinstance(when, type) and issubclass(when, _ComfyType):
+                return (when.io_type,)
+            if isinstance(when, MultiType.Input):
+                result = tuple(dict.fromkeys(t.io_type for t in when.io_types))
+                if "*" in result and len(result) > 1:
+                    raise ValueError(
+                        "DynamicSlot.Option: AnyType cannot be grouped with concrete types; "
+                        "use a separate Option(when=io.AnyType, ...) instead"
+                    )
+                return result
+            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}"
+                        )
+                    if t.io_type not in types:
+                        types.append(t.io_type)
+                if not types:
+                    raise ValueError("DynamicSlot.Option: when=[] is not allowed; use when=None instead")
+                if "*" in types and len(types) > 1:
+                    raise ValueError(
+                        "DynamicSlot.Option: AnyType cannot be grouped with concrete types; "
+                        "use a separate Option(when=io.AnyType, ...) instead"
+                    )
+                return tuple(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 list(self._when_types),
+                "inputs": create_input_dict_v1(self.inputs),
+            }
+
     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)
-            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
+        def __init__(self, id: str, options: list[DynamicSlot.Option],
+                    display_name: str=None, optional: bool=True, tooltip: str=None, lazy: bool=None, extra_dict=None):
+            if not options:
+                raise ValueError("DynamicSlot.Input: at least one Option is required")
+            for opt in options:
+                if not isinstance(opt, DynamicSlot.Option):
+                    raise ValueError(
+                        f"DynamicSlot.Input: options must be DynamicSlot.Option instances, got {opt!r}"
+                    )
+            super().__init__(id, display_name, optional, tooltip, lazy, extra_dict)
+            self.options = options
+            # Enforce uniqueness: each io_type (and the unconnected case) may
+            # appear in at most one option's ``when``. Also derive the slot's
+            # declared connection type as the ordered union of every non-None
+            # option's ``when`` set so authors control displayed precedence.
+            seen_types: set[str] = set()
+            seen_none = False
+            connected_types: list[str] = []
+            for opt in options:
+                if opt._when_types is None:
+                    if seen_none:
+                        raise ValueError("DynamicSlot.Input: only one Option may declare when=None")
+                    seen_none = True
+                    continue
+                for t in opt._when_types:
+                    if t in seen_types:
+                        raise ValueError(
+                            f"DynamicSlot.Input: type {t!r} appears in more than one Option's `when`; "
+                            "each type must be claimed by exactly one option"
+                        )
+                    seen_types.add(t)
+                    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"
+                )
+            # A required slot demands a link, so the when=None branch is unreachable.
+            if not optional and seen_none:
+                raise ValueError(
+                    "DynamicSlot.Input: optional=False forbids when=None options; "
+                    "the unconnected branch is unreachable when a link is required"
+                )
+            self._slot_io_type = ",".join(connected_types)
+
+        # parse_class_inputs dispatches on the class io_type (COMFY_DYNAMICSLOT_V3),
+        # so get_all/get_io_type must not be overridden; slotType is published via as_dict.
 
         def get_all(self) -> list[Input]:
-            return [self.slot] + self.inputs
+            seen_ids: set[str] = set()
+            children: list[Input] = []
+            for opt in self.options:
+                for inp in opt.inputs:
+                    if inp.id in seen_ids:
+                        continue
+                    seen_ids.add(inp.id)
+                    children.append(inp)
+            return [self] + children
 
         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],
+                # Always render as a connector — slotType may include widget-capable
+                # types (INT/STRING/etc.) but a DynamicSlot is a connection point.
+                "forceInput": True,
             })
 
         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 _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):
+    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 state.
+
+        Connected: pick the first option whose ``when`` set intersects the
+        comma-split resolved type. Unconnected: pick the first ``when=None``.
+        With per-option type uniqueness, at most one connected option can match
+        any single concrete type; ordering only matters when upstream declares
+        a multi-type union (e.g. ``"IMAGE,MASK"``).
+        """
+        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 = {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)
-            # 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 (unmatched concrete + no AnyType option).
+        out_dict[input_type][finalized_id] = value
+        out_dict["dynamic_paths"][finalized_id] = finalize_prefix(curr_prefix, curr_prefix[-1])
+
+
+@dataclass
+class FinalizedOutputs:
+    """Resolved set of active output slots for a single prompt execution.
+
+    Produced by :py:func:`get_finalized_class_outputs`; held alongside V1's
+    ``RETURN_TYPES`` view so the execution engine can size results, validate
+    link indices, and reorder named ``NodeOutput`` results against it.
+    """
+    outputs: list[Output]
+    output_ids: list[str]
+    return_types: list[str]
+    return_names: list[str]
+    output_is_list: list[bool]
+    output_tooltips: list[str | None]
+
+    def __len__(self) -> int:
+        return len(self.outputs)
+
+
+class DynamicOutputs:
+    """Container namespace for dynamic output group declarations.
+
+    Place an instance of one of the inner classes directly inside
+    ``Schema.outputs`` to declare a set of outputs whose shape depends on
+    prompt data. The active branch is chosen at execution time by
+    :py:func:`get_finalized_class_outputs`.
+
+    Forms:
+
+      * :py:class:`DynamicOutputs.ByKey` — selector is a literal input
+        (``Combo`` / ``DynamicCombo`` / ``String`` / etc.) and the active
+        option is chosen by direct value comparison against ``Option.key``.
+      * :py:class:`DynamicOutputs.BySlot` — selector is a
+        :py:class:`DynamicSlot.Input` and the active option is chosen by
+        the slot's resolved upstream type, matching ``Option.when``.
+
+    Inactive options do not produce placeholder slots — downstream links to
+    nonexistent finalized slots are rejected by validation.
+    """
+
+    class Option:
+        """A branch of outputs for :py:class:`DynamicOutputs.ByKey`."""
+
+        def __init__(self, key: str, outputs: list[Output]):
+            if not isinstance(key, str) or not key:
+                raise ValueError("DynamicOutputs.Option: key must be a non-empty string")
+            _validate_outputs("DynamicOutputs.Option", outputs)
+            self.key = key
+            self.outputs = outputs
+
+        def as_dict(self):
+            return {"key": self.key, "outputs": _serialize_outputs(self.outputs)}
+
+    class SlotOption:
+        """A branch of outputs for :py:class:`DynamicOutputs.BySlot`.
+
+        ``when`` accepts the same shapes as :py:class:`DynamicSlot.Option.when`
+        (``None``, ``io.AnyType``, a single ComfyType class, a list of classes,
+        or a ``MultiType.Input``); validation requires it to be a subset of the
+        target ``DynamicSlot.Input``'s declared types.
+        """
+
+        def __init__(self, when: Any, outputs: list[Output]):
+            _validate_outputs("DynamicOutputs.SlotOption", outputs)
+            self.when = when
+            self.outputs = outputs
+            self._when_types = DynamicSlot.Option._normalize_when(when)
+            self._when_set: frozenset[str] | None = (
+                None if self._when_types is None else frozenset(self._when_types)
+            )
+
+        def as_dict(self):
+            return {
+                "when": None if self._when_types is None else list(self._when_types),
+                "outputs": _serialize_outputs(self.outputs),
+            }
+
+    class ByKey:
+        """Active outputs picked by the literal value of a same-node input."""
+
+        kind = "by_key"
+
+        def __init__(self, id: str, selector: str, options: list[DynamicOutputs.Option]):
+            if not isinstance(id, str) or not id:
+                raise ValueError("DynamicOutputs.ByKey: id must be a non-empty string")
+            if not isinstance(selector, str) or not selector:
+                raise ValueError("DynamicOutputs.ByKey: selector must be a non-empty string input id")
+            if not options:
+                raise ValueError("DynamicOutputs.ByKey: at least one Option is required")
+            seen_keys: set[str] = set()
+            seen_ids: set[str] = set()
+            for opt in options:
+                if not isinstance(opt, DynamicOutputs.Option):
+                    raise ValueError(
+                        f"DynamicOutputs.ByKey: options must be DynamicOutputs.Option, got {opt!r}"
+                    )
+                if opt.key in seen_keys:
+                    raise ValueError(f"DynamicOutputs.ByKey: duplicate option key {opt.key!r}")
+                seen_keys.add(opt.key)
+                for o in opt.outputs:
+                    if o.id in seen_ids:
+                        raise ValueError(
+                            f"DynamicOutputs.ByKey: output id {o.id!r} appears in more than one option; "
+                            "each output id must be unique within the group"
+                        )
+                    seen_ids.add(o.id)
+            self.id = id
+            self.selector = selector
+            self.options = options
+
+        def as_dict(self):
+            return {
+                "id": self.id,
+                "kind": self.kind,
+                "selector": self.selector,
+                "options": [opt.as_dict() for opt in self.options],
+            }
+
+        def select(self, prompt_inputs: dict[str, Any]) -> DynamicOutputs.Option | None:
+            """Pick the matching ``Option`` for the prompt's selector value, or ``None``."""
+            value = prompt_inputs.get(self.selector)
+            # DynamicCombo wraps the literal in a nested dict {selector_id: {selector_id: key, ...}};
+            # unwrap so authors don't have to special-case it.
+            if isinstance(value, dict):
+                value = value.get(self.selector)
+            # Links are ``[node_id, slot_idx]`` lists; only literals finalize.
+            if isinstance(value, list):
+                return None
+            for opt in self.options:
+                if opt.key == value:
+                    return opt
+            return None
+
+    class BySlot:
+        """Active outputs picked by the resolved upstream type of a DynamicSlot input."""
+
+        kind = "by_slot"
+
+        def __init__(self, id: str, selector: str, options: list[DynamicOutputs.SlotOption]):
+            if not isinstance(id, str) or not id:
+                raise ValueError("DynamicOutputs.BySlot: id must be a non-empty string")
+            if not isinstance(selector, str) or not selector:
+                raise ValueError("DynamicOutputs.BySlot: selector must be a non-empty string input id")
+            if not options:
+                raise ValueError("DynamicOutputs.BySlot: at least one Option is required")
+            seen_types: set[str] = set()
+            seen_none = False
+            seen_ids: set[str] = set()
+            for opt in options:
+                if not isinstance(opt, DynamicOutputs.SlotOption):
+                    raise ValueError(
+                        f"DynamicOutputs.BySlot: options must be DynamicOutputs.SlotOption, got {opt!r}"
+                    )
+                if opt._when_types is None:
+                    if seen_none:
+                        raise ValueError("DynamicOutputs.BySlot: only one option may declare when=None")
+                    seen_none = True
+                else:
+                    for t in opt._when_types:
+                        if t in seen_types:
+                            raise ValueError(
+                                f"DynamicOutputs.BySlot: type {t!r} appears in more than one option; "
+                                "each type must be claimed by exactly one option"
+                            )
+                        seen_types.add(t)
+                for o in opt.outputs:
+                    if o.id in seen_ids:
+                        raise ValueError(
+                            f"DynamicOutputs.BySlot: output id {o.id!r} appears in more than one option; "
+                            "each output id must be unique within the group"
+                        )
+                    seen_ids.add(o.id)
+            self.id = id
+            self.selector = selector
+            self.options = options
+
+        def as_dict(self):
+            return {
+                "id": self.id,
+                "kind": self.kind,
+                "selector": self.selector,
+                "options": [opt.as_dict() for opt in self.options],
+            }
+
+        def select(self, prompt_inputs: dict[str, Any],
+                   live_input_types: dict[str, str] | None) -> DynamicOutputs.SlotOption | None:
+            """Pick the matching ``SlotOption`` for the slot's resolved type."""
+            raw = prompt_inputs.get(self.selector)
+            has_link = isinstance(raw, list) and raw is not None
+            if not has_link:
+                for opt in self.options:
+                    if opt._when_types is None:
+                        return opt
+                return None
+            resolved = (live_input_types or {}).get(self.selector, "*")
+            resolved_set = {t.strip() for t in resolved.split(",")}
+            for opt in self.options:
+                if opt._when_types is None:
+                    continue
+                if resolved_set & opt._when_set:
+                    return opt
+            return None
+
+
+def _validate_outputs(label: str, outputs: list[Output]) -> None:
+    """Sanity-check a list of outputs at option construction."""
+    if not outputs:
+        return
+    for o in outputs:
+        if not isinstance(o, Output):
+            raise ValueError(f"{label}: outputs must contain Output instances, got {o!r}")
+        if o.id is None:
+            raise ValueError(f"{label}: every output must declare an id")
+
+
+def _serialize_outputs(outputs: list[Output]) -> list[dict]:
+    """Inline-serialize outputs for V1 ``dynamic_outputs`` entries."""
+    return [{"id": o.id, "type": o.get_io_type(), **o.as_dict()} for o in outputs]
+
+
+def _output_metadata(o: Output) -> tuple[str, str, str, bool, str | None]:
+    """Return (id, return_type, display_name, is_output_list, tooltip) for an Output."""
+    rt = o.get_io_type()
+    name = o.display_name if o.display_name else rt
+    return o.id, rt, name, o.is_output_list, (o.tooltip if o.tooltip else None)
+
+
+def get_finalized_class_outputs(
+    schema_outputs: list,
+    prompt_inputs: dict[str, Any] | None,
+    live_input_types: dict[str, str] | None = None,
+) -> FinalizedOutputs:
+    """Resolve the active output list for a node.
+
+    Expands :py:class:`DynamicOutputs.ByKey` against ``prompt_inputs`` and
+    :py:class:`DynamicOutputs.BySlot` against ``live_input_types``. Inactive
+    options contribute no slots — downstream links to ranges that only existed
+    under a different branch are caught by validation rather than silently
+    filled with ``AnyType`` placeholders.
+    """
+    inputs = prompt_inputs or {}
+    outputs: list[Output] = []
+    ids: list[str] = []
+    types: list[str] = []
+    names: list[str] = []
+    is_list: list[bool] = []
+    tooltips: list[str | None] = []
+
+    def append_output(o: Output):
+        oid, rt, name, isl, tt = _output_metadata(o)
+        outputs.append(o)
+        ids.append(oid)
+        types.append(rt)
+        names.append(name)
+        is_list.append(isl)
+        tooltips.append(tt)
+
+    for entry in schema_outputs or []:
+        if isinstance(entry, Output):
+            append_output(entry)
+        elif isinstance(entry, DynamicOutputs.ByKey):
+            selected = entry.select(inputs)
+            if selected is not None:
+                for o in selected.outputs:
+                    append_output(o)
+        elif isinstance(entry, DynamicOutputs.BySlot):
+            selected = entry.select(inputs, live_input_types)
+            if selected is not None:
+                for o in selected.outputs:
+                    append_output(o)
+        # else: ignore unknown entries (future-proofing for new dynamic kinds)
+    return FinalizedOutputs(
+        outputs=outputs,
+        output_ids=ids,
+        return_types=types,
+        return_names=names,
+        output_is_list=is_list,
+        output_tooltips=tooltips,
+    )
+
 
 @comfytype(io_type="IMAGECOMPARE")
 class ImageCompare(ComfyTypeI):
@@ -1357,11 +1788,18 @@ class Range(ComfyTypeIO):
             })
 
 
-DYNAMIC_INPUT_LOOKUP: dict[str, Callable[[dict[str, Any], dict[str, Any], tuple[str, dict[str, Any]], str, list[str] | None], None]] = {}
-def register_dynamic_input_func(io_type: str, func: Callable[[dict[str, Any], dict[str, Any], tuple[str, dict[str, Any]], str, list[str] | None], None]):
+# Signature: (out_dict, live_inputs, value, input_type, curr_prefix, live_input_types).
+# live_input_types is {input_id: resolved_io_type} from TypeResolver; existing
+# expanders ignore it, future type-discriminated ones use it as discriminator.
+_DynamicInputFunc = Callable[
+    [dict[str, Any], dict[str, Any], tuple[str, dict[str, Any]], str, list[str] | None, dict[str, str] | None],
+    None,
+]
+DYNAMIC_INPUT_LOOKUP: dict[str, _DynamicInputFunc] = {}
+def register_dynamic_input_func(io_type: str, func: _DynamicInputFunc):
     DYNAMIC_INPUT_LOOKUP[io_type] = func
 
-def get_dynamic_input_func(io_type: str) -> Callable[[dict[str, Any], dict[str, Any], tuple[str, dict[str, Any]], str, list[str] | None], None]:
+def get_dynamic_input_func(io_type: str) -> _DynamicInputFunc:
     return DYNAMIC_INPUT_LOOKUP[io_type]
 
 def setup_dynamic_input_funcs():
@@ -1465,6 +1903,11 @@ class NodeInfoV1:
     search_aliases: list[str]=None
     essentials_category: str=None
     has_intermediate_output: bool=None
+    dynamic_outputs: list[dict] | None = None
+    """Templates for dynamic output groups (``DynamicOutputs.ByKey`` etc.). The active
+    output list depends on prompt data and is finalized per execution; static
+    ``output`` / ``output_name`` / ``output_is_list`` arrays cover only always-present
+    outputs."""
 
 
 @dataclass
@@ -1601,29 +2044,120 @@ class Schema:
 
     def validate(self):
         '''Validate the schema:
-        - verify ids on inputs and outputs are unique - both internally and in relation to each other
+        - input/output ids are unique within and across each other
+        - DynamicOutputs.ByKey / BySlot have unique group ids
+        - ByKey selectors reference a real input (literal alignment validated when
+          the selector is a DynamicCombo)
+        - BySlot selectors reference a real DynamicSlot input and every option's
+          ``when`` is accepted by that slot
         '''
         nested_inputs: list[Input] = []
         for input in self.inputs:
             if not isinstance(input, DynamicInput):
                 nested_inputs.extend(input.get_all())
         input_ids = [i.id for i in nested_inputs]
-        output_ids = [o.id for o in self.outputs]
         input_set = set(input_ids)
+        # Walk DynamicCombo / DynamicSlot options so a DynamicOutputs selector can
+        # target a nested dynamic input by dotted path (e.g. "outer.inner_slot").
+        # Other DynamicInput kinds (e.g. Autogrow) are skipped — their nested ids
+        # only exist at prompt time and aren't statically addressable.
+        selector_targets: dict[str, Input] = {}
+        def _walk(inputs: list[Input], prefix: str | None) -> None:
+            for inp in inputs:
+                dotted = inp.id if prefix is None else f"{prefix}.{inp.id}"
+                selector_targets[dotted] = inp
+                if isinstance(inp, (DynamicCombo.Input, DynamicSlot.Input)):
+                    for opt in inp.options:
+                        _walk(opt.inputs, dotted)
+        _walk(self.inputs, None)
+        combo_inputs = {sid: inp for sid, inp in selector_targets.items() if isinstance(inp, DynamicCombo.Input)}
+        slot_inputs = {sid: inp for sid, inp in selector_targets.items() if isinstance(inp, DynamicSlot.Input)}
+        all_selectable_ids = input_set | set(selector_targets)
+
+        # ``output_ids`` covers every id that may ever appear in a finalized
+        # output list — static outputs + every option's outputs across every
+        # dynamic group — so collisions between branches are caught up front.
+        output_ids: list[str] = []
+        for o in self.outputs:
+            if isinstance(o, Output):
+                output_ids.append(o.id)
+            elif isinstance(o, (DynamicOutputs.ByKey, DynamicOutputs.BySlot)):
+                for opt in o.options:
+                    output_ids.extend(child.id for child in opt.outputs)
+
         output_set = set(output_ids)
         issues: list[str] = []
-        # verify ids are unique per list
         if len(input_set) != len(input_ids):
             issues.append(f"Input ids must be unique, but {[item for item, count in Counter(input_ids).items() if count > 1]} are not.")
         if len(output_set) != len(output_ids):
             issues.append(f"Output ids must be unique, but {[item for item, count in Counter(output_ids).items() if count > 1]} are not.")
+
+        group_ids: list[str] = []
+        for o in self.outputs:
+            if isinstance(o, DynamicOutputs.ByKey):
+                group_ids.append(o.id)
+                if o.selector not in all_selectable_ids:
+                    issues.append(
+                        f"DynamicOutputs.ByKey(id={o.id!r}) selector input {o.selector!r} "
+                        f"does not exist on the schema."
+                    )
+                # If the selector is a DynamicCombo, every option key must match one of its keys.
+                if o.selector in combo_inputs:
+                    valid_keys = {opt.key for opt in combo_inputs[o.selector].options}
+                    stray = [opt.key for opt in o.options if opt.key not in valid_keys]
+                    if stray:
+                        issues.append(
+                            f"DynamicOutputs.ByKey(id={o.id!r}) option key(s) {stray!r} are not "
+                            f"declared on DynamicCombo input {o.selector!r} (valid: {sorted(valid_keys)!r})."
+                        )
+            elif isinstance(o, DynamicOutputs.BySlot):
+                group_ids.append(o.id)
+                if o.selector not in slot_inputs:
+                    issues.append(
+                        f"DynamicOutputs.BySlot(id={o.id!r}) selector {o.selector!r} must reference "
+                        f"a DynamicSlot input on the schema."
+                    )
+                else:
+                    slot = slot_inputs[o.selector]
+                    slot_accepted = set()
+                    slot_has_none = False
+                    for sopt in slot.options:
+                        if sopt._when_types is None:
+                            slot_has_none = True
+                        else:
+                            slot_accepted.update(sopt._when_types)
+                    for opt in o.options:
+                        if opt._when_types is None:
+                            if not slot_has_none:
+                                issues.append(
+                                    f"DynamicOutputs.BySlot(id={o.id!r}) option(when=None) requires "
+                                    f"DynamicSlot {o.selector!r} to declare a when=None option."
+                                )
+                        else:
+                            stray_types = [t for t in opt._when_types if t not in slot_accepted]
+                            if stray_types:
+                                issues.append(
+                                    f"DynamicOutputs.BySlot(id={o.id!r}) option type(s) {stray_types!r} "
+                                    f"are not accepted by DynamicSlot {o.selector!r} "
+                                    f"(accepted: {sorted(slot_accepted)!r})."
+                                )
+        if len(set(group_ids)) != len(group_ids):
+            issues.append(
+                f"DynamicOutputs group ids must be unique, but "
+                f"{[i for i, c in Counter(group_ids).items() if c > 1]} are not."
+            )
         if len(issues) > 0:
             raise ValueError("\n".join(issues))
-        # validate inputs and outputs
+        # per-element validation
         for input in self.inputs:
             input.validate()
         for output in self.outputs:
-            output.validate()
+            if isinstance(output, Output):
+                output.validate()
+            elif isinstance(output, (DynamicOutputs.ByKey, DynamicOutputs.BySlot)):
+                for opt in output.options:
+                    for child in opt.outputs:
+                        child.validate()
         if self.price_badge is not None:
             self.price_badge.validate()
 
@@ -1648,9 +2182,10 @@ class Schema:
                 self.hidden.append(Hidden.prompt)
             if Hidden.extra_pnginfo not in self.hidden:
                 self.hidden.append(Hidden.extra_pnginfo)
-        # give outputs without ids default ids
+        # give outputs without ids default ids (dynamic groups require explicit ids
+        # so we can never accidentally collide synthesized names across branches).
         for i, output in enumerate(self.outputs):
-            if output.id is None:
+            if isinstance(output, Output) and output.id is None:
                 output.id = f"_{i}_{output.io_type}_"
 
     def get_v1_info(self, cls) -> NodeInfoV1:
@@ -1659,15 +2194,21 @@ class Schema:
         if self.hidden:
             for hidden in self.hidden:
                 input.setdefault("hidden", {})[hidden.name] = (hidden.value,)
-        # create separate lists from output fields
+        # create separate lists from output fields (static outputs only — dynamic
+        # groups are advertised separately via ``dynamic_outputs`` so the frontend
+        # and any V1 consumer see a stable always-present prefix).
         output = []
         output_is_list = []
         output_name = []
         output_tooltips = []
         output_matchtypes = []
         any_matchtypes = False
+        dynamic_outputs: list[dict[str, Any]] = []
         if self.outputs:
             for o in self.outputs:
+                if isinstance(o, (DynamicOutputs.ByKey, DynamicOutputs.BySlot)):
+                    dynamic_outputs.append(o.as_dict())
+                    continue
                 output.append(o.io_type)
                 output_is_list.append(o.is_output_list)
                 output_name.append(o.display_name if o.display_name else o.io_type)
@@ -1706,10 +2247,16 @@ class Schema:
             price_badge=self.price_badge.as_dict(self.inputs) if self.price_badge is not None else None,
             search_aliases=self.search_aliases if self.search_aliases else None,
             essentials_category=self.essentials_category,
+            dynamic_outputs=dynamic_outputs or None,
         )
         return info
 
-def get_finalized_class_inputs(d: dict[str, Any], live_inputs: dict[str, Any], include_hidden=False) -> tuple[dict[str, Any], V3Data]:
+def get_finalized_class_inputs(d: dict[str, Any], live_inputs: dict[str, Any], include_hidden=False, live_input_types: dict[str, str] | None = None) -> tuple[dict[str, Any], V3Data]:
+    """Expand a node's V3 schema against a concrete prompt.
+
+    ``live_input_types`` is an optional ``{input_id: resolved_io_type}`` map
+    (from ``TypeResolver``) used by future type-discriminated dynamic inputs.
+    """
     out_dict = {
         "required": {},
         "optional": {},
@@ -1719,7 +2266,7 @@ def get_finalized_class_inputs(d: dict[str, Any], live_inputs: dict[str, Any], i
     d = d.copy()
     # ignore hidden for parsing
     hidden = d.pop("hidden", None)
-    parse_class_inputs(out_dict, live_inputs, d)
+    parse_class_inputs(out_dict, live_inputs, d, None, live_input_types)
     if hidden is not None and include_hidden:
         out_dict["hidden"] = hidden
     v3_data = {}
@@ -1732,7 +2279,7 @@ def get_finalized_class_inputs(d: dict[str, Any], live_inputs: dict[str, Any], i
         v3_data["dynamic_paths_default_value"] = dynamic_paths_default_value
     return out_dict, hidden, v3_data
 
-def parse_class_inputs(out_dict: dict[str, Any], live_inputs: dict[str, Any], curr_dict: dict[str, Any], curr_prefix: list[str] | None=None) -> None:
+def parse_class_inputs(out_dict: dict[str, Any], live_inputs: dict[str, Any], curr_dict: dict[str, Any], curr_prefix: list[str] | None=None, live_input_types: dict[str, str] | None = None) -> None:
     for input_type, inner_d in curr_dict.items():
         for id, value in inner_d.items():
             io_type = value[0]
@@ -1740,7 +2287,7 @@ def parse_class_inputs(out_dict: dict[str, Any], live_inputs: dict[str, Any], cu
                 # dynamic inputs need to be handled with lookup functions
                 dynamic_input_func = get_dynamic_input_func(io_type)
                 new_prefix = handle_prefix(curr_prefix, id)
-                dynamic_input_func(out_dict, live_inputs, value, input_type, new_prefix)
+                dynamic_input_func(out_dict, live_inputs, value, input_type, new_prefix, live_input_types)
             else:
                 # non-dynamic inputs get directly transferred
                 finalized_id = finalize_prefix(curr_prefix, id)
@@ -2099,12 +2646,17 @@ class _ComfyNodeBaseInternal(_ComfyNodeInternal):
             cls._ACCEPT_ALL_INPUTS = schema.accept_all_inputs
 
         if cls._RETURN_TYPES is None:
+            # Class-level RETURN_TYPES / RETURN_NAMES / OUTPUT_IS_LIST cover the
+            # always-present static outputs only; dynamic groups are finalized
+            # per-prompt in get_finalized_class_outputs.
             output = []
             output_name = []
             output_is_list = []
             output_tooltips = []
             if schema.outputs:
                 for o in schema.outputs:
+                    if not isinstance(o, Output):
+                        continue
                     output.append(o.io_type)
                     output_name.append(o.display_name if o.display_name else o.io_type)
                     output_is_list.append(o.is_output_list)
@@ -2171,17 +2723,32 @@ class ComfyNode(_ComfyNodeBaseInternal):
 class NodeOutput(_NodeOutputInternal):
     '''
     Standardized output of a node; can pass in any number of args and/or a UIOutput into 'ui' kwarg.
+
+    For nodes whose active output list depends on prompt data (e.g. those using
+    :py:class:`DynamicOutputs`), pass ``named={output_id: value, ...}`` instead
+    of positional args. The execution engine reorders against the finalized
+    output list at run time; unknown or missing ids raise.
     '''
-    def __init__(self, *args: Any, ui: _UIOutput | dict=None, expand: dict=None, block_execution: str=None):
+    def __init__(self, *args: Any, named: dict[str, Any]=None, ui: _UIOutput | dict=None, expand: dict=None, block_execution: str=None):
+        if args and named is not None:
+            raise ValueError("NodeOutput: cannot mix positional args with named=...; choose one form")
         self.args = args
+        self.named = named
         self.ui = ui
         self.expand = expand
         self.block_execution = block_execution
 
     @property
     def result(self):
+        # Positional tuple only; named results live in ``self.named`` and are
+        # ordered against the finalized output list by the execution engine.
         return self.args if len(self.args) > 0 else None
 
+    @classmethod
+    def from_named(cls, named: dict[str, Any], *, ui: _UIOutput | dict=None, expand: dict=None, block_execution: str=None) -> NodeOutput:
+        """Build a NodeOutput keyed by output id, for dynamic-output nodes."""
+        return cls(named=named, ui=ui, expand=expand, block_execution=block_execution)
+
     @classmethod
     def from_dict(cls, data: dict[str, Any]) -> NodeOutput:
         args = ()

comfy_execution/graph.py

@@ -26,6 +26,7 @@ class DynamicPrompt:
         self.ephemeral_prompt = {}
         self.ephemeral_parents = {}
         self.ephemeral_display = {}
+        self._type_resolver = None  # lazy; invalidated by add_ephemeral_node
 
     def get_node(self, node_id):
         if node_id in self.ephemeral_prompt:
@@ -41,6 +42,17 @@ class DynamicPrompt:
         self.ephemeral_prompt[node_id] = node_info
         self.ephemeral_parents[node_id] = parent_id
         self.ephemeral_display[node_id] = display_id
+        # Selective downstream invalidation would need topological info; the
+        # cache is small, so just wipe it.
+        if self._type_resolver is not None:
+            self._type_resolver.invalidate()
+
+    def get_type_resolver(self):
+        """Lazily build and return the per-prompt TypeResolver."""
+        if self._type_resolver is None:
+            from comfy_execution.type_resolver import TypeResolver
+            self._type_resolver = TypeResolver(self)
+        return self._type_resolver
 
     def get_real_node_id(self, node_id):
         while node_id in self.ephemeral_parents:

comfy_execution/type_resolver.py

@@ -0,0 +1,436 @@
+"""Server-side type resolver for prompt graphs.
+
+Resolves the concrete io_type of any output/input slot by walking the prompt
+graph. Handles V1/V3 ``RETURN_TYPES``, V3 ``MatchType`` template chains, and
+falls back to ``AnyType`` (with a one-shot warning) on cycles, depth overflow,
+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
+
+import logging
+from typing import Any
+
+from comfy_api.latest import _io as io
+from comfy_api.internal import _ComfyNodeInternal
+
+
+def _parse_link(val: Any) -> tuple[str, int] | None:
+    """Return ``(src_node_id, src_slot_idx)`` if ``val`` is a well-formed link, else ``None``.
+
+    A link is ``[node_id: str, slot_idx: int]``. Malformed shapes return ``None``
+    so callers can fall back to AnyType rather than raise.
+    """
+    if not isinstance(val, (list, tuple)) or len(val) != 2:
+        return None
+    src_node, src_slot = val[0], val[1]
+    if not isinstance(src_node, str):
+        return None
+    # bool is a subclass of int — reject so True/False aren't read as slot 1/0.
+    if isinstance(src_slot, bool) or not isinstance(src_slot, int):
+        return None
+    return src_node, src_slot
+
+ANY_TYPE: str = io.AnyType.io_type
+MAX_RESOLVE_DEPTH: int = 64  # belt-and-suspenders cap; real MatchType chains stay tiny
+
+
+class TypeResolver:
+    """Resolves concrete io_types for a prompt graph.
+
+    Instantiate once per prompt (or per ``DynamicPrompt``) and reuse; results
+    are cached. Call :py:meth:`invalidate` (or :py:meth:`invalidate_node`) when
+    the underlying graph mutates (e.g. when an ephemeral node is added).
+    """
+
+    def __init__(self, prompt_source: Any):
+        """Args:
+            prompt_source: Either a ``DynamicPrompt`` (anything with
+                ``get_node(node_id)`` / ``has_node(node_id)``) or a plain
+                ``dict[node_id, {"class_type", "inputs"}]``.
+        """
+        self._source = prompt_source
+        self._output_cache: dict[tuple[str, int], str] = {}
+        self._is_output_list_cache: dict[tuple[str, int], bool] = {}
+        self._warned: set[tuple[str, Any, str]] = set()
+
+    # ---- prompt access ----------------------------------------------------
+    def _has_node(self, node_id: str) -> bool:
+        if hasattr(self._source, "has_node"):
+            return self._source.has_node(node_id)
+        return node_id in self._source
+
+    def _get_node(self, node_id: str) -> dict[str, Any] | None:
+        try:
+            if hasattr(self._source, "get_node"):
+                return self._source.get_node(node_id)
+            return self._source[node_id]
+        except Exception:
+            return None
+
+    @staticmethod
+    def _get_class_def(class_type: str):
+        # Local import: nodes <-> comfy_execution would cycle at import time.
+        import nodes
+        return nodes.NODE_CLASS_MAPPINGS.get(class_type)
+
+    def _get_class_def_for_node(self, node_id: str):
+        """Return (node_dict, class_def) for ``node_id``, or ``(None, None)``."""
+        if not self._has_node(node_id):
+            return None, None
+        node = self._get_node(node_id)
+        if node is None:
+            return None, None
+        class_type = node.get("class_type")
+        if not isinstance(class_type, str):
+            return node, None
+        return node, self._get_class_def(class_type)
+
+    # ---- cache management -------------------------------------------------
+    def invalidate(self) -> None:
+        """Clear all cached resolutions. Cheap; call after any graph mutation."""
+        self._output_cache.clear()
+        self._is_output_list_cache.clear()
+        # Keep self._warned: re-emitting already-logged warnings would just spam.
+
+    def invalidate_node(self, node_id: str) -> None:
+        """Clear cached entries for a single node (e.g. after node-level expand)."""
+        for key in [k for k in self._output_cache if k[0] == node_id]:
+            del self._output_cache[key]
+        for key in [k for k in self._is_output_list_cache if k[0] == node_id]:
+            del self._is_output_list_cache[key]
+
+    # ---- output resolution -----------------------------------------------
+    def resolve_output_type(self, node_id: str, slot_idx: int,
+                            _stack: frozenset[tuple[str, int]] | None = None) -> str:
+        """Return the resolved io_type string of ``node_id``'s output slot.
+
+        Falls back to ``ANY_TYPE`` on cycle, depth-overflow, unknown class,
+        out-of-range slot, missing node, malformed link, or unresolved
+        MatchType template.
+        """
+        # Degrade gracefully on non-int slot_idx (e.g. malformed API JSON).
+        if isinstance(slot_idx, bool) or not isinstance(slot_idx, int):
+            return ANY_TYPE
+
+        cache_key = (node_id, slot_idx)
+        if cache_key in self._output_cache:
+            return self._output_cache[cache_key]
+
+        if _stack is None:
+            _stack = frozenset()
+        if cache_key in _stack:
+            self._warn(node_id, slot_idx, "cycle detected during type resolution; defaulting to AnyType")
+            return ANY_TYPE
+        if len(_stack) >= MAX_RESOLVE_DEPTH:
+            self._warn(node_id, slot_idx, f"exceeded MAX_RESOLVE_DEPTH={MAX_RESOLVE_DEPTH}; defaulting to AnyType")
+            return ANY_TYPE
+        next_stack = _stack | {cache_key}
+
+        node, class_def = self._get_class_def_for_node(node_id)
+        if class_def is None:
+            return ANY_TYPE
+        class_type = node.get("class_type")
+
+        # V3 schemas may declare DynamicOutputs groups whose active slots are
+        # determined by prompt inputs and do not appear in class RETURN_TYPES;
+        # resolve types against the finalized output list when present.
+        finalized = self._get_finalized_outputs(node_id, node, class_def)
+        if finalized is not None:
+            if slot_idx < 0 or slot_idx >= len(finalized):
+                return ANY_TYPE
+            declared = finalized.return_types[slot_idx]
+            resolved_output = finalized.outputs[slot_idx]
+            resolved = declared
+            if isinstance(resolved_output, io.MatchType.Output):
+                schema = getattr(class_def, "SCHEMA", None) or class_def.GET_SCHEMA()
+                resolved = self._resolve_match_template(
+                    node_id, schema, resolved_output.template.template_id, next_stack
+                )
+        else:
+            try:
+                return_types = class_def.RETURN_TYPES
+            except Exception:
+                return ANY_TYPE
+            if return_types is None or slot_idx < 0 or slot_idx >= len(return_types):
+                return ANY_TYPE
+
+            declared = return_types[slot_idx]
+
+            # Only V3 schemas carry MatchType template info; V1 RETURN_TYPES are
+            # always concrete strings.
+            resolved = declared
+            if isinstance(class_def, type) and issubclass(class_def, _ComfyNodeInternal):
+                schema = getattr(class_def, "SCHEMA", None)
+                if schema is None:
+                    # RETURN_TYPES access above usually populates SCHEMA — be defensive.
+                    try:
+                        schema = class_def.GET_SCHEMA()
+                    except Exception:
+                        schema = None
+                if schema is not None and slot_idx < len(schema.outputs):
+                    out = schema.outputs[slot_idx]
+                    if isinstance(out, io.MatchType.Output):
+                        resolved = self._resolve_match_template(
+                            node_id, schema, out.template.template_id, next_stack
+                        )
+
+        # Warn only for V1 wildcards declared as "*"; unresolved MatchType
+        # templates warn separately in _resolve_match_template, avoiding double-warns.
+        if isinstance(resolved, str) and resolved == ANY_TYPE and declared == ANY_TYPE:
+            self._warn(
+                node_id, slot_idx,
+                f"node '{class_type}' output slot {slot_idx} is wildcard; defaulting to AnyType",
+            )
+
+        if not isinstance(resolved, str):
+            # e.g. legacy combo declared as a list of options.
+            self._warn(node_id, slot_idx,
+                       f"node '{class_type}' output slot {slot_idx} has non-string return type {type(resolved).__name__}; defaulting to AnyType")
+            resolved = ANY_TYPE
+
+        self._output_cache[cache_key] = resolved
+        return resolved
+
+    def _resolve_match_template(self, node_id: str, schema, template_id: str,
+                                stack: frozenset[tuple[str, int]]) -> str:
+        """Walk MatchType.Inputs sharing ``template_id``; return first concrete resolution or ``ANY_TYPE``."""
+        node = self._get_node(node_id)
+        inputs_dict = (node or {}).get("inputs", {}) or {}
+        any_input_seen = False
+        for inp in schema.inputs:
+            if not isinstance(inp, io.MatchType.Input):
+                continue
+            if inp.template.template_id != template_id:
+                continue
+            any_input_seen = True
+            val = inputs_dict.get(inp.id)
+            if val is None:
+                continue
+            link = _parse_link(val)
+            if link is not None:
+                t = self.resolve_output_type(link[0], link[1], stack)
+                if t != ANY_TYPE:
+                    return t
+            # Literal or malformed link: MatchType slots have no declared concrete type.
+        if not any_input_seen:
+            # Node-author bug: output template has no matching Input.
+            self._warn(node_id, None,
+                       f"MatchType output template '{template_id}' has no matching Input on the node; defaulting to AnyType")
+        else:
+            self._warn(node_id, None,
+                       f"MatchType template '{template_id}' has no bound concrete upstream input; defaulting to AnyType")
+        return ANY_TYPE
+
+    def _get_finalized_outputs(self, node_id: str, node: dict | None, class_def) -> io.FinalizedOutputs | None:
+        """Return ``FinalizedOutputs`` for V3 nodes with DynamicOutputs groups, else ``None``.
+
+        ``BySlot`` groups need ``live_input_types`` (computed lazily from the
+        resolver itself) so the active option can be picked by resolved type.
+        """
+        if not (isinstance(class_def, type) and issubclass(class_def, _ComfyNodeInternal)):
+            return None
+        try:
+            schema = class_def.GET_SCHEMA()
+        except Exception:
+            return None
+        has_dynamic = any(
+            isinstance(o, (io.DynamicOutputs.ByKey, io.DynamicOutputs.BySlot))
+            for o in schema.outputs
+        )
+        if not has_dynamic:
+            return None
+        prompt_inputs = (node or {}).get("inputs", {}) or {}
+        # live_input_types is only needed for BySlot — skip the resolver pass
+        # otherwise to keep the hot path cheap.
+        needs_live_types = any(isinstance(o, io.DynamicOutputs.BySlot) for o in schema.outputs)
+        live_input_types = self.compute_live_input_types(node_id) if needs_live_types else None
+        return io.get_finalized_class_outputs(
+            schema.outputs, prompt_inputs, live_input_types=live_input_types,
+        )
+
+    def finalized_output_count(self, node_id: str) -> int:
+        """Number of active output slots on ``node_id``'s schema for the current prompt.
+
+        For V3 nodes with :py:class:`comfy_api.latest._io.DynamicOutputs` groups
+        the count is computed against the node's prompt inputs; for static V3
+        / V1 nodes it falls back to ``len(RETURN_TYPES)``. Unknown nodes
+        report ``0``.
+        """
+        node, class_def = self._get_class_def_for_node(node_id)
+        if class_def is None:
+            return 0
+        finalized = self._get_finalized_outputs(node_id, node, class_def)
+        if finalized is not None:
+            return len(finalized)
+        try:
+            return len(class_def.RETURN_TYPES)
+        except Exception:
+            return 0
+
+    def is_output_list(self, node_id: str, slot_idx: int) -> bool:
+        """Whether the source slot is declared as a list output (``OUTPUT_IS_LIST[idx]``)."""
+        if isinstance(slot_idx, bool) or not isinstance(slot_idx, int):
+            return False
+        cache_key = (node_id, slot_idx)
+        if cache_key in self._is_output_list_cache:
+            return self._is_output_list_cache[cache_key]
+        result = False
+        node, class_def = self._get_class_def_for_node(node_id)
+        if class_def is not None:
+            finalized = self._get_finalized_outputs(node_id, node, class_def)
+            if finalized is not None:
+                if 0 <= slot_idx < len(finalized):
+                    result = bool(finalized.output_is_list[slot_idx])
+            else:
+                lst = getattr(class_def, "OUTPUT_IS_LIST", None)
+                if lst is not None and 0 <= slot_idx < len(lst):
+                    result = bool(lst[slot_idx])
+        self._is_output_list_cache[cache_key] = result
+        return result
+
+    # ---- input resolution ------------------------------------------------
+    def resolve_input_type(self, node_id: str, input_id: str) -> str:
+        """Resolve the io_type of the value currently bound to a node's input.
+
+        * If the value is a link, return the resolved type of the source slot.
+        * If the value is a literal, return the declared slot's effective
+          io_type (peeling dynamic-input wrappers — e.g. an Autogrow-of-Image
+          slot resolves to ``IMAGE``, not ``COMFY_AUTOGROW_V3``).
+        * If the value is missing, malformed, or the slot is unknown, return
+          ``ANY_TYPE``.
+        """
+        node = self._get_node(node_id)
+        if node is None:
+            return ANY_TYPE
+        inputs = node.get("inputs", {}) or {}
+        if input_id not in inputs:
+            return ANY_TYPE
+        link = _parse_link(inputs[input_id])
+        if link is not None:
+            return self.resolve_output_type(link[0], link[1])
+        return self.get_declared_slot_io_type(node_id, input_id)
+
+    def is_input_list(self, node_id: str, input_id: str) -> bool:
+        """Whether the value bound to ``input_id`` originates from a list output."""
+        node = self._get_node(node_id)
+        if node is None:
+            return False
+        link = _parse_link((node.get("inputs", {}) or {}).get(input_id))
+        if link is None:
+            return False
+        return self.is_output_list(link[0], link[1])
+
+    def get_declared_slot_io_type(self, node_id: str, input_id: str) -> str:
+        """Return the effective declared io_type of a node's input slot.
+
+        Peels dynamic-input wrappers so that the user-facing element type is
+        returned:
+
+        * Autogrow → wrapped template input's io_type
+        * DynamicSlot → underlying slot's io_type
+        * Anything else → the slot's own io_type
+        * DynamicCombo / unsupported → ``ANY_TYPE`` (the combo key is itself
+          dynamic, not a meaningful type for consumers)
+        """
+        _, class_def = self._get_class_def_for_node(node_id)
+        if class_def is None:
+            return ANY_TYPE
+
+        # Prefer V3 schema (carries dynamic-input wrapper info).
+        if isinstance(class_def, type) and issubclass(class_def, _ComfyNodeInternal):
+            schema = getattr(class_def, "SCHEMA", None)
+            if schema is None:
+                try:
+                    class_def.GET_SCHEMA()
+                    schema = getattr(class_def, "SCHEMA", None)
+                except Exception:
+                    schema = None
+            if schema is not None:
+                # Top-level input id.
+                for inp in schema.inputs:
+                    if inp.id == input_id:
+                        return self._effective_io_type(inp)
+                # Nested (DynamicSlot / DynamicCombo `parent.child`).
+                if "." in input_id:
+                    top, _, _ = input_id.partition(".")
+                    for inp in schema.inputs:
+                        if inp.id != top:
+                            continue
+                        for child in inp.get_all():
+                            if child is inp:
+                                continue
+                            if child.id == input_id.split(".", 1)[1]:
+                                return self._effective_io_type(child)
+                # Fall through to V1 dict (hidden inputs, etc.).
+
+        try:
+            inputs = class_def.INPUT_TYPES()
+        except Exception:
+            return ANY_TYPE
+        for section in ("required", "optional"):
+            section_d = inputs.get(section, {})
+            if input_id in section_d:
+                entry = section_d[input_id]
+                if not entry:
+                    return ANY_TYPE
+                t = entry[0]
+                if isinstance(t, str):
+                    return t
+                if isinstance(t, list):  # legacy combo declared as list of options
+                    return io.Combo.io_type
+                return ANY_TYPE
+        return ANY_TYPE
+
+    @staticmethod
+    def _effective_io_type(inp) -> str:
+        """Return the consumer-facing io_type of a (possibly dynamic) input."""
+        # Autogrow / DynamicSlot wrap a real element type; that's what consumers care about.
+        if isinstance(inp, io.Autogrow.Input):
+            try:
+                return inp.template.input.get_io_type()
+            except Exception:
+                return ANY_TYPE
+        if isinstance(inp, io.DynamicSlot.Input):
+            # 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
+        try:
+            return inp.get_io_type()
+        except Exception:
+            return ANY_TYPE
+
+    # ---- bulk helpers ----------------------------------------------------
+    def compute_live_input_types(self, node_id: str) -> dict[str, str]:
+        """Build the ``{input_id: resolved_io_type}`` map for a node.
+
+        Consumed by ``_io.get_finalized_class_inputs`` so future per-type
+        dynamic-input expansion can branch on what was actually connected.
+        """
+        node = self._get_node(node_id)
+        if node is None:
+            return {}
+        out: dict[str, str] = {}
+        for input_id in (node.get("inputs", {}) or {}).keys():
+            out[input_id] = self.resolve_input_type(node_id, input_id)
+        return out
+
+    # ---- diagnostics -----------------------------------------------------
+    def _warn(self, node_id: str, slot_idx: Any, msg: str) -> None:
+        key = (node_id, slot_idx, msg)
+        if key in self._warned:
+            return
+        self._warned.add(key)
+        logging.warning("TypeResolver: node=%s slot=%s %s", node_id, slot_idx, msg)

execution.py

@@ -83,8 +83,8 @@ class IsChangedCache:
             self.is_changed[node_id] = node["is_changed"]
             return self.is_changed[node_id]
 
-        # Intentionally do not use cached outputs here. We only want constants in IS_CHANGED
-        input_data_all, _, v3_data = get_input_data(node["inputs"], class_def, node_id, None)
+        # Intentionally do not use cached outputs here. We only want constants in IS_CHANGED.
+        input_data_all, _, v3_data = get_input_data(node["inputs"], class_def, node_id, None, self.dynprompt)
         try:
             is_changed = await _async_map_node_over_list(self.prompt_id, node_id, class_def, input_data_all, is_changed_name, v3_data=v3_data)
             is_changed = await resolve_map_node_over_list_results(is_changed)
@@ -152,13 +152,16 @@ class CacheSet:
 
 SENSITIVE_EXTRA_DATA_KEYS = ("auth_token_comfy_org", "api_key_comfy_org")
 
-def get_input_data(inputs, class_def, unique_id, execution_list=None, dynprompt=None, extra_data={}):
+def get_input_data(inputs, class_def, unique_id, execution_list=None, dynprompt=None, extra_data={}, live_input_types=None):
     is_v3 = issubclass(class_def, _ComfyNodeInternal)
     v3_data: io.V3Data = {}
     hidden_inputs_v3 = {}
     valid_inputs = class_def.INPUT_TYPES()
     if is_v3:
-        valid_inputs, hidden, v3_data = _io.get_finalized_class_inputs(valid_inputs, inputs)
+        # Let dynamic schemas branch on resolved upstream types, not just literal values.
+        if live_input_types is None and dynprompt is not None and hasattr(dynprompt, "get_type_resolver"):
+            live_input_types = dynprompt.get_type_resolver().compute_live_input_types(unique_id)
+        valid_inputs, hidden, v3_data = _io.get_finalized_class_inputs(valid_inputs, inputs, live_input_types=live_input_types)
     input_data_all = {}
     missing_keys = {}
     for x in inputs:
@@ -311,15 +314,56 @@ async def _async_map_node_over_list(prompt_id, unique_id, obj, input_data_all, f
     return results
 
 
-def merge_result_data(results, obj):
+def _expected_output_count(obj, finalized_outputs=None):
+    """Size results / blocker tuples by the finalized output list when present."""
+    if finalized_outputs is not None:
+        return len(finalized_outputs)
+    return len(getattr(obj, "RETURN_TYPES", ()))
+
+
+def _normalize_named_result(node_output, finalized_outputs):
+    """Convert a ``NodeOutput.from_named({...})`` payload into an ordered tuple.
+
+    Strict by design: an active output id missing from the payload is an
+    error; an unknown id is an error. Use the finalized output list as the
+    single source of truth for ordering.
+    """
+    if finalized_outputs is None:
+        raise Exception(
+            "NodeOutput(named=...) is only supported for V3 nodes with a finalized "
+            "output schema (e.g. nodes using DynamicOutputs)."
+        )
+    expected_ids = finalized_outputs.output_ids
+    payload = node_output.named
+    missing = [oid for oid in expected_ids if oid not in payload]
+    unknown = [oid for oid in payload if oid not in expected_ids]
+    if missing or unknown:
+        raise Exception(
+            f"NodeOutput(named=...) ids do not match active outputs: "
+            f"missing={missing}, unknown={unknown}, expected={expected_ids}"
+        )
+    return tuple(payload[oid] for oid in expected_ids)
+
+
+def merge_result_data(results, obj, finalized_outputs=None):
     # check which outputs need concatenating
     output = []
-    output_is_list = [False] * len(results[0])
-    if hasattr(obj, "OUTPUT_IS_LIST"):
-        output_is_list = obj.OUTPUT_IS_LIST
+    expected_count = _expected_output_count(obj, finalized_outputs)
+    if finalized_outputs is not None:
+        output_is_list = finalized_outputs.output_is_list
+    else:
+        output_is_list = [False] * len(results[0])
+        if hasattr(obj, "OUTPUT_IS_LIST"):
+            output_is_list = obj.OUTPUT_IS_LIST
+
+    for r in results:
+        if len(r) != expected_count:
+            raise Exception(
+                f"Node returned {len(r)} outputs but active schema has {expected_count}"
+            )
 
     # merge node execution results
-    for i, is_list in zip(range(len(results[0])), output_is_list):
+    for i, is_list in zip(range(expected_count), output_is_list):
         if is_list:
             value = []
             for o in results:
@@ -332,19 +376,20 @@ def merge_result_data(results, obj):
             output.append([o[i] for o in results])
     return output
 
-async def get_output_data(prompt_id, unique_id, obj, input_data_all, execution_block_cb=None, pre_execute_cb=None, v3_data=None):
+async def get_output_data(prompt_id, unique_id, obj, input_data_all, execution_block_cb=None, pre_execute_cb=None, v3_data=None, finalized_outputs=None):
     return_values = await _async_map_node_over_list(prompt_id, unique_id, obj, input_data_all, obj.FUNCTION, allow_interrupt=True, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb, v3_data=v3_data)
     has_pending_task = any(isinstance(r, asyncio.Task) and not r.done() for r in return_values)
     if has_pending_task:
         return return_values, {}, False, has_pending_task
-    output, ui, has_subgraph = get_output_from_returns(return_values, obj)
+    output, ui, has_subgraph = get_output_from_returns(return_values, obj, finalized_outputs=finalized_outputs)
     return output, ui, has_subgraph, False
 
-def get_output_from_returns(return_values, obj):
+def get_output_from_returns(return_values, obj, finalized_outputs=None):
     results = []
     uis = []
     subgraph_results = []
     has_subgraph = False
+    expected_count = _expected_output_count(obj, finalized_outputs)
     for i in range(len(return_values)):
         r = return_values[i]
         if isinstance(r, dict):
@@ -356,12 +401,12 @@ def get_output_from_returns(return_values, obj):
                 new_graph = r['expand']
                 result = r.get("result", None)
                 if isinstance(result, ExecutionBlocker):
-                    result = tuple([result] * len(obj.RETURN_TYPES))
+                    result = tuple([result] * expected_count)
                 subgraph_results.append((new_graph, result))
             elif 'result' in r:
                 result = r.get("result", None)
                 if isinstance(result, ExecutionBlocker):
-                    result = tuple([result] * len(obj.RETURN_TYPES))
+                    result = tuple([result] * expected_count)
                 results.append(result)
                 subgraph_results.append((None, result))
         elif isinstance(r, _NodeOutputInternal):
@@ -371,29 +416,42 @@ def get_output_from_returns(return_values, obj):
                     uis.append(r.ui)
                 else:
                     uis.append(r.ui.as_dict())
-            if r.expand is not None:
+            # Named NodeOutput → reorder against the finalized output list before
+            # downstream code treats this as a fixed-shape tuple.
+            named_result = (
+                _normalize_named_result(r, finalized_outputs)
+                if r.named is not None
+                else None
+            )
+            # block_execution takes precedence: EXECUTE_NORMALIZED converts a bare
+            # ExecutionBlocker return into NodeOutput(block_execution=msg) with no
+            # positional / named result, so we must shape the tuple ourselves.
+            if r.block_execution is not None:
+                result = tuple([ExecutionBlocker(r.block_execution)] * expected_count)
+                if r.expand is not None:
+                    has_subgraph = True
+                    subgraph_results.append((r.expand, result))
+                else:
+                    results.append(result)
+                    subgraph_results.append((None, result))
+            elif r.expand is not None:
                 has_subgraph = True
-                new_graph = r.expand
-                result = r.result
-                if r.block_execution is not None:
-                    result = tuple([ExecutionBlocker(r.block_execution)] * len(obj.RETURN_TYPES))
-                subgraph_results.append((new_graph, result))
-            elif r.result is not None:
-                result = r.result
-                if r.block_execution is not None:
-                    result = tuple([ExecutionBlocker(r.block_execution)] * len(obj.RETURN_TYPES))
+                result = named_result if named_result is not None else r.result
+                subgraph_results.append((r.expand, result))
+            elif named_result is not None or r.result is not None:
+                result = named_result if named_result is not None else r.result
                 results.append(result)
                 subgraph_results.append((None, result))
         else:
             if isinstance(r, ExecutionBlocker):
-                r = tuple([r] * len(obj.RETURN_TYPES))
+                r = tuple([r] * expected_count)
             results.append(r)
             subgraph_results.append((None, r))
 
     if has_subgraph:
         output = subgraph_results
     elif len(results) > 0:
-        output = merge_result_data(results, obj)
+        output = merge_result_data(results, obj, finalized_outputs=finalized_outputs)
     else:
         output = []
     ui = dict()
@@ -441,6 +499,24 @@ async def execute(server, dynprompt, caches, current_item, extra_data, executed,
         execution_list.cache_update(unique_id, cached)
         return (ExecutionResult.SUCCESS, None, None)
 
+    # Finalize active outputs once so initial / async-resume / subgraph-resume paths agree.
+    finalized_outputs = None
+    if issubclass(class_def, _ComfyNodeInternal):
+        schema = class_def.GET_SCHEMA()
+        has_dynamic = any(
+            isinstance(o, (_io.DynamicOutputs.ByKey, _io.DynamicOutputs.BySlot))
+            for o in schema.outputs
+        )
+        if has_dynamic:
+            # live_input_types is only needed for BySlot finalization.
+            needs_live_types = any(isinstance(o, _io.DynamicOutputs.BySlot) for o in schema.outputs)
+            live_input_types = None
+            if needs_live_types and hasattr(dynprompt, "get_type_resolver"):
+                live_input_types = dynprompt.get_type_resolver().compute_live_input_types(unique_id)
+            finalized_outputs = _io.get_finalized_class_outputs(
+                schema.outputs, inputs, live_input_types=live_input_types,
+            )
+
     input_data_all = None
     try:
         if unique_id in pending_async_nodes:
@@ -456,7 +532,7 @@ async def execute(server, dynprompt, caches, current_item, extra_data, executed,
                 else:
                     results.append(r)
             del pending_async_nodes[unique_id]
-            output_data, output_ui, has_subgraph = get_output_from_returns(results, class_def)
+            output_data, output_ui, has_subgraph = get_output_from_returns(results, class_def, finalized_outputs=finalized_outputs)
         elif unique_id in pending_subgraph_results:
             cached_results = pending_subgraph_results[unique_id]
             resolved_outputs = []
@@ -475,7 +551,7 @@ async def execute(server, dynprompt, caches, current_item, extra_data, executed,
                         else:
                             resolved_output.append(r)
                     resolved_outputs.append(tuple(resolved_output))
-            output_data = merge_result_data(resolved_outputs, class_def)
+            output_data = merge_result_data(resolved_outputs, class_def, finalized_outputs=finalized_outputs)
             output_ui = []
             del pending_subgraph_results[unique_id]
             has_subgraph = False
@@ -533,7 +609,7 @@ async def execute(server, dynprompt, caches, current_item, extra_data, executed,
                 GraphBuilder.set_default_prefix(unique_id, call_index, 0)
 
             try:
-                output_data, output_ui, has_subgraph, has_pending_tasks = await get_output_data(prompt_id, unique_id, obj, input_data_all, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb, v3_data=v3_data)
+                output_data, output_ui, has_subgraph, has_pending_tasks = await get_output_data(prompt_id, unique_id, obj, input_data_all, execution_block_cb=execution_block_cb, pre_execute_cb=pre_execute_cb, v3_data=v3_data, finalized_outputs=finalized_outputs)
             finally:
                 if comfy.memory_management.aimdo_enabled:
                     if args.verbose == "DEBUG":
@@ -821,9 +897,17 @@ class PromptExecutor:
             self._notify_prompt_lifecycle("end", prompt_id)
 
 
-async def validate_inputs(prompt_id, prompt, item, validated, visiting=None):
+async def validate_inputs(prompt_id, prompt, item, validated, visiting=None, type_resolver=None):
+    """Validate inputs for a single node, recursing into upstream nodes.
+
+    ``type_resolver`` is built once at the top of recursion and shared so
+    MatchType chains are only walked once per prompt.
+    """
     if visiting is None:
         visiting = []
+    if type_resolver is None:
+        from comfy_execution.type_resolver import TypeResolver
+        type_resolver = TypeResolver(prompt)
 
     unique_id = item
     if unique_id in validated:
@@ -855,10 +939,12 @@ async def validate_inputs(prompt_id, prompt, item, validated, visiting=None):
     v3_data = None
     validate_function_inputs = []
     validate_has_kwargs = False
+    live_input_types = None
     if issubclass(obj_class, _ComfyNodeInternal):
         obj_class: _io._ComfyNodeBaseInternal
         class_inputs = obj_class.INPUT_TYPES()
-        class_inputs, _, v3_data = _io.get_finalized_class_inputs(class_inputs, inputs)
+        live_input_types = type_resolver.compute_live_input_types(unique_id)
+        class_inputs, _, v3_data = _io.get_finalized_class_inputs(class_inputs, inputs, live_input_types=live_input_types)
         validate_function_name = "validate_inputs"
         validate_function = first_real_override(obj_class, validate_function_name)
     else:
@@ -908,11 +994,46 @@ async def validate_inputs(prompt_id, prompt, item, validated, visiting=None):
                 continue
 
             o_id = val[0]
-            o_class_type = prompt[o_id]['class_type']
-            r = nodes.NODE_CLASS_MAPPINGS[o_class_type].RETURN_TYPES
-            received_type = r[val[1]]
+            if not isinstance(o_id, str):
+                errors.append({
+                    "type": "bad_linked_input",
+                    "message": "Bad linked input, source node id must be a string",
+                    "details": f"{x}, linked_node({val})",
+                    "extra_info": {"input_name": x, "linked_node": val}
+                })
+                continue
+            # Reject links pointing at slot indices outside the upstream node's
+            # active output count (e.g. stale link after a DynamicOutputs branch
+            # change).
+            upstream_output_count = type_resolver.finalized_output_count(o_id)
+            if not isinstance(val[1], int) or isinstance(val[1], bool) or val[1] < 0 or val[1] >= upstream_output_count:
+                error = {
+                    "type": "bad_linked_output",
+                    "message": "Linked output slot does not exist on the source node",
+                    "details": f"{x}, linked_node({o_id}), output_index({val[1]}), active_output_count({upstream_output_count})",
+                    "extra_info": {
+                        "input_name": x,
+                        "linked_node": val,
+                        "output_index": val[1],
+                        "active_output_count": upstream_output_count,
+                    }
+                }
+                errors.append(error)
+                continue
+            # Walks MatchType/template chains so API workflows without
+            # 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's declared input_type is just the dispatch tag
+            # (COMFY_DYNAMICSLOT_V3); a link is valid iff some Option would
+            # actually claim the resolved upstream type.
+            if input_type == _io.DynamicSlot.io_type and isinstance(extra_info, dict):
+                link_valid = _io.DynamicSlot._select_option(
+                    extra_info.get("options", []), {x: received_type}, x, has_link=True
+                ) is not None
+            else:
+                link_valid = validate_node_input(received_type, input_type)
+            if 'input_types' not in validate_function_inputs and not link_valid:
                 details = f"{x}, received_type({received_type}) mismatch input_type({input_type})"
                 error = {
                     "type": "return_type_mismatch",
@@ -930,7 +1051,7 @@ async def validate_inputs(prompt_id, prompt, item, validated, visiting=None):
             try:
                 visiting.append(unique_id)
                 try:
-                    r = await validate_inputs(prompt_id, prompt, o_id, validated, visiting)
+                    r = await validate_inputs(prompt_id, prompt, o_id, validated, visiting, type_resolver)
                 finally:
                     visiting.pop()
                 if r[0] is False:
@@ -1058,7 +1179,11 @@ async def validate_inputs(prompt_id, prompt, item, validated, visiting=None):
                         continue
 
     if len(validate_function_inputs) > 0 or validate_has_kwargs:
-        input_data_all, _, v3_data = get_input_data(inputs, obj_class, unique_id)
+        # Reuse the precomputed live_input_types so a custom validate_inputs()
+        # sees the same DynamicSlot branch that finalization picked above.
+        input_data_all, _, v3_data = get_input_data(
+            inputs, obj_class, unique_id, live_input_types=live_input_types
+        )
         input_filtered = {}
         for x in input_data_all:
             if x in validate_function_inputs or validate_has_kwargs:
@@ -1155,11 +1280,14 @@ async def validate_prompt(prompt_id, prompt, partial_execution_list: Union[list[
     errors = []
     node_errors = {}
     validated = {}
+    # Shared across output validations so MatchType chains walk only once.
+    from comfy_execution.type_resolver import TypeResolver
+    type_resolver = TypeResolver(prompt)
     for o in outputs:
         valid = False
         reasons = []
         try:
-            m = await validate_inputs(prompt_id, prompt, o, validated)
+            m = await validate_inputs(prompt_id, prompt, o, validated, None, type_resolver)
             valid = m[0]
             reasons = m[1]
         except Exception as ex:

tests-unit/comfy_api_test/test_dynamic_outputs.py

@@ -0,0 +1,537 @@
+"""Unit tests for ``DynamicOutputs.ByKey`` and the finalized-outputs path."""
+
+import pytest
+
+from comfy_api.latest import _io as io
+
+
+# ---------------------------------------------------------------------------
+# Schema-level construction and validation
+# ---------------------------------------------------------------------------
+
+def _byke():
+    return io.DynamicOutputs.ByKey(
+        id="result",
+        selector="mode",
+        options=[
+            io.DynamicOutputs.Option(key="image",
+                                     outputs=[io.Image.Output("image"), io.Mask.Output("mask")]),
+            io.DynamicOutputs.Option(key="latent",
+                                     outputs=[io.Latent.Output("latent")]),
+        ],
+    )
+
+
+def test_option_rejects_empty_key():
+    with pytest.raises(ValueError, match="non-empty string"):
+        io.DynamicOutputs.Option(key="", outputs=[])
+
+
+def test_option_rejects_non_output_entry():
+    with pytest.raises(ValueError, match="Output instances"):
+        io.DynamicOutputs.Option(key="x", outputs=["not an output"])
+
+
+def test_option_requires_explicit_output_ids():
+    with pytest.raises(ValueError, match="declare an id"):
+        io.DynamicOutputs.Option(key="x", outputs=[io.Image.Output()])  # no id
+
+
+def test_bykey_rejects_empty_options():
+    with pytest.raises(ValueError, match="at least one Option"):
+        io.DynamicOutputs.ByKey(id="r", selector="m", options=[])
+
+
+def test_bykey_rejects_duplicate_keys():
+    with pytest.raises(ValueError, match="duplicate option key"):
+        io.DynamicOutputs.ByKey(
+            id="r", selector="m",
+            options=[
+                io.DynamicOutputs.Option(key="x", outputs=[io.Image.Output("a")]),
+                io.DynamicOutputs.Option(key="x", outputs=[io.Latent.Output("b")]),
+            ],
+        )
+
+
+def test_bykey_rejects_duplicate_output_ids_across_options():
+    with pytest.raises(ValueError, match="appears in more than one option"):
+        io.DynamicOutputs.ByKey(
+            id="r", selector="m",
+            options=[
+                io.DynamicOutputs.Option(key="x", outputs=[io.Image.Output("dup")]),
+                io.DynamicOutputs.Option(key="y", outputs=[io.Latent.Output("dup")]),
+            ],
+        )
+
+
+# ---------------------------------------------------------------------------
+# Schema integration
+# ---------------------------------------------------------------------------
+
+def _make_node(extra_outputs=None):
+    """Build a V3 node class with a selector input + DynamicOutputs group."""
+    extras = extra_outputs or []
+
+    class DynNode(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="DynNode",
+                inputs=[io.Combo.Input("mode", options=["image", "latent"], default="image")],
+                outputs=[*extras, _byke()],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({"image": None, "mask": None})
+
+    return DynNode
+
+
+def test_schema_validate_rejects_unknown_selector():
+    class BadSelector(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="BadSelector",
+                inputs=[io.Combo.Input("not_mode", options=["a"])],
+                outputs=[
+                    io.DynamicOutputs.ByKey(
+                        id="r", selector="mode",
+                        options=[io.DynamicOutputs.Option(key="a", outputs=[io.Image.Output("a")])],
+                    ),
+                ],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({"a": None})
+
+    with pytest.raises(ValueError, match="selector input 'mode' does not exist"):
+        BadSelector.GET_SCHEMA()
+
+
+def test_schema_validate_rejects_id_collision_with_static_output():
+    class Collision(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="Collision",
+                inputs=[io.Combo.Input("mode", options=["a"])],
+                outputs=[
+                    io.Image.Output("shared"),
+                    io.DynamicOutputs.ByKey(
+                        id="r", selector="mode",
+                        options=[io.DynamicOutputs.Option(key="a", outputs=[io.Latent.Output("shared")])],
+                    ),
+                ],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({"shared": None})
+
+    with pytest.raises(ValueError, match="Output ids must be unique"):
+        Collision.GET_SCHEMA()
+
+
+def test_schema_get_v1_info_emits_dynamic_outputs_field():
+    DynNode = _make_node()
+    DynNode.GET_SCHEMA()
+    info = DynNode.SCHEMA.get_v1_info(DynNode)
+    assert info.dynamic_outputs is not None and len(info.dynamic_outputs) == 1
+    group = info.dynamic_outputs[0]
+    assert group["kind"] == "by_key"
+    assert group["selector"] == "mode"
+    assert {opt["key"] for opt in group["options"]} == {"image", "latent"}
+    # Static output arrays are empty — only the dynamic group is declared.
+    assert info.output == []
+    assert info.output_is_list == []
+
+
+def test_schema_static_outputs_stable_prefix_in_v1_arrays():
+    """A static output before a dynamic group still surfaces in RETURN_TYPES etc."""
+    DynNode = _make_node(extra_outputs=[io.String.Output("status")])
+    DynNode.GET_SCHEMA()
+    # Class-level static arrays are the always-present prefix.
+    assert list(DynNode.RETURN_TYPES) == ["STRING"]
+    assert list(DynNode.RETURN_NAMES) == ["status"]
+    assert list(DynNode.OUTPUT_IS_LIST) == [False]
+
+
+# ---------------------------------------------------------------------------
+# get_finalized_class_outputs
+# ---------------------------------------------------------------------------
+
+def test_finalize_picks_active_branch():
+    schema_outputs = [_byke()]
+    finalized = io.get_finalized_class_outputs(schema_outputs, {"mode": "latent"})
+    assert finalized.output_ids == ["latent"]
+    assert finalized.return_types == ["LATENT"]
+    assert finalized.output_is_list == [False]
+
+
+def test_finalize_unknown_selector_yields_empty():
+    schema_outputs = [_byke()]
+    finalized = io.get_finalized_class_outputs(schema_outputs, {"mode": "nonexistent"})
+    assert len(finalized) == 0
+
+
+def test_finalize_link_selector_yields_empty():
+    """Link as selector value is treated as 'not finalizable' — no branch."""
+    schema_outputs = [_byke()]
+    finalized = io.get_finalized_class_outputs(schema_outputs, {"mode": ["src", 0]})
+    assert len(finalized) == 0
+
+
+def test_finalize_static_prefix_preserved():
+    schema_outputs = [io.String.Output("status"), _byke()]
+    finalized = io.get_finalized_class_outputs(schema_outputs, {"mode": "image"})
+    assert finalized.output_ids == ["status", "image", "mask"]
+    assert finalized.return_types == ["STRING", "IMAGE", "MASK"]
+
+
+# ---------------------------------------------------------------------------
+# NodeOutput.from_named
+# ---------------------------------------------------------------------------
+
+def test_nodeoutput_from_named_stores_dict():
+    out = io.NodeOutput.from_named({"a": 1, "b": 2})
+    assert out.named == {"a": 1, "b": 2}
+    assert out.args == ()
+    assert out.result is None  # `.result` is the positional tuple
+
+
+def test_nodeoutput_rejects_mixed_positional_and_named():
+    with pytest.raises(ValueError, match="cannot mix positional"):
+        io.NodeOutput(1, 2, named={"a": 1})
+
+
+# ---------------------------------------------------------------------------
+# Group-id uniqueness
+# ---------------------------------------------------------------------------
+
+def test_schema_rejects_duplicate_dynamic_group_ids():
+    class Dup(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="Dup",
+                inputs=[io.Combo.Input("mode", options=["a"])],
+                outputs=[
+                    io.DynamicOutputs.ByKey(id="r", selector="mode",
+                        options=[io.DynamicOutputs.Option(key="a", outputs=[io.Image.Output("x")])]),
+                    io.DynamicOutputs.ByKey(id="r", selector="mode",
+                        options=[io.DynamicOutputs.Option(key="a", outputs=[io.Latent.Output("y")])]),
+                ],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({"x": None, "y": None})
+
+    with pytest.raises(ValueError, match="DynamicOutputs group ids must be unique"):
+        Dup.GET_SCHEMA()
+
+
+# ---------------------------------------------------------------------------
+# DynamicOutputs.ByKey with a DynamicCombo selector
+# ---------------------------------------------------------------------------
+
+def _combo_input():
+    return io.DynamicCombo.Input("mode", options=[
+        io.DynamicCombo.Option(key="image", inputs=[io.Image.Input("img")]),
+        io.DynamicCombo.Option(key="latent", inputs=[io.Latent.Input("lat")]),
+    ])
+
+
+def _bykey_outputs():
+    return io.DynamicOutputs.ByKey(id="result", selector="mode", options=[
+        io.DynamicOutputs.Option(key="image", outputs=[io.Image.Output("processed"), io.Mask.Output("alpha")]),
+        io.DynamicOutputs.Option(key="latent", outputs=[io.Latent.Output("denoised")]),
+    ])
+
+
+def test_bykey_with_dynamic_combo_finalizes_branch():
+    finalized = io.get_finalized_class_outputs(
+        [io.String.Output("status"), _bykey_outputs()],
+        {"mode": {"mode": "image", "img": None}},  # DynamicCombo dispatch shape
+    )
+    assert finalized.output_ids == ["status", "processed", "alpha"]
+    assert finalized.return_types == ["STRING", "IMAGE", "MASK"]
+
+
+def test_bykey_with_dynamic_combo_other_branch():
+    finalized = io.get_finalized_class_outputs(
+        [_bykey_outputs()],
+        {"mode": {"mode": "latent", "lat": None}},
+    )
+    assert finalized.output_ids == ["denoised"]
+
+
+def test_schema_rejects_bykey_key_not_on_dynamic_combo():
+    class StrayKey(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="StrayKey",
+                inputs=[_combo_input()],
+                outputs=[io.DynamicOutputs.ByKey(id="r", selector="mode", options=[
+                    io.DynamicOutputs.Option(key="image", outputs=[io.Image.Output("a")]),
+                    io.DynamicOutputs.Option(key="audio", outputs=[io.String.Output("b")]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    with pytest.raises(ValueError, match=r"option key\(s\) \['audio'\] are not declared"):
+        StrayKey.GET_SCHEMA()
+
+
+# ---------------------------------------------------------------------------
+# DynamicOutputs.BySlot
+# ---------------------------------------------------------------------------
+
+def _slot_input():
+    return io.DynamicSlot.Input("slot", options=[
+        io.DynamicSlot.Option(when=io.Image),
+        io.DynamicSlot.Option(when=io.Latent),
+        io.DynamicSlot.Option(when=None, inputs=[io.Int.Input("seed")]),
+    ])
+
+
+def _byslot_outputs():
+    return io.DynamicOutputs.BySlot(id="slot_out", selector="slot", options=[
+        io.DynamicOutputs.SlotOption(when=io.Image, outputs=[io.Image.Output("processed"), io.Mask.Output("alpha")]),
+        io.DynamicOutputs.SlotOption(when=io.Latent, outputs=[io.Latent.Output("denoised")]),
+        io.DynamicOutputs.SlotOption(when=None, outputs=[]),
+    ])
+
+
+def test_byslot_finalizes_by_resolved_type():
+    finalized = io.get_finalized_class_outputs(
+        [_byslot_outputs()],
+        {"slot": ["upstream", 0]},
+        live_input_types={"slot": "IMAGE"},
+    )
+    assert finalized.output_ids == ["processed", "alpha"]
+    finalized = io.get_finalized_class_outputs(
+        [_byslot_outputs()],
+        {"slot": ["upstream", 0]},
+        live_input_types={"slot": "LATENT"},
+    )
+    assert finalized.output_ids == ["denoised"]
+
+
+def test_byslot_unconnected_uses_when_none():
+    finalized = io.get_finalized_class_outputs([_byslot_outputs()], {})
+    # when=None option declares outputs=[] → no active outputs
+    assert finalized.output_ids == []
+
+
+def test_byslot_unmatched_type_yields_empty():
+    finalized = io.get_finalized_class_outputs(
+        [_byslot_outputs()],
+        {"slot": ["upstream", 0]},
+        live_input_types={"slot": "AUDIO"},
+    )
+    assert finalized.output_ids == []
+
+
+def test_byslot_rejects_duplicate_when_types():
+    with pytest.raises(ValueError, match="appears in more than one option"):
+        io.DynamicOutputs.BySlot(id="r", selector="slot", options=[
+            io.DynamicOutputs.SlotOption(when=io.Image, outputs=[io.Image.Output("a")]),
+            io.DynamicOutputs.SlotOption(when=io.Image, outputs=[io.Mask.Output("b")]),
+        ])
+
+
+def test_byslot_rejects_duplicate_when_none():
+    with pytest.raises(ValueError, match="only one option may declare when=None"):
+        io.DynamicOutputs.BySlot(id="r", selector="slot", options=[
+            io.DynamicOutputs.SlotOption(when=None, outputs=[]),
+            io.DynamicOutputs.SlotOption(when=None, outputs=[io.Image.Output("x")]),
+        ])
+
+
+def test_schema_rejects_byslot_selector_not_a_dynamic_slot():
+    class WrongSel(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="WrongSel",
+                inputs=[io.Combo.Input("not_a_slot", options=["a"])],
+                outputs=[io.DynamicOutputs.BySlot(id="r", selector="not_a_slot", options=[
+                    io.DynamicOutputs.SlotOption(when=io.Image, outputs=[io.Image.Output("x")]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    with pytest.raises(ValueError, match="must reference a DynamicSlot input"):
+        WrongSel.GET_SCHEMA()
+
+
+def test_schema_rejects_byslot_when_type_not_on_slot():
+    class StrayWhen(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="StrayWhen",
+                inputs=[_slot_input()],
+                outputs=[io.DynamicOutputs.BySlot(id="r", selector="slot", options=[
+                    io.DynamicOutputs.SlotOption(when=io.Audio, outputs=[io.Audio.Output("x")]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    with pytest.raises(ValueError, match=r"type\(s\) \['AUDIO'\] are not accepted"):
+        StrayWhen.GET_SCHEMA()
+
+
+def test_schema_rejects_byslot_when_none_without_slot_when_none():
+    class NoNone(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="NoNone",
+                inputs=[io.DynamicSlot.Input("slot", optional=False, options=[
+                    io.DynamicSlot.Option(when=io.Image),
+                ])],
+                outputs=[io.DynamicOutputs.BySlot(id="r", selector="slot", options=[
+                    io.DynamicOutputs.SlotOption(when=None, outputs=[]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    with pytest.raises(ValueError, match="requires DynamicSlot 'slot' to declare a when=None"):
+        NoNone.GET_SCHEMA()
+
+
+def test_v1_info_emits_byslot_entry():
+    class N(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="SlotV1",
+                inputs=[_slot_input()],
+                outputs=[_byslot_outputs()],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    N.GET_SCHEMA()
+    info = N.SCHEMA.get_v1_info(N)
+    assert info.dynamic_outputs is not None and len(info.dynamic_outputs) == 1
+    entry = info.dynamic_outputs[0]
+    assert entry["kind"] == "by_slot"
+    assert entry["selector"] == "slot"
+    whens = [opt["when"] for opt in entry["options"]]
+    assert whens == [["IMAGE"], ["LATENT"], None]
+
+
+# ---------------------------------------------------------------------------
+# Nested dynamic selectors (selector reaches into a DynamicCombo / DynamicSlot
+# option's nested inputs via a dotted path).
+# ---------------------------------------------------------------------------
+
+def test_schema_accepts_byslot_selector_into_nested_dynamic_slot():
+    """A BySlot selector may target a DynamicSlot nested inside a DynamicCombo option."""
+    class Nested(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="NestedBySlot",
+                inputs=[
+                    io.DynamicCombo.Input("outer", options=[
+                        io.DynamicCombo.Option(key="a", inputs=[
+                            io.DynamicSlot.Input("inner_slot", options=[
+                                io.DynamicSlot.Option(when=io.Image),
+                                io.DynamicSlot.Option(when=None),
+                            ]),
+                        ]),
+                    ]),
+                ],
+                outputs=[io.DynamicOutputs.BySlot(id="r", selector="outer.inner_slot", options=[
+                    io.DynamicOutputs.SlotOption(when=io.Image, outputs=[io.Image.Output("img")]),
+                    io.DynamicOutputs.SlotOption(when=None, outputs=[]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    # Must not raise — selector resolves to the nested DynamicSlot via dotted path.
+    Nested.GET_SCHEMA()
+
+
+def test_schema_accepts_bykey_selector_into_nested_dynamic_combo():
+    """A ByKey selector may target a DynamicCombo nested inside another DynamicCombo option."""
+    class Nested(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="NestedByKey",
+                inputs=[
+                    io.DynamicCombo.Input("outer", options=[
+                        io.DynamicCombo.Option(key="branch", inputs=[
+                            io.DynamicCombo.Input("inner", options=[
+                                io.DynamicCombo.Option(key="x", inputs=[]),
+                                io.DynamicCombo.Option(key="y", inputs=[]),
+                            ]),
+                        ]),
+                    ]),
+                ],
+                outputs=[io.DynamicOutputs.ByKey(id="r", selector="outer.inner", options=[
+                    io.DynamicOutputs.Option(key="x", outputs=[io.Int.Output("ix")]),
+                    io.DynamicOutputs.Option(key="y", outputs=[io.String.Output("iy")]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    Nested.GET_SCHEMA()
+
+
+def test_schema_rejects_byslot_nested_when_type_not_on_target_slot():
+    """The when-type alignment check follows the dotted selector into the nested slot."""
+    class BadNested(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="BadNested",
+                inputs=[
+                    io.DynamicCombo.Input("outer", options=[
+                        io.DynamicCombo.Option(key="a", inputs=[
+                            io.DynamicSlot.Input("inner_slot", options=[
+                                io.DynamicSlot.Option(when=io.Image),
+                            ]),
+                        ]),
+                    ]),
+                ],
+                outputs=[io.DynamicOutputs.BySlot(id="r", selector="outer.inner_slot", options=[
+                    io.DynamicOutputs.SlotOption(when=io.Audio, outputs=[io.Audio.Output("x")]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    with pytest.raises(ValueError, match=r"type\(s\) \['AUDIO'\] are not accepted"):
+        BadNested.GET_SCHEMA()

tests-unit/comfy_api_test/test_dynamic_slot.py

@@ -0,0 +1,350 @@
+"""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._when_set is None
+    assert o.as_dict()["when"] is None
+
+
+def test_option_when_single_type():
+    o = _opt(io.Image)
+    assert o._when_types == ("IMAGE",)
+    assert o._when_set == frozenset({"IMAGE"})
+    assert o.as_dict()["when"] == ["IMAGE"]
+
+
+def test_option_when_anytype():
+    o = _opt(io.AnyType)
+    assert o._when_types == ("*",)
+    assert o.as_dict()["when"] == ["*"]
+
+
+def test_option_when_list_preserves_order():
+    """Declaration order is preserved in both the tuple and the serialized form."""
+    o = _opt([io.Mask, io.Image])
+    assert o._when_types == ("MASK", "IMAGE")
+    assert o.as_dict()["when"] == ["MASK", "IMAGE"]
+
+
+def test_option_when_list_dedups_within_option():
+    o = _opt([io.Image, io.Image, io.Mask])
+    assert o._when_types == ("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 == ("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=[])
+
+
+def test_option_when_list_with_anytype_rejected():
+    """AnyType must stand alone — it represents the unresolvable-wildcard
+    state, not a concrete type that can share a branch with concrete types."""
+    with pytest.raises(ValueError, match="AnyType cannot be grouped"):
+        io.DynamicSlot.Option(when=[io.Image, io.AnyType], inputs=[])
+
+
+def test_option_when_multitype_with_anytype_rejected():
+    mt = io.MultiType.Input("x", types=[io.Image, io.AnyType])
+    with pytest.raises(ValueError, match="AnyType cannot be grouped"):
+        io.DynamicSlot.Option(when=mt, 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.
+    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_rejects_duplicate_type_across_options():
+    with pytest.raises(ValueError, match="appears in more than one"):
+        io.DynamicSlot.Input("x", options=[
+            _opt(io.Image, ["a"]),
+            _opt([io.Image, io.Mask], ["b"]),
+        ])
+
+
+def test_input_rejects_duplicate_anytype_across_options():
+    with pytest.raises(ValueError, match="appears in more than one"):
+        io.DynamicSlot.Input("x", options=[
+            _opt(io.AnyType, ["a"]),
+            _opt(io.AnyType, ["b"]),
+        ])
+
+
+def test_input_rejects_duplicate_when_none():
+    with pytest.raises(ValueError, match="only one Option may declare when=None"):
+        io.DynamicSlot.Input("x", options=[
+            _opt(io.Image, ["a"]),
+            _opt(None, ["b"]),
+            _opt(None, ["c"]),
+        ])
+
+
+def test_input_rejects_non_option_entry():
+    with pytest.raises(ValueError, match="must be DynamicSlot.Option instances"):
+        io.DynamicSlot.Input("x", options=[_opt(io.Image, ["a"]), "not an option"])
+
+
+def test_input_defaults_to_optional_and_always_force_input():
+    """The slot is always rendered as a connector, never as a widget, even
+    when slotType includes widget-capable types like INT/STRING."""
+    inp = io.DynamicSlot.Input("x", options=[_opt(io.Int, ["n"])])
+    d = inp.as_dict()
+    assert d["forceInput"] is True
+    # default optional=True → slot lives in optional bucket via DynamicInput
+    assert inp.optional is True
+
+
+def test_input_required_slot_allowed_without_when_none():
+    inp = io.DynamicSlot.Input("x", optional=False, options=[_opt(io.Image, ["a"])])
+    assert inp.optional is False
+
+
+def test_input_required_slot_rejects_when_none_option():
+    with pytest.raises(ValueError, match="optional=False forbids when=None"):
+        io.DynamicSlot.Input(
+            "x",
+            optional=False,
+            options=[_opt(io.Image, ["a"]), _opt(None, ["b"])],
+        )
+
+
+def test_input_get_all_prepends_self_and_dedups_children():
+    inp = io.DynamicSlot.Input("x", options=[
+        _opt(io.Image, ["shared", "image_only"]),
+        _opt(io.Mask, ["shared", "mask_only"]),
+    ])
+    items = inp.get_all()
+    # Convention shared with Autogrow / DynamicCombo: parent first, then children.
+    assert items[0] is inp
+    assert [i.id for i in items[1:]] == ["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_anytype_branch_does_not_swallow_unenumerated_concrete():
+    """Regression: a slot exposing IMAGE + AnyType must reject LATENT upstream
+    instead of expanding the AnyType branch. validate_inputs relies on this
+    to compute link validity (slotType="IMAGE,*" alone would over-accept)."""
+    options = [_opt(io.Image, ["image_widget"]), _opt(io.AnyType, ["any_widget"])]
+    assert _select(options, {"x": "LATENT"}, has_link=True) is None
+    # Sanity: IMAGE still matches the IMAGE branch and "*" still matches AnyType.
+    assert _select(options, {"x": "IMAGE"}, has_link=True)["when"] == ["IMAGE"]
+    assert _select(options, {"x": "*"}, has_link=True)["when"] == ["*"]
+
+
+def test_select_first_match_wins_on_union_upstream():
+    """Ordering only matters when upstream declares a multi-type union; with
+    per-option type uniqueness, single concrete types can never match two
+    options."""
+    options = [
+        _opt([io.Image, io.Mask], ["image_or_mask"]),
+        _opt(io.Latent, ["latent_only"]),
+    ]
+    # Upstream union "IMAGE,LATENT" intersects both options; first option wins.
+    sel = _select(options, {"x": "IMAGE,LATENT"}, has_link=True)
+    first_input_id = next(iter(sel["inputs"]["required"].keys()))
+    assert first_input_id == "image_or_mask"
+
+
+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"]

tests-unit/execution_test/test_dynamic_outputs_resolver.py

@@ -0,0 +1,376 @@
+"""TypeResolver + execution-helper tests for ``DynamicOutputs.ByKey``.
+
+Covers the wiring between the per-prompt finalized output list and the
+execution layer:
+
+  * type resolver returns the active branch's declared type
+  * type resolver reports the active output count for stale-link validation
+  * ``is_output_list`` reflects the active branch
+  * execution helpers refuse to consume ``NodeOutput(named=...)`` against a
+    non-dynamic node, and reorder against the finalized list for dynamic ones
+"""
+
+from __future__ import annotations
+
+import sys
+import types as _pytypes
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Shared fixtures (mirror tests-unit/execution_test/test_type_resolver.py)
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def fake_nodes_module():
+    real_nodes = sys.modules.get("nodes")
+    fake = _pytypes.ModuleType("nodes")
+    fake.NODE_CLASS_MAPPINGS = {}
+    sys.modules["nodes"] = fake
+    try:
+        yield fake.NODE_CLASS_MAPPINGS
+    finally:
+        if real_nodes is not None:
+            sys.modules["nodes"] = real_nodes
+        else:
+            del sys.modules["nodes"]
+
+
+@pytest.fixture
+def TypeResolver(fake_nodes_module):
+    from comfy_execution.type_resolver import TypeResolver as TR
+    return TR
+
+
+def _v1_node(return_types: tuple[str, ...]):
+    class _V1:
+        RETURN_TYPES = return_types
+
+        @classmethod
+        def INPUT_TYPES(cls):
+            return {"required": {}}
+
+    return _V1
+
+
+def _make_dyn_node():
+    """V3 node: ``mode`` selector with two branches."""
+    from comfy_api.latest import _io as io
+
+    class DynBranch(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="DynBranch",
+                inputs=[io.Combo.Input("mode", options=["image", "latent"], default="image")],
+                outputs=[
+                    io.DynamicOutputs.ByKey(
+                        id="result", selector="mode",
+                        options=[
+                            io.DynamicOutputs.Option(key="image", outputs=[
+                                io.Image.Output("image"),
+                                io.Mask.Output("mask"),
+                            ]),
+                            io.DynamicOutputs.Option(key="latent", outputs=[
+                                io.Latent.Output("latent"),
+                            ]),
+                        ],
+                    ),
+                ],
+            )
+
+        @classmethod
+        def execute(cls, mode):
+            if mode == "latent":
+                return io.NodeOutput.from_named({"latent": None})
+            return io.NodeOutput.from_named({"image": None, "mask": None})
+
+    DynBranch.GET_SCHEMA()
+    return DynBranch
+
+
+# ---------------------------------------------------------------------------
+# TypeResolver against finalized outputs
+# ---------------------------------------------------------------------------
+
+def test_dynamic_resolve_picks_active_branch_image(fake_nodes_module, TypeResolver):
+    fake_nodes_module["DynBranch"] = _make_dyn_node()
+    prompt = {"n1": {"class_type": "DynBranch", "inputs": {"mode": "image"}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("n1", 0) == "IMAGE"
+    assert r.resolve_output_type("n1", 1) == "MASK"
+
+
+def test_dynamic_resolve_picks_active_branch_latent(fake_nodes_module, TypeResolver):
+    fake_nodes_module["DynBranch"] = _make_dyn_node()
+    prompt = {"n1": {"class_type": "DynBranch", "inputs": {"mode": "latent"}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("n1", 0) == "LATENT"
+
+
+def test_dynamic_finalized_output_count(fake_nodes_module, TypeResolver):
+    fake_nodes_module["DynBranch"] = _make_dyn_node()
+    fake_nodes_module["Static"] = _v1_node(("INT", "FLOAT"))
+    prompt = {
+        "img": {"class_type": "DynBranch", "inputs": {"mode": "image"}},
+        "lat": {"class_type": "DynBranch", "inputs": {"mode": "latent"}},
+        "stat": {"class_type": "Static", "inputs": {}},
+    }
+    r = TypeResolver(prompt)
+    assert r.finalized_output_count("img") == 2  # image + mask
+    assert r.finalized_output_count("lat") == 1
+    assert r.finalized_output_count("stat") == 2  # static V1 falls through
+
+
+def test_dynamic_is_output_list_reflects_branch(fake_nodes_module, TypeResolver):
+    from comfy_api.latest import _io as io
+
+    class DynList(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="DynList",
+                inputs=[io.Combo.Input("mode", options=["one", "many"], default="one")],
+                outputs=[
+                    io.DynamicOutputs.ByKey(
+                        id="r", selector="mode",
+                        options=[
+                            io.DynamicOutputs.Option(key="one", outputs=[
+                                io.Image.Output("img"),
+                            ]),
+                            io.DynamicOutputs.Option(key="many", outputs=[
+                                io.Image.Output("imgs", is_output_list=True),
+                            ]),
+                        ],
+                    ),
+                ],
+            )
+
+        @classmethod
+        def execute(cls, mode):
+            return io.NodeOutput.from_named({"img": None} if mode == "one" else {"imgs": [None]})
+
+    DynList.GET_SCHEMA()
+    fake_nodes_module["DynList"] = DynList
+    prompt = {
+        "one": {"class_type": "DynList", "inputs": {"mode": "one"}},
+        "many": {"class_type": "DynList", "inputs": {"mode": "many"}},
+    }
+    r = TypeResolver(prompt)
+    assert r.is_output_list("one", 0) is False
+    assert r.is_output_list("many", 0) is True
+
+
+def test_dynamic_out_of_range_returns_any(fake_nodes_module, TypeResolver):
+    """Slot index beyond the finalized branch resolves to AnyType (validation rejects separately)."""
+    fake_nodes_module["DynBranch"] = _make_dyn_node()
+    prompt = {"n1": {"class_type": "DynBranch", "inputs": {"mode": "latent"}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("n1", 5) == "*"
+
+
+# ---------------------------------------------------------------------------
+# Execution-side helpers
+# ---------------------------------------------------------------------------
+
+def test_normalize_named_result_reorders_to_finalized():
+    from comfy_api.latest import _io as io
+    from execution import _normalize_named_result
+
+    finalized = io.get_finalized_class_outputs(
+        [io.DynamicOutputs.ByKey(
+            id="r", selector="mode",
+            options=[io.DynamicOutputs.Option(key="x", outputs=[
+                io.Image.Output("a"), io.Mask.Output("b"), io.Latent.Output("c"),
+            ])],
+        )],
+        {"mode": "x"},
+    )
+    node_output = io.NodeOutput.from_named({"c": 30, "a": 10, "b": 20})
+    assert _normalize_named_result(node_output, finalized) == (10, 20, 30)
+
+
+def test_normalize_named_result_rejects_unknown_or_missing_ids():
+    from comfy_api.latest import _io as io
+    from execution import _normalize_named_result
+
+    finalized = io.get_finalized_class_outputs(
+        [io.DynamicOutputs.ByKey(
+            id="r", selector="mode",
+            options=[io.DynamicOutputs.Option(key="x", outputs=[
+                io.Image.Output("a"), io.Mask.Output("b"),
+            ])],
+        )],
+        {"mode": "x"},
+    )
+    with pytest.raises(Exception, match="missing"):
+        _normalize_named_result(io.NodeOutput.from_named({"a": 1}), finalized)
+    with pytest.raises(Exception, match="unknown"):
+        _normalize_named_result(io.NodeOutput.from_named({"a": 1, "b": 2, "z": 3}), finalized)
+
+
+def test_normalize_named_result_requires_dynamic_node():
+    from comfy_api.latest import _io as io
+    from execution import _normalize_named_result
+
+    with pytest.raises(Exception, match="DynamicOutputs"):
+        _normalize_named_result(io.NodeOutput.from_named({"a": 1}), None)
+
+
+# ---------------------------------------------------------------------------
+# Blocker / output-shape paths through get_output_from_returns
+# ---------------------------------------------------------------------------
+
+def _dyn_finalized(branch_outputs):
+    from comfy_api.latest import _io as io
+    return io.get_finalized_class_outputs(
+        [io.DynamicOutputs.ByKey(id="r", selector="mode", options=[
+            io.DynamicOutputs.Option(key="x", outputs=branch_outputs),
+        ])],
+        {"mode": "x"},
+    )
+
+
+def test_blocker_sized_to_finalized_outputs_for_node_output():
+    """V3 node returning a bare ``ExecutionBlocker`` must yield blocker tuples
+    sized to the active output count, not the empty static RETURN_TYPES."""
+    from comfy_api.latest import _io as io
+    from comfy_execution.graph_utils import ExecutionBlocker
+    from execution import get_output_from_returns
+
+    finalized = _dyn_finalized([io.Image.Output("a"), io.Mask.Output("b")])
+
+    class _Obj:
+        RETURN_TYPES = ()  # only static outputs — dynamic group lives in schema
+
+    out = io.NodeOutput(block_execution="paused")
+    output, _ui, has_subgraph = get_output_from_returns([out], _Obj(), finalized_outputs=finalized)
+    assert has_subgraph is False
+    # merge_result_data flattens per-slot, one input → list-of-one per slot
+    assert len(output) == 2
+    for slot in output:
+        assert len(slot) == 1
+        assert isinstance(slot[0], ExecutionBlocker)
+        assert slot[0].message == "paused"
+
+
+# ---------------------------------------------------------------------------
+# DynamicOutputs.ByKey driven by a DynamicCombo selector (end-to-end resolver)
+# ---------------------------------------------------------------------------
+
+def _make_combo_bykey_node():
+    from comfy_api.latest import _io as io
+
+    class ComboBK(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="ComboBK",
+                inputs=[
+                    io.DynamicCombo.Input("mode", options=[
+                        io.DynamicCombo.Option(key="image", inputs=[io.Image.Input("img")]),
+                        io.DynamicCombo.Option(key="latent", inputs=[io.Latent.Input("lat")]),
+                    ]),
+                ],
+                outputs=[io.DynamicOutputs.ByKey(id="result", selector="mode", options=[
+                    io.DynamicOutputs.Option(key="image",
+                        outputs=[io.Image.Output("processed"), io.Mask.Output("alpha")]),
+                    io.DynamicOutputs.Option(key="latent",
+                        outputs=[io.Latent.Output("denoised")]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, mode, **kwargs):
+            if mode["mode"] == "latent":
+                return io.NodeOutput.from_named({"denoised": None})
+            return io.NodeOutput.from_named({"processed": None, "alpha": None})
+
+    ComboBK.GET_SCHEMA()
+    return ComboBK
+
+
+def _make_slot_byslot_node():
+    from comfy_api.latest import _io as io
+
+    class SlotBS(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="SlotBS",
+                inputs=[
+                    io.DynamicSlot.Input("slot", options=[
+                        io.DynamicSlot.Option(when=io.Image),
+                        io.DynamicSlot.Option(when=io.Latent),
+                        io.DynamicSlot.Option(when=None),
+                    ]),
+                ],
+                outputs=[io.DynamicOutputs.BySlot(id="slot_out", selector="slot", options=[
+                    io.DynamicOutputs.SlotOption(when=io.Image,
+                        outputs=[io.Image.Output("processed"), io.Mask.Output("alpha")]),
+                    io.DynamicOutputs.SlotOption(when=io.Latent,
+                        outputs=[io.Latent.Output("denoised")]),
+                    io.DynamicOutputs.SlotOption(when=None, outputs=[]),
+                ])],
+            )
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput.from_named({})
+
+    SlotBS.GET_SCHEMA()
+    return SlotBS
+
+
+def test_combo_bykey_resolver_picks_branch(fake_nodes_module, TypeResolver):
+    fake_nodes_module["ComboBK"] = _make_combo_bykey_node()
+    prompt = {
+        "img": {"class_type": "ComboBK", "inputs": {"mode": {"mode": "image", "img": None}}},
+        "lat": {"class_type": "ComboBK", "inputs": {"mode": {"mode": "latent", "lat": None}}},
+    }
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("img", 0) == "IMAGE"
+    assert r.resolve_output_type("img", 1) == "MASK"
+    assert r.resolve_output_type("lat", 0) == "LATENT"
+    assert r.finalized_output_count("img") == 2
+    assert r.finalized_output_count("lat") == 1
+
+
+def test_slot_byslot_resolver_picks_by_resolved_type(fake_nodes_module, TypeResolver):
+    fake_nodes_module["SlotBS"] = _make_slot_byslot_node()
+    fake_nodes_module["ImageSrc"] = _v1_node(("IMAGE",))
+    fake_nodes_module["LatentSrc"] = _v1_node(("LATENT",))
+    prompt = {
+        "img_src": {"class_type": "ImageSrc", "inputs": {}},
+        "lat_src": {"class_type": "LatentSrc", "inputs": {}},
+        "image_consumer": {"class_type": "SlotBS", "inputs": {"slot": ["img_src", 0]}},
+        "latent_consumer": {"class_type": "SlotBS", "inputs": {"slot": ["lat_src", 0]}},
+        "unconnected": {"class_type": "SlotBS", "inputs": {}},
+    }
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("image_consumer", 0) == "IMAGE"
+    assert r.resolve_output_type("image_consumer", 1) == "MASK"
+    assert r.resolve_output_type("latent_consumer", 0) == "LATENT"
+    # Unconnected → when=None branch declares outputs=[]
+    assert r.finalized_output_count("unconnected") == 0
+
+
+def test_bare_execution_blocker_sized_to_finalized_outputs():
+    """The non-NodeOutput path (bare ``ExecutionBlocker`` from V1-style returns)
+    also sizes against the finalized list."""
+    from comfy_api.latest import _io as io
+    from comfy_execution.graph_utils import ExecutionBlocker
+    from execution import get_output_from_returns
+
+    finalized = _dyn_finalized([io.Image.Output("a"), io.Mask.Output("b"), io.Latent.Output("c")])
+
+    class _Obj:
+        RETURN_TYPES = ()
+
+    blocker = ExecutionBlocker("stopped")
+    output, _ui, has_subgraph = get_output_from_returns([blocker], _Obj(), finalized_outputs=finalized)
+    assert has_subgraph is False
+    assert len(output) == 3
+    for slot in output:
+        assert slot[0] is blocker

tests-unit/execution_test/test_type_resolver.py

@@ -0,0 +1,482 @@
+"""Unit tests for :mod:`comfy_execution.type_resolver`.
+
+These tests stand up a small in-memory ``NODE_CLASS_MAPPINGS`` for the test
+node classes (V1 and V3) and a fake DynamicPrompt-like dict, then verify the
+resolver's behaviour for:
+
+  * Static V1 ``RETURN_TYPES`` resolution.
+  * V1 wildcard outputs (must yield ``AnyType`` and warn once).
+  * V3 ``MatchType`` chains resolved via the downstream node's bound inputs.
+  * ``MatchType`` with no upstream bound (fall back to ``AnyType`` + warn).
+  * ``MatchType`` cycles (termination at ``AnyType`` + warn, no recursion blow-up).
+  * Deep chains capped by ``MAX_RESOLVE_DEPTH``.
+  * Input-type resolution for both literal values and links.
+  * Effective slot io_type peeling for ``Autogrow`` (returns the wrapped type).
+  * ``compute_live_input_types`` produces the right shape.
+  * Cache invalidation.
+
+The tests deliberately patch ``nodes.NODE_CLASS_MAPPINGS`` so they don't need
+the whole ComfyUI bootstrap.
+"""
+
+from __future__ import annotations
+
+import logging
+import sys
+import types as _pytypes
+
+import pytest
+
+
+# ---------------------------------------------------------------------------
+# Lightweight V1 test node factory
+# ---------------------------------------------------------------------------
+
+def _v1_node(return_types: tuple[str, ...], input_types_dict: dict | None = None,
+             output_is_list: tuple[bool, ...] | None = None):
+    """Build a V1 node class with the given RETURN_TYPES / INPUT_TYPES()."""
+    if input_types_dict is None:
+        input_types_dict = {"required": {}}
+
+    class _V1:
+        RETURN_TYPES = return_types
+        if output_is_list is not None:
+            OUTPUT_IS_LIST = output_is_list
+
+        @classmethod
+        def INPUT_TYPES(cls):
+            return input_types_dict
+
+    return _V1
+
+
+# ---------------------------------------------------------------------------
+# Fixture: install fake nodes module before importing the resolver
+# ---------------------------------------------------------------------------
+
+@pytest.fixture
+def fake_nodes_module():
+    """Install a synthetic ``nodes`` module with an empty mappings dict.
+
+    Yields the mappings dict so tests can populate it per case. Cleans up
+    afterwards. We also have to make sure comfy_execution.type_resolver picks
+    up our fake module on its local import.
+    """
+    real_nodes = sys.modules.get("nodes")
+    fake = _pytypes.ModuleType("nodes")
+    fake.NODE_CLASS_MAPPINGS = {}
+    sys.modules["nodes"] = fake
+    try:
+        yield fake.NODE_CLASS_MAPPINGS
+    finally:
+        if real_nodes is not None:
+            sys.modules["nodes"] = real_nodes
+        else:
+            del sys.modules["nodes"]
+
+
+@pytest.fixture
+def TypeResolver(fake_nodes_module):
+    # Late import so it picks up our fake `nodes` module.
+    from comfy_execution.type_resolver import TypeResolver as TR
+    return TR
+
+
+# ---------------------------------------------------------------------------
+# V1 resolution
+# ---------------------------------------------------------------------------
+
+def test_v1_static_return_types_resolves(fake_nodes_module, TypeResolver):
+    fake_nodes_module["AddNode"] = _v1_node(("INT",))
+    prompt = {"n1": {"class_type": "AddNode", "inputs": {}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("n1", 0) == "INT"
+
+
+def test_v1_wildcard_warns_once_and_returns_any(fake_nodes_module, TypeResolver, caplog):
+    fake_nodes_module["WildNode"] = _v1_node(("*",))
+    prompt = {"n1": {"class_type": "WildNode", "inputs": {}}}
+    r = TypeResolver(prompt)
+    with caplog.at_level(logging.WARNING, logger="root"):
+        assert r.resolve_output_type("n1", 0) == "*"
+        # second call should still return * but not produce a second warning
+        assert r.resolve_output_type("n1", 0) == "*"
+    warnings = [rec for rec in caplog.records if "TypeResolver" in rec.message]
+    assert len(warnings) == 1, f"expected exactly one warning, got {warnings}"
+
+
+def test_unknown_node_returns_any(fake_nodes_module, TypeResolver):
+    prompt = {"n1": {"class_type": "NopeNode", "inputs": {}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("n1", 0) == "*"
+
+
+def test_out_of_range_slot_returns_any(fake_nodes_module, TypeResolver):
+    fake_nodes_module["AddNode"] = _v1_node(("INT",))
+    prompt = {"n1": {"class_type": "AddNode", "inputs": {}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("n1", 5) == "*"
+
+
+def test_missing_node_returns_any(fake_nodes_module, TypeResolver):
+    fake_nodes_module["AddNode"] = _v1_node(("INT",))
+    prompt = {"n1": {"class_type": "AddNode", "inputs": {}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("nonexistent", 0) == "*"
+
+
+# ---------------------------------------------------------------------------
+# is_output_list / is_input_list
+# ---------------------------------------------------------------------------
+
+def test_is_output_list(fake_nodes_module, TypeResolver):
+    fake_nodes_module["ListNode"] = _v1_node(("IMAGE", "MASK"), output_is_list=(True, False))
+    prompt = {"n1": {"class_type": "ListNode", "inputs": {}}}
+    r = TypeResolver(prompt)
+    assert r.is_output_list("n1", 0) is True
+    assert r.is_output_list("n1", 1) is False
+
+
+def test_is_input_list_follows_link(fake_nodes_module, TypeResolver):
+    fake_nodes_module["ListNode"] = _v1_node(("IMAGE",), output_is_list=(True,))
+    fake_nodes_module["Consumer"] = _v1_node(
+        ("INT",),
+        {"required": {"img": ("IMAGE",)}},
+    )
+    prompt = {
+        "src": {"class_type": "ListNode", "inputs": {}},
+        "dst": {"class_type": "Consumer", "inputs": {"img": ["src", 0]}},
+    }
+    r = TypeResolver(prompt)
+    assert r.is_input_list("dst", "img") is True
+
+
+# ---------------------------------------------------------------------------
+# V3 MatchType resolution
+# ---------------------------------------------------------------------------
+
+def _make_switch_node_class():
+    """Build a V3 Switch-like node with MatchType inputs/outputs."""
+    from comfy_api.latest import io
+
+    class Switch(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            template = io.MatchType.Template("switch")
+            return io.Schema(
+                node_id="TestSwitch",
+                inputs=[
+                    io.Boolean.Input("switch"),
+                    io.MatchType.Input("on_false", template=template, optional=True),
+                    io.MatchType.Input("on_true", template=template, optional=True),
+                ],
+                outputs=[io.MatchType.Output(template=template)],
+            )
+
+        @classmethod
+        def execute(cls, switch, on_false=None, on_true=None):
+            return io.NodeOutput(on_true if switch else on_false)
+
+    # Force schema computation so SCHEMA / RETURN_TYPES are populated.
+    Switch.GET_SCHEMA()
+    return Switch
+
+
+def test_matchtype_resolves_to_upstream_concrete(fake_nodes_module, TypeResolver):
+    fake_nodes_module["TestSwitch"] = _make_switch_node_class()
+    fake_nodes_module["ImageSrc"] = _v1_node(("IMAGE",))
+    prompt = {
+        "img": {"class_type": "ImageSrc", "inputs": {}},
+        "sw": {
+            "class_type": "TestSwitch",
+            "inputs": {"switch": True, "on_true": ["img", 0]},
+        },
+    }
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("sw", 0) == "IMAGE"
+
+
+def test_matchtype_first_concrete_wins(fake_nodes_module, TypeResolver):
+    fake_nodes_module["TestSwitch"] = _make_switch_node_class()
+    fake_nodes_module["ImageSrc"] = _v1_node(("IMAGE",))
+    fake_nodes_module["LatentSrc"] = _v1_node(("LATENT",))
+    prompt = {
+        "img": {"class_type": "ImageSrc", "inputs": {}},
+        "lat": {"class_type": "LatentSrc", "inputs": {}},
+        "sw": {
+            "class_type": "TestSwitch",
+            "inputs": {
+                "switch": False,
+                "on_false": ["img", 0],   # listed first in schema → wins
+                "on_true":  ["lat", 0],
+            },
+        },
+    }
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("sw", 0) == "IMAGE"
+
+
+def test_matchtype_no_bound_input_returns_any(fake_nodes_module, TypeResolver, caplog):
+    fake_nodes_module["TestSwitch"] = _make_switch_node_class()
+    prompt = {"sw": {"class_type": "TestSwitch", "inputs": {"switch": True}}}
+    r = TypeResolver(prompt)
+    with caplog.at_level(logging.WARNING, logger="root"):
+        assert r.resolve_output_type("sw", 0) == "*"
+    assert any("MatchType" in rec.message for rec in caplog.records)
+
+
+def test_matchtype_skips_wildcard_input(fake_nodes_module, TypeResolver):
+    """If the first matched input resolves to AnyType, the resolver tries the next."""
+    fake_nodes_module["TestSwitch"] = _make_switch_node_class()
+    fake_nodes_module["WildNode"] = _v1_node(("*",))
+    fake_nodes_module["ImageSrc"] = _v1_node(("IMAGE",))
+    prompt = {
+        "wild": {"class_type": "WildNode", "inputs": {}},
+        "img": {"class_type": "ImageSrc", "inputs": {}},
+        "sw": {
+            "class_type": "TestSwitch",
+            "inputs": {
+                "switch": True,
+                "on_false": ["wild", 0],
+                "on_true":  ["img", 0],
+            },
+        },
+    }
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("sw", 0) == "IMAGE"
+
+
+def test_matchtype_cycle_terminates_at_any(fake_nodes_module, TypeResolver):
+    """Two switches that feed each other must not recurse forever."""
+    fake_nodes_module["TestSwitch"] = _make_switch_node_class()
+    prompt = {
+        "a": {"class_type": "TestSwitch", "inputs": {"switch": True, "on_true": ["b", 0]}},
+        "b": {"class_type": "TestSwitch", "inputs": {"switch": True, "on_true": ["a", 0]}},
+    }
+    r = TypeResolver(prompt)
+    # Must not raise / recurse forever; both resolve to AnyType.
+    assert r.resolve_output_type("a", 0) == "*"
+    assert r.resolve_output_type("b", 0) == "*"
+
+
+def test_matchtype_chain_resolves_through(fake_nodes_module, TypeResolver):
+    """A → B → C → IMAGE: chain must walk all the way."""
+    fake_nodes_module["TestSwitch"] = _make_switch_node_class()
+    fake_nodes_module["ImageSrc"] = _v1_node(("IMAGE",))
+    prompt = {
+        "src": {"class_type": "ImageSrc", "inputs": {}},
+        "a": {"class_type": "TestSwitch", "inputs": {"switch": True, "on_true": ["src", 0]}},
+        "b": {"class_type": "TestSwitch", "inputs": {"switch": True, "on_true": ["a", 0]}},
+        "c": {"class_type": "TestSwitch", "inputs": {"switch": True, "on_true": ["b", 0]}},
+    }
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("c", 0) == "IMAGE"
+
+
+# ---------------------------------------------------------------------------
+# Input resolution and effective io_type peeling
+# ---------------------------------------------------------------------------
+
+def test_resolve_input_type_literal_uses_declared(fake_nodes_module, TypeResolver):
+    fake_nodes_module["Sink"] = _v1_node(("INT",), {"required": {"steps": ("INT",)}})
+    prompt = {"n1": {"class_type": "Sink", "inputs": {"steps": 20}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_input_type("n1", "steps") == "INT"
+
+
+def test_resolve_input_type_link(fake_nodes_module, TypeResolver):
+    fake_nodes_module["Src"] = _v1_node(("LATENT",))
+    fake_nodes_module["Sink"] = _v1_node(("INT",), {"required": {"x": ("*",)}})
+    prompt = {
+        "src": {"class_type": "Src", "inputs": {}},
+        "sink": {"class_type": "Sink", "inputs": {"x": ["src", 0]}},
+    }
+    r = TypeResolver(prompt)
+    assert r.resolve_input_type("sink", "x") == "LATENT"
+
+
+def test_effective_slot_type_peels_autogrow(fake_nodes_module, TypeResolver):
+    from comfy_api.latest import io
+
+    class AutogrowImg(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            template = io.Autogrow.TemplatePrefix(
+                input=io.Image.Input("img"),
+                prefix="img",
+                min=1,
+            )
+            return io.Schema(
+                node_id="AutogrowImg",
+                inputs=[io.Autogrow.Input("imgs", template=template)],
+                outputs=[io.Image.Output()],
+            )
+
+        @classmethod
+        def execute(cls, imgs):
+            return io.NodeOutput(None)
+
+    AutogrowImg.GET_SCHEMA()
+    fake_nodes_module["AutogrowImg"] = AutogrowImg
+    prompt = {"n1": {"class_type": "AutogrowImg", "inputs": {}}}
+    r = TypeResolver(prompt)
+    # The user-facing element type, not the autogrow wrapper.
+    assert r.get_declared_slot_io_type("n1", "imgs") == "IMAGE"
+
+
+# ---------------------------------------------------------------------------
+# compute_live_input_types
+# ---------------------------------------------------------------------------
+
+def test_effective_slot_type_on_v3_plain_input(fake_nodes_module, TypeResolver):
+    """V3 input that is neither Autogrow/DynamicSlot/DynamicCombo must still resolve.
+
+    Regression test: importing ``io`` from the public re-export skipped
+    ``DynamicSlot``, so an ``isinstance`` chain in ``_effective_io_type`` raised
+    ``AttributeError`` the first time it ran against a plain V3 input.
+    """
+    from comfy_api.latest import io
+
+    class BoolSink(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(
+                node_id="BoolSink",
+                inputs=[io.Boolean.Input("flag")],
+                outputs=[io.Boolean.Output()],
+            )
+
+        @classmethod
+        def execute(cls, flag):
+            return io.NodeOutput(flag)
+
+    BoolSink.GET_SCHEMA()
+    fake_nodes_module["BoolSink"] = BoolSink
+    prompt = {"n": {"class_type": "BoolSink", "inputs": {"flag": True}}}
+    r = TypeResolver(prompt)
+    assert r.get_declared_slot_io_type("n", "flag") == "BOOLEAN"
+    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(
+        ("INT",),
+        {"required": {"model": ("MODEL",), "steps": ("INT",)}},
+    )
+    prompt = {
+        "src": {"class_type": "Src", "inputs": {}},
+        "sink": {
+            "class_type": "Sink",
+            "inputs": {"model": ["src", 0], "steps": 20},
+        },
+    }
+    r = TypeResolver(prompt)
+    assert r.compute_live_input_types("sink") == {"model": "MODEL", "steps": "INT"}
+
+
+# ---------------------------------------------------------------------------
+# Cache invalidation
+# ---------------------------------------------------------------------------
+
+def test_invalidate_clears_cache(fake_nodes_module, TypeResolver):
+    fake_nodes_module["Src"] = _v1_node(("IMAGE",))
+    prompt = {"n1": {"class_type": "Src", "inputs": {}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("n1", 0) == "IMAGE"
+    # Mutate the underlying class and invalidate; the resolver must re-read.
+    fake_nodes_module["Src"] = _v1_node(("LATENT",))
+    r.invalidate()
+    assert r.resolve_output_type("n1", 0) == "LATENT"
+
+
+# ---------------------------------------------------------------------------
+# Malformed input robustness
+# ---------------------------------------------------------------------------
+
+def test_malformed_link_does_not_crash(fake_nodes_module, TypeResolver):
+    """A link with a non-int slot index must not raise; resolver returns AnyType."""
+    fake_nodes_module["Src"] = _v1_node(("IMAGE",))
+    fake_nodes_module["Sink"] = _v1_node(("INT",), {"required": {"x": ("*",)}})
+    prompt = {
+        "src": {"class_type": "Src", "inputs": {}},
+        # slot index sent as a string (common API JSON mistake)
+        "sink": {"class_type": "Sink", "inputs": {"x": ["src", "0"]}},
+    }
+    r = TypeResolver(prompt)
+    # Falls back to declared slot type (still "*"), no exception.
+    assert r.resolve_input_type("sink", "x") == "*"
+
+
+def test_malformed_link_wrong_arity_does_not_crash(fake_nodes_module, TypeResolver):
+    fake_nodes_module["Src"] = _v1_node(("IMAGE",))
+    fake_nodes_module["Sink"] = _v1_node(("INT",), {"required": {"x": ("*",)}})
+    prompt = {
+        "src": {"class_type": "Src", "inputs": {}},
+        "sink": {"class_type": "Sink", "inputs": {"x": ["src"]}},  # arity 1
+    }
+    r = TypeResolver(prompt)
+    assert r.resolve_input_type("sink", "x") == "*"
+
+
+def test_direct_resolve_output_type_with_bad_slot_idx_returns_any(fake_nodes_module, TypeResolver):
+    fake_nodes_module["Src"] = _v1_node(("IMAGE",))
+    prompt = {"src": {"class_type": "Src", "inputs": {}}}
+    r = TypeResolver(prompt)
+    # type-wise these should be unreachable through normal validation but the
+    # resolver must still degrade gracefully.
+    assert r.resolve_output_type("src", "0") == "*"
+    assert r.resolve_output_type("src", True) == "*"  # bool is a subclass of int
+    assert r.is_output_list("src", "0") is False
+
+
+def test_non_string_class_type_returns_any(fake_nodes_module, TypeResolver):
+    prompt = {"n1": {"class_type": 42, "inputs": {}}}
+    r = TypeResolver(prompt)
+    assert r.resolve_output_type("n1", 0) == "*"
+
+
+def test_invalidate_node_only_clears_that_node(fake_nodes_module, TypeResolver):
+    fake_nodes_module["SrcA"] = _v1_node(("IMAGE",))
+    fake_nodes_module["SrcB"] = _v1_node(("LATENT",))
+    prompt = {
+        "a": {"class_type": "SrcA", "inputs": {}},
+        "b": {"class_type": "SrcB", "inputs": {}},
+    }
+    r = TypeResolver(prompt)
+    r.resolve_output_type("a", 0)
+    r.resolve_output_type("b", 0)
+    fake_nodes_module["SrcA"] = _v1_node(("MASK",))
+    r.invalidate_node("a")
+    assert r.resolve_output_type("a", 0) == "MASK"
+    # b's cached result survives even though SrcB was unchanged
+    assert ("b", 0) in r._output_cache