Restore wrapped public payload converters

CHANGELOG.md

@@ -22,6 +22,10 @@ to include examples, links to docs, or any other relevant information.
 
 ### Changed
 
+- Payload converters exposed by data converters and workflow/activity accessors
+  retain transfer type conversion, so direct use behaves consistently with SDK
+  serialization.
+
 ### Deprecated
 
 ### :boom: Breaking Changes

temporalio/activity.py

@@ -241,9 +241,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 = _TemporalTransferTypePayloadConverter.wrap(
+                    self.payload_converter_class_or_instance
+                )
             else:
-                self._payload_converter = self.payload_converter_class_or_instance()
+                self._payload_converter = _TemporalTransferTypePayloadConverter.wrap(
+                    self.payload_converter_class_or_instance()
+                )
         return self._payload_converter
 
     @property
@@ -458,9 +462,7 @@ def payload_converter() -> temporalio.converter.PayloadConverter:
     The returned converter has :py:class:`temporalio.converter.ActivitySerializationContext` set.
     This is often used for dynamic activities to convert payloads.
     """
-    return _TemporalTransferTypePayloadConverter.unwrap(
-        _Context.current().payload_converter
-    )
+    return _Context.current().payload_converter
 
 
 def metric_meter() -> temporalio.common.MetricMeter:

temporalio/converter/_data_converter.py

@@ -73,7 +73,9 @@ class DataConverter(WithSerializationContext):
     """Class to instantiate for failure conversion."""
 
     payload_converter: PayloadConverter = dataclasses.field(init=False)
-    """Payload converter created from the :py:attr:`payload_converter_class`."""
+    """Payload converter created from :py:attr:`payload_converter_class` and wrapped
+    with SDK transfer type conversion, including when called directly.
+    """
 
     failure_converter: FailureConverter = dataclasses.field(init=False)
     """Failure converter created from the :py:attr:`failure_converter_class`."""
@@ -89,19 +91,11 @@ 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 _get_internal_payload_converter(self) -> PayloadConverter:
-        """Return the configured converter with SDK transfer type hooks enabled."""
-        # Reuse the configured instance so its state and any applied serialization
-        # context are preserved. A new wrapper does not create a new user converter.
-        return _TemporalTransferTypePayloadConverter.wrap(self.payload_converter)
-
-    def _new_internal_payload_converter(self) -> PayloadConverter:
+    def _new_payload_converter(self) -> PayloadConverter:
         """Create a payload converter instance with SDK transfer type hooks enabled."""
-        # Unlike _get_internal_payload_converter, instantiate a fresh user converter
-        # so workflow instances do not share mutable custom converter state.
         return _TemporalTransferTypePayloadConverter.wrap(
             self.payload_converter_class()
         )
@@ -121,7 +115,7 @@ async def encode(
             same number as values given, but must be at least one and cannot be
             more than was given.
         """
-        payloads = self._get_internal_payload_converter().to_payloads(values)
+        payloads = self.payload_converter.to_payloads(values)
         payloads = await self._encode_payload_sequence(payloads)
         payloads = await self._external_store_payload_sequence(payloads)
         return payloads
@@ -143,9 +137,7 @@ async def decode(
         """
         payloads = await self._external_retrieve_payload_sequence(payloads)
         payloads = await self._decode_payload_sequence(payloads)
-        return self._get_internal_payload_converter().from_payloads(
-            payloads, type_hints
-        )
+        return self.payload_converter.from_payloads(payloads, type_hints)
 
     async def encode_wrapper(
         self, values: Sequence[Any]
@@ -171,19 +163,15 @@ async def encode_failure(
         self, exception: BaseException, failure: temporalio.api.failure.v1.Failure
     ) -> None:
         """Convert and encode failure."""
-        self.failure_converter.to_failure(
-            exception, self._get_internal_payload_converter(), failure
-        )
+        self.failure_converter.to_failure(exception, self.payload_converter, failure)
         await _apply_to_failure_payloads(failure, self._transform_outbound_payloads)
 
     async def decode_failure(
         self, failure: temporalio.api.failure.v1.Failure
     ) -> BaseException:
         """Decode and convert failure."""
         await _apply_to_failure_payloads(failure, self._transform_inbound_payloads)
-        return self.failure_converter.from_failure(
-            failure, self._get_internal_payload_converter()
-        )
+        return self.failure_converter.from_failure(failure, self.payload_converter)
 
     def with_context(self, context: SerializationContext) -> Self:
         """Return an instance with context set on the component converters."""

temporalio/converter/_payload_converter.py

@@ -612,13 +612,6 @@ def wrap(payload_converter: PayloadConverter) -> PayloadConverter:
             return payload_converter
         return _TemporalTransferTypePayloadConverter(payload_converter)
 
-    @staticmethod
-    def unwrap(payload_converter: PayloadConverter) -> PayloadConverter:
-        """Remove this wrapper from a payload converter, if present."""
-        if isinstance(payload_converter, _TemporalTransferTypePayloadConverter):
-            return payload_converter._inner_payload_converter
-        return payload_converter
-
     def to_payloads(
         self, values: Sequence[Any]
     ) -> list[temporalio.api.common.v1.Payload]:

temporalio/nexus/system/__init__.py

@@ -59,9 +59,7 @@ def _current_user_converters() -> _SystemNexusUserConverters:
 
 def _current_user_payload_converter() -> temporalio.converter.PayloadConverter:  # pyright: ignore[reportUnusedFunction]
     """Return the active user payload converter for system Nexus model conversion."""
-    return _TemporalTransferTypePayloadConverter.unwrap(
-        _current_user_converters().payload_converter
-    )
+    return _current_user_converters().payload_converter
 
 
 def _current_user_failure_converter() -> temporalio.converter.FailureConverter:  # pyright: ignore[reportUnusedFunction]
@@ -98,12 +96,12 @@ class _SystemNexusPayloadConverter(temporalio.converter.PayloadConverter):
 
     def __init__(
         self,
-        internal_payload_converter: temporalio.converter.PayloadConverter,
+        user_payload_converter: temporalio.converter.PayloadConverter,
         user_failure_converter: temporalio.converter.FailureConverter,
     ) -> None:
         """Create a payload converter for system Nexus outer envelopes."""
         self._user_converters = _SystemNexusUserConverters(
-            internal_payload_converter, user_failure_converter
+            user_payload_converter, user_failure_converter
         )
         self._outer_payload_converter = _TemporalTransferTypePayloadConverter.wrap(
             _SystemNexusOuterPayloadConverter()
@@ -170,14 +168,12 @@ async def maybe_visit_payload(
     return payload_converter.to_payload(value)
 
 
-def _get_system_nexus_payload_converter(  # pyright: ignore[reportUnusedFunction]
-    internal_payload_converter: temporalio.converter.PayloadConverter,
+def _get_payload_converter(  # pyright: ignore[reportUnusedFunction]
+    user_payload_converter: temporalio.converter.PayloadConverter,
     user_failure_converter: temporalio.converter.FailureConverter,
 ) -> temporalio.converter.PayloadConverter:
-    """Return the system envelope converter given an internal, wrapped payload converter."""
-    return _SystemNexusPayloadConverter(
-        internal_payload_converter, user_failure_converter
-    )
+    """Return the fixed payload converter for system Nexus outer envelopes."""
+    return _SystemNexusPayloadConverter(user_payload_converter, user_failure_converter)
 
 
 def _get_serialization_context(  # pyright: ignore[reportUnusedFunction]

temporalio/worker/_activity.py

@@ -662,7 +662,7 @@ async def _execute_activity(
                     if not running_activity.cancel_thread_raiser
                     else running_activity.cancel_thread_raiser.shielded
                 ),
-                payload_converter_class_or_instance=data_converter._get_internal_payload_converter(),
+                payload_converter_class_or_instance=data_converter.payload_converter,
                 runtime_metric_meter=None if sync_non_threaded else self._metric_meter,
                 client=self._client if not running_activity.sync else None,
                 cancellation_details=running_activity.cancellation_details,
@@ -847,7 +847,7 @@ async def heartbeat_with_context(*details: Any) -> None:
             # The payload converter is the already instantiated one for thread
             # or the picklable class for non-thread
             payload_converter_class_or_instance = (
-                self._worker._data_converter._get_internal_payload_converter()
+                self._worker._data_converter.payload_converter
                 if isinstance(input.executor, concurrent.futures.ThreadPoolExecutor)
                 else self._worker._data_converter.payload_converter_class
             )

temporalio/worker/_nexus.py

@@ -323,7 +323,7 @@ async def _handle_cancel_operation_task(
                 )
                 data_converter.failure_converter.to_failure(
                     handler_error,
-                    data_converter._get_internal_payload_converter(),
+                    data_converter.payload_converter,
                     completion.failure,
                 )
                 await self._encode_completion(completion, data_converter)
@@ -385,7 +385,7 @@ async def _handle_start_operation_task(
                 handler_error = _exception_to_handler_error(err)
                 data_converter.failure_converter.to_failure(
                     handler_error,
-                    data_converter._get_internal_payload_converter(),
+                    data_converter.payload_converter,
                     completion.failure,
                 )
 
@@ -471,11 +471,7 @@ async def _start_operation(
                     )
                 )
             elif isinstance(result, nexusrpc.handler.StartOperationResultSync):
-                [payload] = (
-                    data_converter._get_internal_payload_converter().to_payloads(
-                        [result.value]
-                    )
-                )
+                [payload] = data_converter.payload_converter.to_payloads([result.value])
                 return temporalio.api.nexus.v1.StartOperationResponse(
                     sync_success=temporalio.api.nexus.v1.StartOperationResponse.Sync(
                         payload=payload,
@@ -506,7 +502,7 @@ async def _start_operation(
                 response = temporalio.api.nexus.v1.StartOperationResponse()
                 data_converter.failure_converter.to_failure(
                     new_err,
-                    data_converter._get_internal_payload_converter(),
+                    data_converter.payload_converter,
                     response.failure,
                 )
                 return response
@@ -601,13 +597,11 @@ async def deserialize(
             ) from err
 
         try:
-            payload_converter = dc._get_internal_payload_converter()
+            payload_converter = dc.payload_converter
             if temporalio.nexus.system._is_system_payload(payload):
-                payload_converter = (
-                    temporalio.nexus.system._get_system_nexus_payload_converter(
-                        payload_converter,
-                        dc.failure_converter,
-                    )
+                payload_converter = temporalio.nexus.system._get_payload_converter(
+                    dc.payload_converter,
+                    dc.failure_converter,
                 )
             [input] = payload_converter.from_payloads(
                 [payload],

temporalio/worker/_workflow.py

@@ -452,7 +452,7 @@ async def _handle_activation(
                 try:
                     data_converter.failure_converter.to_failure(
                         err,
-                        data_converter._get_internal_payload_converter(),
+                        data_converter.payload_converter,
                         failure,
                     )
                 except Exception as inner_err:
@@ -468,7 +468,7 @@ async def _handle_activation(
                 try:
                     data_converter.failure_converter.to_failure(
                         err,
-                        data_converter._get_internal_payload_converter(),
+                        data_converter.payload_converter,
                         completion.failed.failure,
                     )
                 except Exception as inner_err:
@@ -728,9 +728,7 @@ def _create_workflow_instance(
 
         # Create instance from details
         det = WorkflowInstanceDetails(
-            # Sharing the underlying converter could let another workflow's state
-            # affect conversion and replay, so each instance needs a fresh one.
-            payload_converter_factory=self._data_converter._new_internal_payload_converter,
+            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

@@ -60,9 +60,6 @@
 import temporalio.nexus.system
 import temporalio.workflow
 from temporalio.converter import StorageDriverStoreContext, StorageDriverWorkflowInfo
-from temporalio.converter._payload_converter import (
-    _TemporalTransferTypePayloadConverter,
-)
 from temporalio.nexus.system.workflow_service._system_nexus_interceptor import (
     _start_system_nexus_operation,
     _SystemNexusWorkflowOutboundInterceptorTerminal,
@@ -1063,11 +1060,9 @@ def _apply_resolve_nexus_operation(
                     )
                 )
                 if serialization_context is not None:
-                    payload_converter = (
-                        temporalio.nexus.system._get_system_nexus_payload_converter(
-                            self._payload_converter_with_context(serialization_context),
-                            self._failure_converter_with_context(serialization_context),
-                        )
+                    payload_converter = temporalio.nexus.system._get_payload_converter(
+                        self._payload_converter_with_context(serialization_context),
+                        self._failure_converter_with_context(serialization_context),
                     )
             [output] = self._convert_payloads(
                 [result.completed],
@@ -1498,9 +1493,7 @@ def workflow_patch(
         return use_patch
 
     def workflow_payload_converter(self) -> temporalio.converter.PayloadConverter:
-        return _TemporalTransferTypePayloadConverter.unwrap(
-            self._workflow_context_payload_converter
-        )
+        return self._workflow_context_payload_converter
 
     def workflow_random(self) -> random.Random:
         self._assert_not_read_only("random")
@@ -2247,20 +2240,18 @@ async def operation_handle_fn() -> OutputT:
                 input.operation_name,
                 input.input,
             )
-            internal_payload_converter = self._workflow_context_payload_converter
+            user_payload_converter = self._workflow_context_payload_converter
             user_failure_converter = self._workflow_context_failure_converter
             if serialization_context is not None:
-                internal_payload_converter = self._payload_converter_with_context(
+                user_payload_converter = self._payload_converter_with_context(
                     serialization_context
                 )
                 user_failure_converter = self._failure_converter_with_context(
                     serialization_context
                 )
-            payload_converter = (
-                temporalio.nexus.system._get_system_nexus_payload_converter(
-                    internal_payload_converter,
-                    user_failure_converter,
-                )
+            payload_converter = temporalio.nexus.system._get_payload_converter(
+                user_payload_converter,
+                user_failure_converter,
             )
             failure_converter = user_failure_converter
         else:

temporalio/worker/workflow_sandbox/_runner.py

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