Fix dynamic input validation and lazy scheduling

comfy_execution/graph.py

@@ -5,6 +5,8 @@ import asyncio
 import inspect
 from comfy_execution.graph_utils import is_link, ExecutionBlocker
 from comfy.comfy_types.node_typing import ComfyNodeABC, InputTypeDict, InputTypeOptions
+from comfy_api.internal import _ComfyNodeInternal
+from comfy_api.latest import _io
 
 # NOTE: ExecutionBlocker code got moved to graph_utils.py to prevent torch being imported too soon during unit tests
 ExecutionBlocker = ExecutionBlocker
@@ -123,9 +125,12 @@ class TopologicalSort:
         self.unblockedEvent = asyncio.Event()
 
     def get_input_info(self, unique_id, input_name):
-        class_type = self.dynprompt.get_node(unique_id)["class_type"]
-        class_def = nodes.NODE_CLASS_MAPPINGS[class_type]
-        return get_input_info(class_def, input_name)
+        node = self.dynprompt.get_node(unique_id)
+        class_def = nodes.NODE_CLASS_MAPPINGS[node["class_type"]]
+        valid_inputs = class_def.INPUT_TYPES()
+        if issubclass(class_def, _ComfyNodeInternal):
+            valid_inputs, _, _ = _io.get_finalized_class_inputs(valid_inputs, node["inputs"])
+        return get_input_info(class_def, input_name, valid_inputs)
 
     def make_input_strong_link(self, to_node_id, to_input):
         inputs = self.dynprompt.get_node(to_node_id)["inputs"]

execution.py

@@ -1081,29 +1081,35 @@ async def validate_inputs(prompt_id, prompt, item, validated, visiting=None):
 
     if len(validate_function_inputs) > 0 or validate_has_kwargs:
         input_data_all, _, v3_data = get_input_data(inputs, obj_class, unique_id)
+        dynamic_paths = v3_data.get("dynamic_paths", {})
+        if not validate_has_kwargs:
+            dynamic_paths = {key: path for key, path in dynamic_paths.items() if path.split(".")[0] in validate_function_inputs}
+            v3_data = v3_data | {
+                "dynamic_paths": dynamic_paths,
+                "list_paths": {path for path in v3_data.get("list_paths", set()) if path.split(".")[0] in validate_function_inputs},
+            }
         input_filtered = {}
         for x in input_data_all:
-            if x in validate_function_inputs or validate_has_kwargs:
+            input_name = dynamic_paths[x].split(".")[0] if x in dynamic_paths else x
+            if input_name in validate_function_inputs or validate_has_kwargs:
                 input_filtered[x] = input_data_all[x]
         if 'input_types' in validate_function_inputs:
             input_filtered['input_types'] = [received_types]
 
         ret = await _async_map_node_over_list(prompt_id, unique_id, obj_class, input_filtered, validate_function_name, v3_data=v3_data)
         ret = await resolve_map_node_over_list_results(ret)
-        for x in input_filtered:
+        for x in input_filtered or (None,):
             for i, r in enumerate(ret):
                 if r is not True and not isinstance(r, ExecutionBlocker):
-                    details = f"{x}"
+                    details = x if x is not None else ""
                     if r is not False:
-                        details += f" - {str(r)}"
+                        details = f"{x} - {r}" if x is not None else str(r)
 
                     error = {
                         "type": "custom_validation_failed",
                         "message": "Custom validation failed for node",
                         "details": details,
-                        "extra_info": {
-                            "input_name": x,
-                        }
+                        "extra_info": {"input_name": x} if x is not None else {},
                     }
                     errors.append(error)
                     continue

tests-unit/execution_test/dynamic_inputs_test.py

@@ -0,0 +1,217 @@
+from types import SimpleNamespace
+from unittest.mock import Mock
+
+import pytest
+import torch
+
+from comfy.cli_args import args
+
+if not torch.cuda.is_available():
+    args.cpu = True
+
+import execution
+import nodes
+from comfy_api.latest import io
+
+
+pytestmark = pytest.mark.asyncio
+
+
+@pytest.mark.parametrize("input_spec,values,expected", [
+    pytest.param(
+        io.Autogrow.Input("rows", template=io.Autogrow.TemplatePrefix(io.Float.Input("x"), prefix="item", min=0)),
+        {"rows.item0": 0.5}, {"item0": 0.5}, id="autogrow",
+    ),
+    pytest.param(
+        io.Autogrow.Input("rows", template=io.Autogrow.TemplatePrefix(io.Float.Input("x"), prefix="item", min=0)),
+        {}, {}, id="empty-autogrow",
+    ),
+    pytest.param(
+        io.DynamicCombo.Input("rows", options=[io.DynamicCombo.Option("on", [io.Float.Input("x")])]),
+        {"rows": "on", "rows.x": 0.5}, {"rows": "on", "x": 0.5}, id="combo",
+    ),
+    pytest.param(
+        io.DynamicCombo.Input("rows", options=[io.DynamicCombo.Option("on", [
+            io.Autogrow.Input("items", template=io.Autogrow.TemplatePrefix(io.Float.Input("x"), prefix="item", min=0)),
+        ])]),
+        {"rows": "on", "rows.items.item0": 0.5}, {"rows": "on", "items": {"item0": 0.5}}, id="nested-autogrow",
+    ),
+])
+@pytest.mark.parametrize("validation_result", [True, False, "Rows rejected"])
+async def test_custom_validation_receives_only_requested_roots(monkeypatch, input_spec, values, expected, validation_result):
+    received = []
+
+    class Validator(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(node_id=cls.__name__, inputs=[
+                input_spec,
+                io.Float.Input("ignored"),
+            ], outputs=[], is_output_node=True)
+
+        @classmethod
+        def validate_inputs(cls, rows):
+            received.append(rows)
+            return validation_result
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput()
+
+    monkeypatch.setitem(nodes.NODE_CLASS_MAPPINGS, "Validator", Validator)
+    prompt = {"probe": {"class_type": "Validator", "inputs": {"ignored": 4.0, **values}}}
+    valid, _, _, errors = await execution.validate_prompt("validation", prompt, None)
+
+    assert received == [expected]
+    assert valid is (validation_result is True)
+    if validation_result is True:
+        assert errors == {}
+    else:
+        assert {error["type"] for error in errors["probe"]["errors"]} == {"custom_validation_failed"}
+        assert [error["extra_info"] for error in errors["probe"]["errors"]] == (
+            [{"input_name": key} for key in values] or [{}]
+        )
+
+
+async def test_validation_omits_unrequested_dynamic_roots(monkeypatch):
+    received = []
+
+    class Validator(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(node_id=cls.__name__, inputs=[
+                io.Float.Input("fixed"),
+                io.Autogrow.Input("rows", template=io.Autogrow.TemplatePrefix(io.Float.Input("x"), prefix="item", min=0)),
+            ], outputs=[], is_output_node=True)
+
+        @classmethod
+        def validate_inputs(cls, fixed):
+            received.append(fixed)
+            return True
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput()
+
+    monkeypatch.setitem(nodes.NODE_CLASS_MAPPINGS, "Validator", Validator)
+    prompt = {"probe": {"class_type": "Validator", "inputs": {"fixed": 0.5, "rows.item0": 9.0}}}
+    valid, _, _, errors = await execution.validate_prompt("static", prompt, None)
+
+    assert valid, errors
+    assert received == [0.5]
+
+
+async def test_kwargs_validator_retains_dynamic_and_static_inputs(monkeypatch):
+    received = []
+
+    class Validator(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(node_id=cls.__name__, inputs=[
+                io.Autogrow.Input("rows", template=io.Autogrow.TemplatePrefix(io.Float.Input("x"), prefix="item", min=0)),
+                io.Float.Input("fixed"),
+            ], outputs=[], is_output_node=True)
+
+        @classmethod
+        def validate_inputs(cls, **kwargs):
+            received.append(kwargs)
+            return True
+
+        @classmethod
+        def execute(cls, **kwargs):
+            return io.NodeOutput()
+
+    monkeypatch.setitem(nodes.NODE_CLASS_MAPPINGS, "Validator", Validator)
+    prompt = {"probe": {"class_type": "Validator", "inputs": {"rows.item0": 0.5, "fixed": 2.0}}}
+    valid, _, _, errors = await execution.validate_prompt("kwargs", prompt, None)
+
+    assert valid, errors
+    assert received == [{"rows": {"item0": 0.5}, "fixed": 2.0}]
+
+
+@pytest.mark.parametrize("values", [{}, {"value": 0.5}])
+async def test_legacy_validation_rejection_is_recorded_even_without_inputs(monkeypatch, values):
+    class Validator:
+        @classmethod
+        def INPUT_TYPES(cls):
+            return {"optional": {"value": ("FLOAT", {})}}
+
+        @classmethod
+        def VALIDATE_INPUTS(cls, **kwargs):
+            return "Rejected"
+
+        RETURN_TYPES = ()
+        OUTPUT_NODE = True
+
+    monkeypatch.setitem(nodes.NODE_CLASS_MAPPINGS, "Validator", Validator)
+    prompt = {"probe": {"class_type": "Validator", "inputs": values}}
+    valid, _, _, errors = await execution.validate_prompt("legacy", prompt, None)
+
+    assert not valid
+    assert errors["probe"]["errors"] == [{
+        "type": "custom_validation_failed",
+        "message": "Custom validation failed for node",
+        "details": "value - Rejected" if values else "Rejected",
+        "extra_info": {"input_name": "value"} if values else {},
+    }]
+
+
+@pytest.mark.parametrize("kind", ["combo", "autogrow"])
+@pytest.mark.parametrize("lazy,request_first", [(True, False), (True, True), (False, False)])
+async def test_dynamic_lazy_inputs_execute_only_when_requested(monkeypatch, kind, lazy, request_first):
+    executed = []
+    if kind == "combo":
+        input_spec = io.DynamicCombo.Input("rows", options=[io.DynamicCombo.Option("on", [
+            io.Float.Input("item0", lazy=lazy), io.Float.Input("item1", lazy=lazy),
+        ])])
+    else:
+        input_spec = io.Autogrow.Input("rows", template=io.Autogrow.TemplatePrefix(io.Float.Input("x", lazy=lazy), prefix="item"))
+    inputs = {"rows.item0": ["first", 0], "rows.item1": ["second", 0]}
+    if kind == "combo":
+        inputs["rows"] = "on"
+
+    class Source(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(node_id=cls.__name__, inputs=[io.Float.Input("value")], outputs=[io.Float.Output()])
+
+        @classmethod
+        def execute(cls, value):
+            executed.append(value)
+            return io.NodeOutput(value)
+
+    class Consumer(io.ComfyNode):
+        @classmethod
+        def define_schema(cls):
+            return io.Schema(node_id=cls.__name__, inputs=[input_spec], outputs=[], is_output_node=True)
+
+        @classmethod
+        def check_lazy_status(cls, rows):
+            value, key = rows["item0"]
+            return [key] if request_first and value is None else []
+
+        @classmethod
+        def execute(cls, rows):
+            return io.NodeOutput(ui={"rows": [rows]})
+
+    monkeypatch.setitem(nodes.NODE_CLASS_MAPPINGS, "Source", Source)
+    monkeypatch.setitem(nodes.NODE_CLASS_MAPPINGS, "Consumer", Consumer)
+    prompt = {
+        "first": {"class_type": "Source", "inputs": {"value": 2.0}},
+        "second": {"class_type": "Source", "inputs": {"value": 3.0}},
+        "consumer": {"class_type": "Consumer", "inputs": inputs},
+    }
+    valid, _, outputs, errors = await execution.validate_prompt("lazy", prompt, None)
+    assert valid, errors
+    server = SimpleNamespace(client_id=None, send_sync=Mock())
+    executor = execution.PromptExecutor(server, cache_type=execution.CacheType.NONE, cache_args={"ram": 0, "ram_inactive": 0})
+    await executor.execute_async(prompt, "lazy", execute_outputs=outputs)
+
+    assert executor.success, executor.status_messages
+    first = 2.0 if request_first or not lazy else None
+    second = 3.0 if not lazy else None
+    assert sorted(executed) == [value for value in [first, second] if value is not None]
+    expected = {"item0": first, "item1": second}
+    if kind == "combo":
+        expected["rows"] = "on"
+    assert executor.history_result["outputs"]["consumer"]["rows"] == [expected]