fix: preflight missing remote inputs

README.md

@@ -158,6 +158,7 @@ When the remote PromptExecutor still fails, the Modal worker now logs the normal
 If ComfyUI's remote cache setup would fail because a prompt references a node class missing from the Modal worker registry, the worker now retries external custom-node import once for the synced bundle and then reports the missing class names and candidate files found in the extracted bundle instead of surfacing a raw `KeyError`.
 Remote queue rewrite now also rejects remote execution when the sync engine is still using the local mirror backend, because those `/custom_nodes/...` and `/assets/...` paths are not visible inside Modal workers.
 Modal volume uploads are idempotent for content-addressed paths: if Modal reports that an uploaded asset or custom-node archive already exists, the sync engine treats that as a successful reuse.
+The remote worker preflights declared required inputs before invoking `PromptExecutor`, so a malformed remote prompt now reports missing inputs like `latent_image` directly instead of surfacing a later Python method-arity error from the node implementation.
 
 ```bash
 export COMFY_MODAL_EXECUTION_MODE=remote

comfyui_modal_sync_cloud.py

@@ -3090,6 +3090,22 @@ def _node_input_type_map(node_class: type[Any]) -> dict[str, str]:
     return input_type_map
 
 
+def _node_required_input_names(node_class: type[Any]) -> set[str]:
+    """Return the declared required input names for one node class."""
+    input_types_callable = getattr(node_class, "INPUT_TYPES", None)
+    if not callable(input_types_callable):
+        return set()
+
+    raw_input_types = input_types_callable()
+    if not isinstance(raw_input_types, dict):
+        return set()
+
+    required_inputs = raw_input_types.get("required")
+    if not isinstance(required_inputs, dict):
+        return set()
+    return {str(input_name) for input_name in required_inputs}
+
+
 def _coerce_primitive_prompt_input_value(
     *,
     node_id: str,
@@ -3216,6 +3232,27 @@ def _validate_prompt_input_shapes(
                 )
 
 
+def _validate_required_prompt_inputs(
+    prompt: dict[str, Any],
+    node_mapping: dict[str, type[Any]],
+) -> None:
+    """Reject remote prompt nodes that are missing declared required inputs."""
+    for node_id, prompt_node in sorted(prompt.items()):
+        class_type = str(prompt_node.get("class_type"))
+        node_class = node_mapping.get(class_type)
+        if node_class is None:
+            continue
+        inputs = prompt_node.get("inputs") or {}
+        missing_inputs = sorted(_node_required_input_names(node_class) - set(inputs.keys()))
+        if not missing_inputs:
+            continue
+        raise RemoteSubgraphExecutionError(
+            "Remote subgraph prompt is missing required node inputs before execution."
+            f" node_id={node_id!r} node_type={class_type!r}"
+            f" missing_inputs={missing_inputs!r} available_inputs={sorted(str(key) for key in inputs)!r}"
+        )
+
+
 def _log_prompt_executor_failure_details(
     *,
     component_id: str,
@@ -3507,6 +3544,7 @@ def _execute_subgraph_prompt(
         resolved_node_mapping,
         list(normalized_payload.get("boundary_inputs", [])),
     )
+    _validate_required_prompt_inputs(prompt, resolved_node_mapping)
 
     with (
         _temporary_node_mapping(None),

docs/LEARNINGS.md

@@ -225,3 +225,4 @@
 - Stale deployed Modal apps are not limited to missing-app failures. The local client needs an explicit app protocol handshake with `RemoteEngine`, and an unversioned or older remote app should be stopped and replaced before payload execution so local code changes are not silently tested against an old image.
 - Remote execution and asset sync must derive their mode from the same settings snapshot. If sync uses `LocalMirrorVolume` but invocation reads a newer raw `COMFY_MODAL_EXECUTION_MODE=remote` value, the logs misleadingly show custom-node sync success while the Modal worker cannot see the bundle; queue rewrite should fail before dispatch in that split-brain state.
 - Modal volume duplicate-file errors are benign for content-addressed upload paths. If `/custom_nodes/entries/.../<sha>_custom_nodes_bundle.zip` already exists, the digest in the path is the identity check, so the sync engine should mark the upload as present instead of failing the run.
+- ComfyUI's execution-error `current_inputs` display stringifies complex values like `CONDITIONING`, so a quoted tensor dump in the error payload is not proof that transport serialization converted the conditioning to text. If the real failure is a missing sampler input, preflight required node inputs before `PromptExecutor` so the error points at the malformed prompt shape directly.

remote/modal_app.py

@@ -1229,6 +1229,22 @@ def _node_input_type_map(node_class: type[Any]) -> dict[str, str]:
     return input_type_map
 
 
+def _node_required_input_names(node_class: type[Any]) -> set[str]:
+    """Return the declared required input names for one node class."""
+    input_types_callable = getattr(node_class, "INPUT_TYPES", None)
+    if not callable(input_types_callable):
+        return set()
+
+    raw_input_types = input_types_callable()
+    if not isinstance(raw_input_types, dict):
+        return set()
+
+    required_inputs = raw_input_types.get("required")
+    if not isinstance(required_inputs, dict):
+        return set()
+    return {str(input_name) for input_name in required_inputs}
+
+
 def _coerce_primitive_prompt_input_value(
     *,
     node_id: str,
@@ -1355,6 +1371,27 @@ def _validate_prompt_input_shapes(
                 )
 
 
+def _validate_required_prompt_inputs(
+    prompt: dict[str, Any],
+    node_mapping: dict[str, type[Any]],
+) -> None:
+    """Reject remote prompt nodes that are missing declared required inputs."""
+    for node_id, prompt_node in sorted(prompt.items()):
+        class_type = str(prompt_node.get("class_type"))
+        node_class = node_mapping.get(class_type)
+        if node_class is None:
+            continue
+        inputs = prompt_node.get("inputs") or {}
+        missing_inputs = sorted(_node_required_input_names(node_class) - set(inputs.keys()))
+        if not missing_inputs:
+            continue
+        raise RemoteSubgraphExecutionError(
+            "Remote subgraph prompt is missing required node inputs before execution."
+            f" node_id={node_id!r} node_type={class_type!r}"
+            f" missing_inputs={missing_inputs!r} available_inputs={sorted(str(key) for key in inputs)!r}"
+        )
+
+
 def _resolve_required_subgraph_nodes(
     prompt: dict[str, Any],
     execute_node_ids: list[str],
@@ -1601,6 +1638,7 @@ def _execute_subgraph_with_mapping(
         node_mapping,
         list(normalized_payload.get("boundary_inputs", [])),
     )
+    _validate_required_prompt_inputs(prompt, node_mapping)
     required_node_ids = _resolve_required_subgraph_nodes(
         prompt=prompt,
         execute_node_ids=list(normalized_payload.get("execute_node_ids", [])),
@@ -1762,6 +1800,7 @@ def _execute_subgraph_prompt(
         resolved_node_mapping,
         list(normalized_payload.get("boundary_inputs", [])),
     )
+    _validate_required_prompt_inputs(prompt, resolved_node_mapping)
 
     with _temporary_node_mapping(node_mapping):
         executor = execution.PromptExecutor(_NullPromptServer())

tests/test_modal_executor_node.py

@@ -9050,6 +9050,42 @@ def test_apply_boundary_inputs_preserves_singleton_conditioning_lists(
     assert prompt["remote_1"]["inputs"]["value"] == conditioning
 
 
+@pytest.mark.parametrize(
+    ("module_fixture_name",),
+    [
+        ("remote_modal_app_module",),
+        ("modal_cloud_module",),
+    ],
+)
+def test_validate_required_prompt_inputs_reports_missing_latent_image(
+    request: Any,
+    module_fixture_name: str,
+) -> None:
+    """Remote execution should fail before PromptExecutor calls samplers with missing required inputs."""
+    target_module = request.getfixturevalue(module_fixture_name)
+    prompt = {
+        "3": {
+            "class_type": "KSamplerLoraSigmaInverse",
+            "inputs": {
+                "seed": 1,
+                "steps": 24,
+                "positive": [["cond", {"reference_latents": ["reference"]}]],
+            },
+        }
+    }
+
+    with pytest.raises(target_module.RemoteSubgraphExecutionError) as exc_info:
+        target_module._validate_required_prompt_inputs(
+            prompt,
+            {"KSamplerLoraSigmaInverse": _FakeImplicitBatchKSamplerNode},
+        )
+
+    message = str(exc_info.value)
+    assert "missing required node inputs" in message
+    assert "latent_image" in message
+    assert "available_inputs=['positive', 'seed', 'steps']" in message
+
+
 @pytest.mark.parametrize(
     ("module_fixture_name",),
     [

tests/test_serialization.py

@@ -54,6 +54,39 @@ def test_serialize_mapping_supports_nested_tensors(serialization_module: Any) ->
     assert torch.equal(decoded_tensor, tensor)
 
 
+def test_conditioning_with_reference_latents_round_trips(serialization_module: Any) -> None:
+    """CONDITIONING payloads may include reference latent tensors in their metadata."""
+    torch = pytest.importorskip("torch")
+    conditioning = [
+        [
+            torch.arange(12, dtype=torch.float32).reshape(1, 3, 4),
+            {
+                "pooled_output": None,
+                "attention_mask": torch.ones((1, 4), dtype=torch.int64),
+                "reference_latents": [
+                    torch.arange(16, dtype=torch.float16).reshape(1, 4, 2, 2),
+                ],
+            },
+        ]
+    ]
+
+    encoded = serialization_module.serialize_node_inputs({"positive": conditioning})
+    decoded = serialization_module.deserialize_node_inputs(encoded)
+
+    decoded_conditioning = decoded["positive"]
+    assert len(decoded_conditioning) == 1
+    assert torch.equal(decoded_conditioning[0][0], conditioning[0][0])
+    assert decoded_conditioning[0][1]["pooled_output"] is None
+    assert torch.equal(
+        decoded_conditioning[0][1]["attention_mask"],
+        conditioning[0][1]["attention_mask"],
+    )
+    assert torch.equal(
+        decoded_conditioning[0][1]["reference_latents"][0],
+        conditioning[0][1]["reference_latents"][0],
+    )
+
+
 def test_coerce_serialized_node_outputs_accepts_raw_tensor_outputs(serialization_module: Any) -> None:
     """Raw node outputs should be normalized into transport bytes before crossing the wire."""
     torch = pytest.importorskip("torch")