Keep DataConverter payload converters unwrapped

CHANGELOG.md

@@ -29,6 +29,8 @@ to include examples, links to docs, or any other relevant information.
 
 ### Fixed
 
+- `DataConverter.payload_converter` and current workflow and activity payload converter accessors
+  now return the configured converter without SDK-internal transfer type conversion.
 - Restore pickling of Pydantic data converters, preserving the type adapter cache
   size limit while excluding cached adapters.
 - Current workflow and activity payload converter accessors now return the configured converter

temporalio/activity.py

@@ -241,13 +241,9 @@ def payload_converter(self) -> temporalio.converter.PayloadConverter:
                 self.payload_converter_class_or_instance,
                 temporalio.converter.PayloadConverter,
             ):
-                self._payload_converter = _TemporalTransferTypePayloadConverter.wrap(
-                    self.payload_converter_class_or_instance
-                )
+                self._payload_converter = self.payload_converter_class_or_instance
             else:
-                self._payload_converter = _TemporalTransferTypePayloadConverter.wrap(
-                    self.payload_converter_class_or_instance()
-                )
+                self._payload_converter = self.payload_converter_class_or_instance()
         return self._payload_converter
 
     @property

temporalio/converter/_data_converter.py

@@ -89,11 +89,19 @@ class DataConverter(WithSerializationContext):
     """Singleton default data converter."""
 
     def __post_init__(self) -> None:  # noqa: D105
-        object.__setattr__(self, "payload_converter", self._new_payload_converter())
+        object.__setattr__(self, "payload_converter", self.payload_converter_class())
         object.__setattr__(self, "failure_converter", self.failure_converter_class())
 
-    def _new_payload_converter(self) -> PayloadConverter:
+    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:
         """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()
         )
@@ -113,7 +121,7 @@ async def encode(
             same number as values given, but must be at least one and cannot be
             more than was given.
         """
-        payloads = self.payload_converter.to_payloads(values)
+        payloads = self._get_internal_payload_converter().to_payloads(values)
         payloads = await self._encode_payload_sequence(payloads)
         payloads = await self._external_store_payload_sequence(payloads)
         return payloads
@@ -135,7 +143,9 @@ async def decode(
         """
         payloads = await self._external_retrieve_payload_sequence(payloads)
         payloads = await self._decode_payload_sequence(payloads)
-        return self.payload_converter.from_payloads(payloads, type_hints)
+        return self._get_internal_payload_converter().from_payloads(
+            payloads, type_hints
+        )
 
     async def encode_wrapper(
         self, values: Sequence[Any]
@@ -161,15 +171,19 @@ 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.payload_converter, failure)
+        self.failure_converter.to_failure(
+            exception, self._get_internal_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.payload_converter)
+        return self.failure_converter.from_failure(
+            failure, self._get_internal_payload_converter()
+        )
 
     def with_context(self, context: SerializationContext) -> Self:
         """Return an instance with context set on the component converters."""

temporalio/nexus/system/__init__.py

@@ -98,12 +98,12 @@ class _SystemNexusPayloadConverter(temporalio.converter.PayloadConverter):
 
     def __init__(
         self,
-        user_payload_converter: temporalio.converter.PayloadConverter,
+        internal_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(
-            user_payload_converter, user_failure_converter
+            internal_payload_converter, user_failure_converter
         )
         self._outer_payload_converter = _TemporalTransferTypePayloadConverter.wrap(
             _SystemNexusOuterPayloadConverter()
@@ -170,12 +170,14 @@ async def maybe_visit_payload(
     return payload_converter.to_payload(value)
 
 
-def _get_payload_converter(  # pyright: ignore[reportUnusedFunction]
-    user_payload_converter: temporalio.converter.PayloadConverter,
+def _get_system_nexus_payload_converter(  # pyright: ignore[reportUnusedFunction]
+    internal_payload_converter: temporalio.converter.PayloadConverter,
     user_failure_converter: temporalio.converter.FailureConverter,
 ) -> temporalio.converter.PayloadConverter:
-    """Return the fixed payload converter for system Nexus outer envelopes."""
-    return _SystemNexusPayloadConverter(user_payload_converter, user_failure_converter)
+    """Return the system envelope converter given an internal, wrapped payload converter."""
+    return _SystemNexusPayloadConverter(
+        internal_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.payload_converter,
+                payload_converter_class_or_instance=data_converter._get_internal_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.payload_converter
+                self._worker._data_converter._get_internal_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.payload_converter,
+                    data_converter._get_internal_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.payload_converter,
+                    data_converter._get_internal_payload_converter(),
                     completion.failure,
                 )
 
@@ -471,7 +471,11 @@ async def _start_operation(
                     )
                 )
             elif isinstance(result, nexusrpc.handler.StartOperationResultSync):
-                [payload] = data_converter.payload_converter.to_payloads([result.value])
+                [payload] = (
+                    data_converter._get_internal_payload_converter().to_payloads(
+                        [result.value]
+                    )
+                )
                 return temporalio.api.nexus.v1.StartOperationResponse(
                     sync_success=temporalio.api.nexus.v1.StartOperationResponse.Sync(
                         payload=payload,
@@ -502,7 +506,7 @@ async def _start_operation(
                 response = temporalio.api.nexus.v1.StartOperationResponse()
                 data_converter.failure_converter.to_failure(
                     new_err,
-                    data_converter.payload_converter,
+                    data_converter._get_internal_payload_converter(),
                     response.failure,
                 )
                 return response
@@ -597,11 +601,13 @@ async def deserialize(
             ) from err
 
         try:
-            payload_converter = dc.payload_converter
+            payload_converter = dc._get_internal_payload_converter()
             if temporalio.nexus.system._is_system_payload(payload):
-                payload_converter = temporalio.nexus.system._get_payload_converter(
-                    dc.payload_converter,
-                    dc.failure_converter,
+                payload_converter = (
+                    temporalio.nexus.system._get_system_nexus_payload_converter(
+                        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.payload_converter,
+                        data_converter._get_internal_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.payload_converter,
+                        data_converter._get_internal_payload_converter(),
                         completion.failed.failure,
                     )
                 except Exception as inner_err:
@@ -727,7 +727,9 @@ def _create_workflow_instance(
 
         # Create instance from details
         det = WorkflowInstanceDetails(
-            payload_converter_factory=self._data_converter._new_payload_converter,
+            # 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,
             failure_converter_class=self._data_converter.failure_converter_class,
             interceptor_classes=self._interceptor_classes,
             defn=defn,

temporalio/worker/_workflow_instance.py

@@ -1057,9 +1057,11 @@ def _apply_resolve_nexus_operation(
                     )
                 )
                 if serialization_context is not None:
-                    payload_converter = temporalio.nexus.system._get_payload_converter(
-                        self._payload_converter_with_context(serialization_context),
-                        self._failure_converter_with_context(serialization_context),
+                    payload_converter = (
+                        temporalio.nexus.system._get_system_nexus_payload_converter(
+                            self._payload_converter_with_context(serialization_context),
+                            self._failure_converter_with_context(serialization_context),
+                        )
                     )
             [output] = self._convert_payloads(
                 [result.completed],
@@ -2186,18 +2188,20 @@ async def operation_handle_fn() -> OutputT:
                 input.operation_name,
                 input.input,
             )
-            user_payload_converter = self._workflow_context_payload_converter
+            internal_payload_converter = self._workflow_context_payload_converter
             user_failure_converter = self._workflow_context_failure_converter
             if serialization_context is not None:
-                user_payload_converter = self._payload_converter_with_context(
+                internal_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_payload_converter(
-                user_payload_converter,
-                user_failure_converter,
+            payload_converter = (
+                temporalio.nexus.system._get_system_nexus_payload_converter(
+                    internal_payload_converter,
+                    user_failure_converter,
+                )
             )
             failure_converter = user_failure_converter
         else:

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_factory=temporalio.converter.DataConverter.default._new_payload_converter,
+                payload_converter_factory=temporalio.converter.DataConverter.default._new_internal_payload_converter,
                 failure_converter_class=temporalio.converter.DataConverter.default.failure_converter_class,
                 interceptor_classes=[],
                 defn=defn,