[APO-2731] Filter undefined values from subworkflow inputs

When optional subworkflow inputs have dependencies that resolve to undefined,
the workflow should execute seamlessly using default values instead of raising
INVALID_INPUTS errors. This change filters out undefined values in both
SubworkflowDeploymentNode and InlineSubworkflowNode input compilation methods.

Co-Authored-By: vargas@vellum.ai <vargas@vellum.ai>

Author: devin-ai-integration[bot]

src/vellum/workflows/nodes/core/inline_subworkflow_node/node.py

@@ -152,10 +152,15 @@ def _compile_subworkflow_inputs(self) -> InputsType:
                 inputs_dict = {}
                 for descriptor in inputs_class:
                     if hasattr(self, descriptor.name):
-                        inputs_dict[descriptor.name] = getattr(self, descriptor.name)
+                        value = getattr(self, descriptor.name)
+                        # Filter out undefined values to allow optional inputs to use their defaults
+                        if value is not undefined:
+                            inputs_dict[descriptor.name] = value
                 return inputs_class(**inputs_dict)
             elif isinstance(self.subworkflow_inputs, dict):
-                return inputs_class(**self.subworkflow_inputs)
+                # Filter out undefined values to allow optional inputs to use their defaults
+                filtered_inputs = {k: v for k, v in self.subworkflow_inputs.items() if v is not undefined}
+                return inputs_class(**filtered_inputs)
             elif isinstance(self.subworkflow_inputs, inputs_class):
                 return self.subworkflow_inputs
             else:

src/vellum/workflows/nodes/core/inline_subworkflow_node/tests/test_node.py

@@ -1,5 +1,6 @@
 import pytest
 
+from vellum.workflows.constants import undefined
 from vellum.workflows.errors.types import WorkflowErrorCode
 from vellum.workflows.exceptions import NodeException
 from vellum.workflows.inputs.base import BaseInputs
@@ -178,3 +179,38 @@ class Outputs(BaseWorkflow.Outputs):
 
     assert initiated_events[0].body.workflow_definition.is_dynamic is False  # Main workflow
     assert initiated_events[1].body.workflow_definition.is_dynamic is True  # Inline workflow
+
+
+def test_inline_subworkflow_node__undefined_input_filtered():
+    """Test that undefined input values are filtered out when compiling subworkflow inputs."""
+
+    # GIVEN a subworkflow with a required input and an optional input with a default
+    class SubworkflowInputs(BaseInputs):
+        required_input: str
+        optional_input: str = "default_value"
+
+    class InnerNode(BaseNode):
+        class Outputs(BaseNode.Outputs):
+            result = SubworkflowInputs.required_input
+
+    class TestSubworkflow(BaseWorkflow[SubworkflowInputs, BaseState]):
+        graph = InnerNode
+
+        class Outputs(BaseWorkflow.Outputs):
+            result = InnerNode.Outputs.result
+
+    # AND a node that passes an undefined value for the optional input
+    class TestNode(InlineSubworkflowNode):
+        subworkflow = TestSubworkflow
+        subworkflow_inputs = {
+            "required_input": "hello",
+            "optional_input": undefined,
+        }
+
+    # WHEN we run the node
+    node = TestNode()
+    events = list(node.run())
+
+    # THEN the node should complete successfully using the default value for optional_input
+    assert events[-1].name == "result"
+    assert events[-1].value == "hello"

src/vellum/workflows/nodes/displayable/subworkflow_deployment_node/node.py

@@ -88,9 +88,9 @@ def _compile_subworkflow_inputs(self) -> List[WorkflowRequestInputRequest]:
     def _add_compiled_input(
         self, compiled_inputs: List[WorkflowRequestInputRequest], input_name: str, input_value: Any
     ) -> None:
-        # Exclude inputs that resolved to be null. This ensure that we don't pass input values
+        # Exclude inputs that resolved to be null or undefined. This ensures that we don't pass input values
         # to optional subworkflow inputs whose values were unresolved.
-        if input_value is None:
+        if input_value is None or input_value is undefined:
             return
         if isinstance(input_value, str):
             compiled_inputs.append(
@@ -155,12 +155,12 @@ def _compile_subworkflow_inputs_for_direct_invocation(self, workflow: "BaseWorkf
         if isinstance(self.subworkflow_inputs, BaseInputs):
             inputs_dict = {}
             for input_descriptor, input_value in self.subworkflow_inputs:
-                if input_value is not None:
+                if input_value is not None and input_value is not undefined:
                     inputs_dict[input_descriptor.name] = input_value
             return inputs_class(**inputs_dict)
         else:
-            # Filter out None values for direct invocation
-            filtered_inputs = {k: v for k, v in self.subworkflow_inputs.items() if v is not None}
+            # Filter out None and undefined values for direct invocation
+            filtered_inputs = {k: v for k, v in self.subworkflow_inputs.items() if v is not None and v is not undefined}
             return inputs_class(**filtered_inputs)
 
     def _run_resolved_workflow(

src/vellum/workflows/nodes/displayable/subworkflow_deployment_node/tests/test_node.py

@@ -14,8 +14,10 @@
 from vellum.client.types.workflow_request_chat_history_input_request import WorkflowRequestChatHistoryInputRequest
 from vellum.client.types.workflow_request_json_input_request import WorkflowRequestJsonInputRequest
 from vellum.client.types.workflow_request_number_input_request import WorkflowRequestNumberInputRequest
+from vellum.client.types.workflow_request_string_input_request import WorkflowRequestStringInputRequest
 from vellum.client.types.workflow_result_event import WorkflowResultEvent
 from vellum.client.types.workflow_stream_event import WorkflowStreamEvent
+from vellum.workflows.constants import undefined
 from vellum.workflows.context import execution_context
 from vellum.workflows.errors import WorkflowErrorCode
 from vellum.workflows.exceptions import NodeException
@@ -493,3 +495,61 @@ class ExampleSubworkflowDeploymentNode(SubworkflowDeploymentNode):
     # AND the error should indicate the missing required input
     assert exc_info.value.code == WorkflowErrorCode.INVALID_INPUTS
     assert exc_info.value.message == "Missing required input for 'my_var_1'"
+
+
+def test_run_workflow__undefined_input_filtered(vellum_client):
+    """Confirm that undefined input values are filtered out when compiling subworkflow inputs."""
+
+    # GIVEN a Subworkflow Deployment Node with an undefined input value
+    class ExampleSubworkflowDeploymentNode(SubworkflowDeploymentNode):
+        deployment = "example_subworkflow_deployment"
+        subworkflow_inputs = {
+            "required_input": "hello",
+            "optional_input": undefined,
+        }
+
+    # AND we know what the Subworkflow Deployment will respond with
+    def generate_subworkflow_events(*args: Any, **kwargs: Any) -> Iterator[WorkflowStreamEvent]:
+        execution_id = str(uuid4())
+        expected_events: List[WorkflowStreamEvent] = [
+            WorkflowExecutionWorkflowResultEvent(
+                execution_id=execution_id,
+                data=WorkflowResultEvent(
+                    id=str(uuid4()),
+                    state="INITIATED",
+                    ts=datetime.now(),
+                ),
+            ),
+            WorkflowExecutionWorkflowResultEvent(
+                execution_id=execution_id,
+                data=WorkflowResultEvent(
+                    id=str(uuid4()),
+                    state="FULFILLED",
+                    ts=datetime.now(),
+                    outputs=[
+                        WorkflowOutputString(
+                            id=str(uuid4()),
+                            name="result",
+                            value="success",
+                        )
+                    ],
+                ),
+            ),
+        ]
+        yield from expected_events
+
+    vellum_client.execute_workflow_stream.side_effect = generate_subworkflow_events
+
+    # WHEN we run the node
+    node = ExampleSubworkflowDeploymentNode()
+    events = list(node.run())
+
+    # THEN the node should have completed successfully
+    assert events[-1].name == "result"
+    assert events[-1].value == "success"
+
+    # AND the undefined input should be filtered out from the API call
+    call_kwargs = vellum_client.execute_workflow_stream.call_args.kwargs
+    assert call_kwargs["inputs"] == [
+        WorkflowRequestStringInputRequest(name="required_input", value="hello"),
+    ]