[APO-2194] Raise error on json serialization errors for outputs

Author: Harrison Ngo

src/vellum/workflows/nodes/bases/tests/test_base_node.py

@@ -6,6 +6,7 @@
 from vellum.client.types.string_vellum_value_request import StringVellumValueRequest
 from vellum.workflows.constants import undefined
 from vellum.workflows.descriptors.tests.test_utils import FixtureState
+from vellum.workflows.errors.types import WorkflowErrorCode
 from vellum.workflows.inputs.base import BaseInputs
 from vellum.workflows.nodes import FinalOutputNode
 from vellum.workflows.nodes.bases.base import BaseNode
@@ -15,6 +16,7 @@
 from vellum.workflows.references.node import NodeReference
 from vellum.workflows.references.output import OutputReference
 from vellum.workflows.state.base import BaseState, StateMeta
+from vellum.workflows.workflows.base import BaseWorkflow
 
 
 def test_base_node__node_resolution__unset_pydantic_fields():
@@ -379,3 +381,28 @@ class Ports(MyNode.Ports):
     # Potentially in the future, we support inheriting ports from multiple parents.
     # For now, we take only the declared ports, so that not all nodes have the default port.
     assert ports == ["bar"]
+
+
+def test_base_node__bytes_output_raises_serialization_error():
+    """Test that returning bytes in node outputs rejects the workflow execution."""
+
+    class BytesOutputNode(BaseNode):
+        class Outputs(BaseNode.Outputs):
+            result: str
+
+        def run(self) -> "BytesOutputNode.Outputs":
+            b = b"hello"
+            return self.Outputs(result=b)
+
+    class BytesWorkflow(BaseWorkflow):
+        graph = BytesOutputNode
+
+    workflow = BytesWorkflow()
+
+    # WHEN we run the workflow
+    result = workflow.run()
+
+    # THEN the execution is rejected with a helpful error
+    assert result.name == "workflow.execution.rejected"
+    assert result.error.code == WorkflowErrorCode.INVALID_OUTPUTS
+    assert "bytes" in result.error.message.lower()

src/vellum/workflows/runner/runner.py

@@ -49,7 +49,14 @@
     NodeExecutionRejectedBody,
     NodeExecutionStreamingBody,
 )
-from vellum.workflows.events.types import BaseEvent, NodeParentContext, ParentContext, SpanLink, WorkflowParentContext
+from vellum.workflows.events.types import (
+    BaseEvent,
+    NodeParentContext,
+    ParentContext,
+    SpanLink,
+    WorkflowParentContext,
+    default_serializer,
+)
 from vellum.workflows.events.workflow import (
     WorkflowEventStream,
     WorkflowExecutionFulfilledBody,
@@ -565,6 +572,16 @@ def initiate_node_streaming_output(
 
                 with execution_context(parent_context=updated_parent_context, trace_id=execution.trace_id):
                     for output in node_run_response:
+                        try:
+                            default_serializer(output)
+                        except (TypeError, ValueError) as exc:
+                            raise NodeException(
+                                message=(
+                                    f"Node {node.__class__.__name__} produced output: "
+                                    f"'{output.name}' that could not be serialized to JSON: {exc}"
+                                ),
+                                code=WorkflowErrorCode.INVALID_OUTPUTS,
+                            ) from exc
                         invoked_ports = output > ports
                         if output.is_initiated:
                             yield from initiate_node_streaming_output(output)
@@ -599,6 +616,20 @@ def initiate_node_streaming_output(
                                 parent=execution.parent_context,
                             )
 
+            for descriptor, output_value in outputs:
+                if output_value is undefined:
+                    continue
+                try:
+                    default_serializer(output_value)
+                except (TypeError, ValueError) as exc:
+                    raise NodeException(
+                        message=(
+                            f"Node {node.__class__.__name__} produced output '{descriptor.name}' "
+                            f"that could not be serialized to JSON: {exc}"
+                        ),
+                        code=WorkflowErrorCode.INVALID_OUTPUTS,
+                    ) from exc
+
             node.state.meta.node_execution_cache.fulfill_node_execution(node.__class__, span_id)
 
             with execution_context(parent_context=updated_parent_context, trace_id=execution.trace_id):