diff --git a/CHANGELOG.md b/CHANGELOG.md
index 31036bfed..aa1e34683 100644
--- a/CHANGELOG.md
+++ b/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`.
diff --git a/temporalio/activity.py b/temporalio/activity.py
index 4e632701e..a51a8c007 100644
--- a/temporalio/activity.py
+++ b/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
diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py
index 823d1cc13..4f184c13c 100644
--- a/temporalio/converter/_data_converter.py
+++ b/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]:
diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py
index 8ee85ef72..7d29d65b8 100644
--- a/temporalio/converter/_payload_converter.py
+++ b/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.
 
diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py
index 21c5a1408..b83cbf892 100644
--- a/temporalio/nexus/system/__init__.py
+++ b/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,9 +104,11 @@ 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__ = [
@@ -53,5 +116,4 @@ def get_payload_converter() -> temporalio.converter.PayloadConverter:
     "get_payload_converter",
     "is_system_endpoint",
     "maybe_visit_payload",
-    "SystemNexusPayloadConverter",
 ]
diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py
index b9513068d..900d14b43 100644
--- a/temporalio/worker/_workflow.py
+++ b/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,
diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py
index 726ff85e0..eaf2aa874 100644
--- a/temporalio/worker/_workflow_instance.py
+++ b/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
         )
diff --git a/temporalio/worker/workflow_sandbox/_runner.py b/temporalio/worker/workflow_sandbox/_runner.py
index 17f473d64..7f06bfcd6 100644
--- a/temporalio/worker/workflow_sandbox/_runner.py
+++ b/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,
diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py
index b689ee8d9..4629d230b 100644
--- a/tests/nexus/test_temporal_system_nexus.py
+++ b/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
diff --git a/tests/test_converter.py b/tests/test_converter.py
index 10365f9c1..6e3a31685 100644
--- a/tests/test_converter.py
+++ b/tests/test_converter.py
@@ -48,6 +48,7 @@
     encode_search_attribute_values,
     value_to_type,
 )
+from temporalio.converter._payload_converter import _TemporalDataModelPayloadConverter
 from temporalio.exceptions import (
     ApplicationError,
     FailureError,
@@ -254,6 +255,52 @@ def test_binary_proto():
     assert decoded == proto
 
 
+@dataclass
+class TemporalIntermediateModel:
+    value: str
+
+    @classmethod
+    def _temporal_from_data_model(
+        cls,
+        data_model: temporalio.api.common.v1.WorkflowExecution,
+    ) -> TemporalIntermediateModel:
+        return cls(value=data_model.workflow_id)
+
+    def _temporal_to_data_model(self) -> temporalio.api.common.v1.WorkflowExecution:
+        return temporalio.api.common.v1.WorkflowExecution(
+            workflow_id=self.value,
+            run_id="run-id",
+        )
+
+
+class CustomDefaultPayloadConverter(DefaultPayloadConverter):
+    pass
+
+
+def test_temporal_data_model_payload_converter_wraps_user_converter():
+    data_converter = DataConverter(
+        payload_converter_class=CustomDefaultPayloadConverter
+    )
+    converter = data_converter.payload_converter
+    assert isinstance(converter, _TemporalDataModelPayloadConverter)
+    value = TemporalIntermediateModel("workflow-id")
+
+    payload = converter.to_payload(value)
+
+    assert payload.metadata["encoding"] == b"json/protobuf"
+    assert (
+        payload.metadata["messageType"] == b"temporal.api.common.v1.WorkflowExecution"
+    )
+    assert all("temporal-wire" not in key for key in payload.metadata)
+    assert all(b"temporal-wire" not in value for value in payload.metadata.values())
+    assert converter.from_payload(payload, TemporalIntermediateModel) == value
+
+    plain_proto_payload = converter.to_payload(
+        temporalio.api.common.v1.WorkflowExecution(workflow_id="id1", run_id="id2")
+    )
+    assert plain_proto_payload.metadata["encoding"] == b"json/protobuf"
+
+
 def test_encode_search_attribute_values():
     with pytest.raises(TypeError, match="of type tuple not one of"):
         encode_search_attribute_values([("bad type",)])  # type: ignore[arg-type]
diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py
index bd4004625..aa9a931e1 100644
--- a/tests/worker/test_visitor.py
+++ b/tests/worker/test_visitor.py
@@ -357,7 +357,9 @@ async def _visit(self) -> None:
             finally:
                 active_visits -= 1
 
-    payload_converter = nexus_system.get_payload_converter()
+    payload_converter = nexus_system.get_payload_converter(
+        temporalio.converter.PayloadConverter.default
+    )
     system_request = workflowservice_pb2.SignalWithStartWorkflowExecutionRequest(
         input=Payloads(payloads=[Payload(data=b"workflow-input")]),
         signal_input=Payloads(payloads=[Payload(data=b"signal-input")]),