[APO-2731] Filter undefined values from subworkflow inputs

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

@@ -19,6 +19,7 @@
 from vellum.workflows.context import execution_context
 from vellum.workflows.errors import WorkflowErrorCode
 from vellum.workflows.exceptions import NodeException
+from vellum.workflows.nodes.bases.base import BaseNode
 from vellum.workflows.nodes.displayable.subworkflow_deployment_node.node import SubworkflowDeploymentNode
 
 
@@ -493,3 +494,29 @@ 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_compile_subworkflow_inputs__undefined_input_filtered():
+    """Confirm that undefined values in subworkflow_inputs are filtered out when compiling inputs for API calls."""
+
+    # GIVEN an upstream node with an output that is declared but never set (resolves to undefined)
+    class UpstreamNode(BaseNode):
+        class Outputs(BaseNode.Outputs):
+            unset_output: str
+
+    # AND a Subworkflow Deployment Node that references the unset output for an optional input
+    class ExampleSubworkflowDeploymentNode(SubworkflowDeploymentNode):
+        deployment = "example_subworkflow_deployment"
+        subworkflow_inputs = {
+            "required_input": "hello",
+            "optional_input": UpstreamNode.Outputs.unset_output,
+        }
+
+    # WHEN we compile the subworkflow inputs
+    node = ExampleSubworkflowDeploymentNode()
+    compiled_inputs = node._compile_subworkflow_inputs()
+
+    # THEN only the required_input should be included (undefined values should be filtered out)
+    assert len(compiled_inputs) == 1
+    assert compiled_inputs[0].name == "required_input"
+    assert compiled_inputs[0].value == "hello"