Fall back to model_dump_json for OpenAI payload serialization

OpenAI response and stream event types whose pydantic serializer is a
lazily-built MockValSer cannot be serialized by the generic any-schema
serializer, raising PydanticSerializationError (e.g. when streaming via
WorkflowStreamClient). The model's own model_dump_json() handles them.

Fixes #1585

Author: brianstrauch

temporalio/contrib/openai_agents/_temporal_openai_agents.py

@@ -8,6 +8,7 @@
 from datetime import timedelta
 
 import pydantic
+import pydantic_core
 from agents import ModelProvider, Trace, set_trace_provider
 from agents.run import get_default_agent_runner, set_default_agent_runner
 from agents.tracing import get_trace_provider
@@ -110,6 +111,21 @@ class _OpenAIJSONPlainPayloadConverter(PydanticJSONPlainPayloadConverter):
     side does not, so fall back to lenient construction when validation fails.
     """
 
+    def to_payload(self, value: typing.Any) -> temporalio.api.common.v1.Payload | None:
+        try:
+            return super().to_payload(value)
+        except pydantic_core.PydanticSerializationError:
+            dump = getattr(value, "model_dump_json", None)
+            if dump is None:
+                raise
+            exclude_unset = (
+                self._to_json_options.exclude_unset if self._to_json_options else False
+            )
+            return temporalio.api.common.v1.Payload(
+                metadata={"encoding": self.encoding.encode()},
+                data=dump(exclude_unset=exclude_unset).encode(),
+            )
+
     def from_payload(
         self,
         payload: temporalio.api.common.v1.Payload,

tests/contrib/openai_agents/test_openai.py

@@ -1415,6 +1415,43 @@ async def test_response_serialization():
     await pydantic_data_converter.encode([model_response])
 
 
+def test_openai_converter_serialization_fallback(monkeypatch: pytest.MonkeyPatch):
+    import pydantic_core
+    from openai.types.responses import ResponseOutputMessage, ResponseOutputText
+
+    from temporalio.contrib.openai_agents._temporal_openai_agents import (
+        _OpenAIJSONPlainPayloadConverter,
+    )
+    from temporalio.contrib.pydantic import ToJsonOptions
+
+    converter = _OpenAIJSONPlainPayloadConverter(ToJsonOptions(exclude_unset=True))
+
+    class _RaisingSerializer:
+        def to_json(self, *_args: Any, **_kwargs: Any) -> bytes:
+            raise pydantic_core.PydanticSerializationError("forced")
+
+    monkeypatch.setattr(converter, "_schema_serializer", _RaisingSerializer())
+
+    value = ResponseOutputMessage(
+        id="",
+        content=[ResponseOutputText(text="hello", annotations=[], type="output_text")],
+        role="assistant",
+        status="completed",
+        type="message",
+    )
+    payload = converter.to_payload(value)
+    assert payload is not None
+    assert payload.metadata["encoding"] == b"json/plain"
+    assert payload.data == value.model_dump_json(exclude_unset=True).encode()
+
+    decoded = converter.from_payload(payload, type(value))
+    assert decoded == value
+
+    # A non-model value has no model_dump_json, so the error propagates.
+    with pytest.raises(pydantic_core.PydanticSerializationError):
+        converter.to_payload(object())
+
+
 async def assert_status_retry_behavior(status: int, client: Client, should_retry: bool):
     def status_error(status: int):
         with workflow.unsafe.imports_passed_through():