TypeResolver: address code review (link parsing + slot_idx guard + back-compat shim)

* Add _parse_link helper validating both node_id (str) and slot_idx (int,
  rejecting bool) so malformed API JSON (e.g. ['n1', '0']) degrades to
  AnyType instead of crashing with TypeError.
* Add slot_idx type guards in resolve_output_type and is_output_list.
* Extract _get_class_def_for_node helper to dedupe node/class lookup
  across resolve_output_type, is_output_list, get_declared_slot_io_type.
* register_dynamic_input_func now detects 5-argument legacy callables
  via inspect.signature and silently wraps them; preserves backward
  compatibility for any custom node that registered its own dynamic
  input expansion against the pre-live_input_types signature.
* Tests: malformed link (str slot idx, wrong arity), bad slot type
  directly to resolve_output_type, non-string class_type. Tests for
  the legacy 5-arg shim and the modern 6-arg passthrough, including
  callables with uninspectable signatures.

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

tests-unit/comfy_api_test/test_register_dynamic_input_func.py

@@ -0,0 +1,79 @@
+"""Backward-compat tests for ``register_dynamic_input_func``.
+
+When ``live_input_types`` was added as a sixth argument to the dynamic-input
+expansion callback, third-party custom nodes that registered against the
+original 5-argument signature would otherwise crash with ``TypeError`` the
+first time their input was expanded. ``register_dynamic_input_func`` wraps
+such legacy callables transparently.
+"""
+
+from __future__ import annotations
+
+from comfy_api.latest import _io
+
+
+def test_legacy_5arg_callback_is_wrapped_transparently():
+    received = {}
+
+    def legacy(out_dict, live_inputs, value, input_type, curr_prefix):
+        received["args"] = (out_dict, live_inputs, value, input_type, curr_prefix)
+
+    io_type = "TEST_LEGACY_5ARG_V3"
+    try:
+        _io.register_dynamic_input_func(io_type, legacy)
+        fn = _io.get_dynamic_input_func(io_type)
+
+        # Caller invokes with 6 arguments (current signature). The shim must
+        # strip the trailing live_input_types argument before delegating.
+        fn({"required": {}}, {"a": 1}, ("X", {}), "required", ["p"], {"a": "INT"})
+
+        assert received["args"] == (
+            {"required": {}}, {"a": 1}, ("X", {}), "required", ["p"]
+        )
+    finally:
+        _io.DYNAMIC_INPUT_LOOKUP.pop(io_type, None)
+
+
+def test_new_6arg_callback_passes_live_input_types_through():
+    received = {}
+
+    def modern(out_dict, live_inputs, value, input_type, curr_prefix, live_input_types=None):
+        received["live_input_types"] = live_input_types
+
+    io_type = "TEST_MODERN_6ARG_V3"
+    try:
+        _io.register_dynamic_input_func(io_type, modern)
+        fn = _io.get_dynamic_input_func(io_type)
+        fn({}, {}, ("X", {}), "required", None, {"foo": "IMAGE"})
+        assert received["live_input_types"] == {"foo": "IMAGE"}
+    finally:
+        _io.DYNAMIC_INPUT_LOOKUP.pop(io_type, None)
+
+
+def test_callable_with_uninspectable_signature_assumed_modern():
+    """``functools.partial`` and C builtins may have no introspectable signature.
+
+    The shim must not blow up; falling back to the new signature is the safe
+    choice (we get a clean TypeError if the callable really is too old, which
+    is no worse than the pre-shim behavior).
+    """
+    calls = []
+
+    class _CallableObj:
+        # Lambdas / objects with __call__ are introspectable; partial with
+        # opaque builtins are not. Simulate the latter by raising in __signature__.
+        def __call__(self, *args, **kwargs):
+            calls.append((args, kwargs))
+
+        @property
+        def __signature__(self):
+            raise ValueError("uninspectable")
+
+    io_type = "TEST_UNINSPECTABLE_V3"
+    try:
+        _io.register_dynamic_input_func(io_type, _CallableObj())
+        fn = _io.get_dynamic_input_func(io_type)
+        fn({}, {}, ("X", {}), "required", None, {"x": "INT"})
+        assert calls and len(calls[0][0]) == 6
+    finally:
+        _io.DYNAMIC_INPUT_LOOKUP.pop(io_type, None)

tests-unit/execution_test/test_type_resolver.py

@@ -360,6 +360,52 @@ def test_invalidate_clears_cache(fake_nodes_module, TypeResolver):
     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",))