💥 Add data model payload conversion hooks

CHANGELOG.md

@@ -20,6 +20,11 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- Added SDK payload converter support for values and type hints that expose
+  `_temporal_to_data_model` and `_temporal_from_data_model` hooks. This lets
+  hook-aware types delegate their wire representation to the configured payload
+  converter, preserving SDK behavior such as serialization contexts for nested
+  payload fields.
 - Added `TLSConfig.verification_server_name` to verify the server certificate against a fixed name
   instead of the connection's server name. Unlike `domain`, it does not change the TLS SNI or
   HTTP/2 authority values, which keep following the connected host, so it can be used when the
@@ -37,6 +42,10 @@ to include examples, links to docs, or any other relevant information.
 
 ### Breaking Changes
 
+- Custom workflow runners that construct `WorkflowInstanceDetails` must now pass
+  `payload_converter_factory` instead of `payload_converter_class`. The factory
+  returns the already wrapped payload converter that workflow instances should
+  use.
 - Payload size limits have moved from `DataConverter` to `Client.connect`. Pass
   `payload_limits=PayloadLimitsConfig(...)` (now exported from
   `temporalio.client`) instead of setting `payload_limits` on `DataConverter`.

temporalio/activity.py

@@ -29,6 +29,7 @@
 import temporalio.bridge.proto.activity_task
 import temporalio.common
 import temporalio.converter
+from temporalio.converter._payload_converter import _TemporalDataModelPayloadConverter
 
 from .types import CallableType
 
@@ -238,9 +239,13 @@ def payload_converter(self) -> temporalio.converter.PayloadConverter:
                 self.payload_converter_class_or_instance,
                 temporalio.converter.PayloadConverter,
             ):
-                self._payload_converter = self.payload_converter_class_or_instance
+                self._payload_converter = _TemporalDataModelPayloadConverter.wrap(
+                    self.payload_converter_class_or_instance
+                )
             else:
-                self._payload_converter = self.payload_converter_class_or_instance()
+                self._payload_converter = _TemporalDataModelPayloadConverter.wrap(
+                    self.payload_converter_class_or_instance()
+                )
         return self._payload_converter
 
     @property

temporalio/converter/_data_converter.py

@@ -28,6 +28,7 @@
 )
 from temporalio.converter._payload_converter import (
     PayloadConverter,
+    _TemporalDataModelPayloadConverter,
 )
 from temporalio.converter._serialization_context import (
     SerializationContext,
@@ -90,9 +91,13 @@ class DataConverter(WithSerializationContext):
     """Singleton default data converter."""
 
     def __post_init__(self) -> None:  # noqa: D105
-        object.__setattr__(self, "payload_converter", self.payload_converter_class())
+        object.__setattr__(self, "payload_converter", self._new_payload_converter())
         object.__setattr__(self, "failure_converter", self.failure_converter_class())
 
+    def _new_payload_converter(self) -> PayloadConverter:
+        """Create a payload converter instance with SDK data model hooks enabled."""
+        return _TemporalDataModelPayloadConverter.wrap(self.payload_converter_class())
+
     async def encode(
         self, values: Sequence[Any]
     ) -> list[temporalio.api.common.v1.Payload]:

temporalio/converter/_payload_converter.py

@@ -514,6 +514,81 @@ def from_payload(
             raise RuntimeError("Failed parsing") from err
 
 
+class _TemporalDataModelPayloadConverter(PayloadConverter, WithSerializationContext):
+    """Payload converter wrapper for generated Temporal data model hooks.
+
+    Values with a ``_temporal_to_data_model`` method are first converted to
+    their data model value, then encoded by the wrapped payload converter. When
+    decoding to a type with ``_temporal_from_data_model``, the wrapped
+    converter first decodes the payload to the data model value and this
+    wrapper constructs the requested user-facing type from it.
+    """
+
+    _inner_payload_converter: PayloadConverter
+
+    def __init__(self, inner_payload_converter: PayloadConverter) -> None:
+        """Create a Temporal data model payload converter."""
+        self._inner_payload_converter = inner_payload_converter
+
+    @staticmethod
+    def wrap(payload_converter: PayloadConverter) -> PayloadConverter:
+        """Wrap a payload converter unless it is already wrapped."""
+        if isinstance(payload_converter, _TemporalDataModelPayloadConverter):
+            return payload_converter
+        return _TemporalDataModelPayloadConverter(payload_converter)
+
+    def to_payloads(
+        self, values: Sequence[Any]
+    ) -> list[temporalio.api.common.v1.Payload]:
+        """See base class."""
+        data_model_values: list[Any] = []
+        for value in values:
+            to_data_model = getattr(value, "_temporal_to_data_model", None)
+            if to_data_model is not None:
+                value = to_data_model()
+            data_model_values.append(value)
+        return self._inner_payload_converter.to_payloads(data_model_values)
+
+    def from_payloads(
+        self,
+        payloads: Sequence[temporalio.api.common.v1.Payload],
+        type_hints: list[type] | None = None,
+    ) -> list[Any]:
+        """See base class."""
+        if type_hints is None:
+            return self._inner_payload_converter.from_payloads(payloads, None)
+        normalized_type_hints: list[type | None] = list(type_hints)
+        if len(normalized_type_hints) < len(payloads):
+            normalized_type_hints.extend([None] * (len(payloads) - len(type_hints)))
+        inner_type_hints = [
+            None
+            if getattr(type_hint, "_temporal_from_data_model", None) is not None
+            else type_hint
+            for type_hint in normalized_type_hints
+        ]
+        values = self._inner_payload_converter.from_payloads(
+            payloads, typing.cast("list[type]", inner_type_hints)
+        )
+        return [
+            from_data_model(value)
+            if (
+                from_data_model := getattr(type_hint, "_temporal_from_data_model", None)
+            )
+            is not None
+            else value
+            for value, type_hint in zip(values, normalized_type_hints)
+        ]
+
+    def with_context(self, context: SerializationContext) -> Self:
+        """Return a new instance with context set on the inner converter."""
+        if not isinstance(self._inner_payload_converter, WithSerializationContext):
+            return self
+        inner_payload_converter = self._inner_payload_converter.with_context(context)
+        if inner_payload_converter is self._inner_payload_converter:
+            return self
+        return type(self)(inner_payload_converter)
+
+
 class AdvancedJSONEncoder(json.JSONEncoder):
     """Advanced JSON encoder.
 

temporalio/nexus/system/__init__.py

@@ -2,22 +2,83 @@
 
 from __future__ import annotations
 
+import contextlib
+import contextvars
+from collections.abc import Iterator, Sequence
+from typing import Any
+
 import temporalio.api.common.v1
 import temporalio.converter
 from temporalio.bridge._visitor_functions import VisitorFunctions
 from temporalio.converter import BinaryProtoPayloadConverter, CompositePayloadConverter
+from temporalio.converter._payload_converter import _TemporalDataModelPayloadConverter
 
 TEMPORAL_SYSTEM_ENDPOINT = "__temporal_system"
+_user_payload_converter: contextvars.ContextVar[
+    temporalio.converter.PayloadConverter | None
+] = contextvars.ContextVar("temporal-system-nexus-user-payload-converter", default=None)
 
 
-class SystemNexusPayloadConverter(CompositePayloadConverter):
-    """Payload converter for system Nexus outer envelopes."""
+@contextlib.contextmanager
+def _user_payload_converter_context(
+    payload_converter: temporalio.converter.PayloadConverter,
+) -> Iterator[None]:
+    """Set the user payload converter for system Nexus model conversion."""
+    token = _user_payload_converter.set(payload_converter)
+    try:
+        yield
+    finally:
+        _user_payload_converter.reset(token)
+
+
+def _current_user_payload_converter() -> temporalio.converter.PayloadConverter:  # pyright: ignore[reportUnusedFunction]
+    """Return the active user payload converter for system Nexus model conversion."""
+    payload_converter = _user_payload_converter.get()
+    if payload_converter is None:
+        raise RuntimeError("System Nexus user payload converter context is not active")
+    return payload_converter
+
+
+class _SystemNexusOuterPayloadConverter(CompositePayloadConverter):
+    """Payload converter for system Nexus outer proto envelopes."""
 
     def __init__(self) -> None:
         """Create a payload converter for system Nexus outer envelopes."""
         super().__init__(BinaryProtoPayloadConverter())
 
 
+class _SystemNexusPayloadConverter(temporalio.converter.PayloadConverter):
+    """Payload converter for system Nexus outer envelopes."""
+
+    _user_payload_converter: temporalio.converter.PayloadConverter
+    _outer_payload_converter: temporalio.converter.PayloadConverter
+
+    def __init__(
+        self, user_payload_converter: temporalio.converter.PayloadConverter
+    ) -> None:
+        """Create a payload converter for system Nexus outer envelopes."""
+        self._user_payload_converter = user_payload_converter
+        self._outer_payload_converter = _TemporalDataModelPayloadConverter.wrap(
+            _SystemNexusOuterPayloadConverter()
+        )
+
+    def to_payloads(
+        self, values: Sequence[Any]
+    ) -> list[temporalio.api.common.v1.Payload]:
+        """See base class."""
+        with _user_payload_converter_context(self._user_payload_converter):
+            return self._outer_payload_converter.to_payloads(values)
+
+    def from_payloads(
+        self,
+        payloads: Sequence[temporalio.api.common.v1.Payload],
+        type_hints: list[type] | None = None,
+    ) -> list[Any]:
+        """See base class."""
+        with _user_payload_converter_context(self._user_payload_converter):
+            return self._outer_payload_converter.from_payloads(payloads, type_hints)
+
+
 def is_system_endpoint(endpoint: str) -> bool:
     """Return whether a Nexus endpoint is the Temporal system endpoint."""
     return endpoint == TEMPORAL_SYSTEM_ENDPOINT
@@ -33,7 +94,7 @@ async def maybe_visit_payload(
     if not is_system_endpoint(endpoint):
         return None
 
-    payload_converter = get_payload_converter()
+    payload_converter = _SystemNexusOuterPayloadConverter()
     value = payload_converter.from_payload(payload)
     from ._payload_visitor import PayloadVisitor
 
@@ -43,15 +104,16 @@ async def maybe_visit_payload(
     return payload_converter.to_payload(value)
 
 
-def get_payload_converter() -> temporalio.converter.PayloadConverter:
+def get_payload_converter(
+    user_payload_converter: temporalio.converter.PayloadConverter,
+) -> temporalio.converter.PayloadConverter:
     """Return the fixed payload converter for system Nexus outer envelopes."""
-    return SystemNexusPayloadConverter()
+    return _SystemNexusPayloadConverter(user_payload_converter)
 
 
 __all__ = [
     "TEMPORAL_SYSTEM_ENDPOINT",
     "get_payload_converter",
     "is_system_endpoint",
     "maybe_visit_payload",
-    "SystemNexusPayloadConverter",
 ]

temporalio/worker/_workflow.py

@@ -776,7 +776,7 @@ def _create_workflow_instance(
 
         # Create instance from details
         det = WorkflowInstanceDetails(
-            payload_converter_class=self._data_converter.payload_converter_class,
+            payload_converter_factory=self._data_converter._new_payload_converter,
             failure_converter_class=self._data_converter.failure_converter_class,
             interceptor_classes=self._interceptor_classes,
             defn=defn,

temporalio/worker/_workflow_instance.py

@@ -157,7 +157,7 @@ class PatchActivationInput:
 class WorkflowInstanceDetails:
     """Immutable details for creating a workflow instance."""
 
-    payload_converter_class: type[temporalio.converter.PayloadConverter]
+    payload_converter_factory: Callable[[], temporalio.converter.PayloadConverter]
     failure_converter_class: type[temporalio.converter.FailureConverter]
     interceptor_classes: Sequence[type[WorkflowInboundInterceptor]]
     defn: temporalio.workflow._Definition
@@ -269,7 +269,7 @@ def __init__(self, det: WorkflowInstanceDetails) -> None:
         self._defn = det.defn
         self._workflow_input: ExecuteWorkflowInput | None = None
         self._info = det.info
-        self._context_free_payload_converter = det.payload_converter_class()
+        self._context_free_payload_converter = det.payload_converter_factory()
         self._context_free_failure_converter = det.failure_converter_class()
         workflow_context = temporalio.converter.WorkflowSerializationContext(
             namespace=det.info.namespace,
@@ -2130,7 +2130,9 @@ async def operation_handle_fn() -> OutputT:
                         t.uncancel()  # type: ignore[union-attr]
 
         payload_converter = (
-            temporalio.nexus.system.get_payload_converter()
+            temporalio.nexus.system.get_payload_converter(
+                self._workflow_context_payload_converter
+            )
             if temporalio.nexus.system.is_system_endpoint(input.endpoint)
             else self._context_free_payload_converter
         )

temporalio/worker/workflow_sandbox/_runner.py

@@ -79,7 +79,7 @@ def prepare_workflow(self, defn: temporalio.workflow._Definition) -> None:
         # Just create with fake info which validates
         self.create_instance(
             WorkflowInstanceDetails(
-                payload_converter_class=temporalio.converter.DataConverter.default.payload_converter_class,
+                payload_converter_factory=temporalio.converter.DataConverter.default._new_payload_converter,
                 failure_converter_class=temporalio.converter.DataConverter.default.failure_converter_class,
                 interceptor_classes=[],
                 defn=defn,

tests/nexus/test_temporal_system_nexus.py

@@ -186,7 +186,9 @@ def _new_system_nexus_request_payload() -> temporalio.api.common.v1.Payload:
     assert nested_payload is not None
     request = workflowservice_pb2.SignalWithStartWorkflowExecutionRequest()
     request.input.payloads.add().CopyFrom(nested_payload)
-    payload = nexus_system.get_payload_converter().to_payload(request)
+    payload = nexus_system.get_payload_converter(
+        temporalio.converter.PayloadConverter.default
+    ).to_payload(request)
     assert payload is not None
     return payload
 
@@ -201,7 +203,9 @@ async def test_schedule_system_nexus_endpoint_ignores_operation_registry() -> No
     await PayloadVisitor().visit(visitor, completion)
 
     schedule = completion.successful.commands[0].schedule_nexus_operation
-    decoded = nexus_system.get_payload_converter().from_payload(schedule.input)
+    decoded = nexus_system.get_payload_converter(
+        temporalio.converter.PayloadConverter.default
+    ).from_payload(schedule.input)
     assert isinstance(
         decoded, workflowservice_pb2.SignalWithStartWorkflowExecutionRequest
     )
@@ -339,7 +343,9 @@ def _field_is_repeated(field: FieldDescriptor) -> bool:
     ],
 )
 def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None:
-    payload_converter = nexus_system.get_payload_converter()
+    payload_converter = nexus_system.get_payload_converter(
+        temporalio.converter.PayloadConverter.default
+    )
     proto_value = _build_proto_sample(message_type)
     payload = payload_converter.to_payload(proto_value)
     assert payload is not None