From 188065c0117a51f03580f3e789ebe2421b24f74f Mon Sep 17 00:00:00 2001 From: Jedrzej Kosinski Date: Mon, 1 Jun 2026 20:48:08 -0700 Subject: [PATCH] DynamicOutputs: fix block_execution sizing, harden validation, trim comments Address code-review feedback: * Fix pre-existing bug surfaced by this work: a V3 node returning a bare ExecutionBlocker (normalized to NodeOutput(block_execution=msg) with no positional/named result) was silently dropped in get_output_from_returns. block_execution now takes precedence and shapes the blocker tuple to the active output count first. * Reject malformed linked-input source node ids (non-string) up front in validate_inputs as 'bad_linked_input' instead of letting them fall into the type resolver. * Schema.validate(): reject duplicate DynamicOutputs group ids. * Drop a couple of over-defensive / restating comments. New tests: blocker sizing through both the NodeOutput and bare-blocker paths for dynamic nodes, and duplicate-group-id schema rejection. Amp-Thread-ID: https://ampcode.com/threads/T-019e8568-f382-743d-a97f-0de3ff29d501 Co-authored-by: Amp --- comfy_api/latest/_io.py | 20 +++++-- comfy_execution/type_resolver.py | 6 +- execution.py | 38 ++++++++----- .../comfy_api_test/test_dynamic_outputs.py | 27 +++++++++ .../test_dynamic_outputs_resolver.py | 57 +++++++++++++++++++ 5 files changed, 124 insertions(+), 24 deletions(-) diff --git a/comfy_api/latest/_io.py b/comfy_api/latest/_io.py index 80659ec5c..a92480158 100644 --- a/comfy_api/latest/_io.py +++ b/comfy_api/latest/_io.py @@ -1966,13 +1966,21 @@ class Schema: 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.") - # verify dynamic-output groups point at real inputs + # verify dynamic-output groups: unique group ids + selectors point at real inputs + group_ids: list[str] = [] for o in self.outputs: - if isinstance(o, DynamicOutputs.ByKey) and o.selector not in input_set: - issues.append( - f"DynamicOutputs.ByKey(id={o.id!r}) selector input {o.selector!r} " - f"does not exist on the schema." - ) + if isinstance(o, DynamicOutputs.ByKey): + group_ids.append(o.id) + if o.selector not in input_set: + issues.append( + f"DynamicOutputs.ByKey(id={o.id!r}) selector input {o.selector!r} " + f"does not exist on the schema." + ) + 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 diff --git a/comfy_execution/type_resolver.py b/comfy_execution/type_resolver.py index 6352815a2..508e96c4e 100644 --- a/comfy_execution/type_resolver.py +++ b/comfy_execution/type_resolver.py @@ -233,11 +233,7 @@ class TypeResolver: return ANY_TYPE def _get_finalized_outputs(self, node: dict | None, class_def) -> io.FinalizedOutputs | None: - """Return ``FinalizedOutputs`` for V3 nodes with DynamicOutputs groups, else ``None``. - - ``None`` means "use the class-level static arrays" (V1 nodes or V3 - without any dynamic group), keeping the hot path zero-cost. - """ + """Return ``FinalizedOutputs`` for V3 nodes with DynamicOutputs groups, else ``None``.""" if not (isinstance(class_def, type) and issubclass(class_def, _ComfyNodeInternal)): return None try: diff --git a/execution.py b/execution.py index ca3fa272f..57de84327 100644 --- a/execution.py +++ b/execution.py @@ -420,20 +420,26 @@ def get_output_from_returns(return_values, obj, finalized_outputs=None): # downstream code treats this as a fixed-shape tuple. named_result = ( _normalize_named_result(r, finalized_outputs) - if getattr(r, "named", None) is not None + if r.named is not None else None ) - if r.expand is not 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 = named_result if named_result is not None else r.result - if r.block_execution is not None: - result = tuple([ExecutionBlocker(r.block_execution)] * expected_count) - subgraph_results.append((new_graph, 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 - if r.block_execution is not None: - result = tuple([ExecutionBlocker(r.block_execution)] * expected_count) results.append(result) subgraph_results.append((None, result)) else: @@ -493,9 +499,7 @@ async def execute(server, dynprompt, caches, current_item, extra_data, executed, execution_list.cache_update(unique_id, cached) return (ExecutionResult.SUCCESS, None, None) - # Finalize the active output list for this prompt (no-op for static V3 / V1 - # nodes). Computed once per execute() call so the three output-shaping paths - # below — initial, pending async resume, pending subgraph resume — all agree. + # 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() @@ -979,9 +983,17 @@ async def validate_inputs(prompt_id, prompt, item, validated, visiting=None, typ continue o_id = val[0] + 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 list (e.g. stale link after a DynamicOutputs branch - # change). Reports the active count for clearer diagnostics. + # 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 = { diff --git a/tests-unit/comfy_api_test/test_dynamic_outputs.py b/tests-unit/comfy_api_test/test_dynamic_outputs.py index b54bf2131..f0270ad48 100644 --- a/tests-unit/comfy_api_test/test_dynamic_outputs.py +++ b/tests-unit/comfy_api_test/test_dynamic_outputs.py @@ -205,3 +205,30 @@ def test_nodeoutput_from_named_stores_dict(): 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() diff --git a/tests-unit/execution_test/test_dynamic_outputs_resolver.py b/tests-unit/execution_test/test_dynamic_outputs_resolver.py index 45501394a..d99fc5d82 100644 --- a/tests-unit/execution_test/test_dynamic_outputs_resolver.py +++ b/tests-unit/execution_test/test_dynamic_outputs_resolver.py @@ -216,3 +216,60 @@ def test_normalize_named_result_requires_dynamic_node(): 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" + + +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