From b61dad7e03230f12444b125daddf5e6d4caf1747 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 16 Jul 2026 11:53:50 -0700 Subject: [PATCH 1/2] Wrap payload converters for temporal intermediate models --- temporalio/activity.py | 12 +++- temporalio/converter/__init__.py | 2 + temporalio/converter/_data_converter.py | 7 +- temporalio/converter/_payload_converter.py | 77 ++++++++++++++++++++++ temporalio/worker/_workflow_instance.py | 6 +- tests/test_converter.py | 54 +++++++++++++++ 6 files changed, 154 insertions(+), 4 deletions(-) diff --git a/temporalio/activity.py b/temporalio/activity.py index 4e632701e..1c80438e9 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -238,9 +238,17 @@ 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 = ( + temporalio.converter.TemporalIntermediatePayloadConverter.wrap( + self.payload_converter_class_or_instance + ) + ) else: - self._payload_converter = self.payload_converter_class_or_instance() + self._payload_converter = ( + temporalio.converter.TemporalIntermediatePayloadConverter.wrap( + self.payload_converter_class_or_instance() + ) + ) return self._payload_converter @property diff --git a/temporalio/converter/__init__.py b/temporalio/converter/__init__.py index 3821cbd68..7cdec2a76 100644 --- a/temporalio/converter/__init__.py +++ b/temporalio/converter/__init__.py @@ -33,6 +33,7 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadConverter, + TemporalIntermediatePayloadConverter, value_to_type, ) from temporalio.converter._payload_limits import ( @@ -83,6 +84,7 @@ "PayloadLimitsConfig", "PayloadSizeWarning", "SerializationContext", + "TemporalIntermediatePayloadConverter", "WithSerializationContext", "WorkflowSerializationContext", "decode_search_attributes", diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 13b48e695..b0c87b926 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -29,6 +29,7 @@ ) from temporalio.converter._payload_converter import ( PayloadConverter, + TemporalIntermediatePayloadConverter, ) from temporalio.converter._payload_limits import ( PayloadLimitsConfig, @@ -103,7 +104,11 @@ class DataConverter(WithSerializationContext): """Server-reported limits for payloads.""" def __post_init__(self) -> None: # noqa: D105 - object.__setattr__(self, "payload_converter", self.payload_converter_class()) + object.__setattr__( + self, + "payload_converter", + TemporalIntermediatePayloadConverter.wrap(self.payload_converter_class()), + ) object.__setattr__(self, "failure_converter", self.failure_converter_class()) async def encode( diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index 8ee85ef72..261ad1033 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -514,6 +514,83 @@ def from_payload( raise RuntimeError("Failed parsing") from err +class TemporalIntermediatePayloadConverter(PayloadConverter, WithSerializationContext): + """Payload converter wrapper for generated Temporal intermediate hooks. + + Values with a ``_temporal_to_intermediate`` method are first converted to + their intermediate value, then encoded by the wrapped payload converter. When + decoding to a type with ``_temporal_from_intermediate``, the wrapped + converter first decodes the payload to the intermediate 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 intermediate 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, TemporalIntermediatePayloadConverter): + return payload_converter + return TemporalIntermediatePayloadConverter(payload_converter) + + def to_payloads( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + """See base class.""" + intermediate_values: list[Any] = [] + for value in values: + to_intermediate = getattr(value, "_temporal_to_intermediate", None) + if to_intermediate is not None: + value = to_intermediate(payload_converter=self._inner_payload_converter) + intermediate_values.append(value) + return self._inner_payload_converter.to_payloads(intermediate_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_intermediate", 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_intermediate(value, payload_converter=self._inner_payload_converter) + if ( + from_intermediate := getattr( + type_hint, "_temporal_from_intermediate", 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/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 74edc66b7..4d329059b 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -246,7 +246,11 @@ 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 = ( + temporalio.converter.TemporalIntermediatePayloadConverter.wrap( + det.payload_converter_class() + ) + ) self._context_free_failure_converter = det.failure_converter_class() workflow_context = temporalio.converter.WorkflowSerializationContext( namespace=det.info.namespace, diff --git a/tests/test_converter.py b/tests/test_converter.py index 10365f9c1..2281513f2 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -44,6 +44,8 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadCodec, + PayloadConverter, + TemporalIntermediatePayloadConverter, decode_search_attributes, encode_search_attribute_values, value_to_type, @@ -254,6 +256,58 @@ def test_binary_proto(): assert decoded == proto +@dataclass +class TemporalIntermediateModel: + value: str + + @classmethod + def _temporal_from_intermediate( + cls, + intermediate: temporalio.api.common.v1.WorkflowExecution, + *, + payload_converter: PayloadConverter | None = None, + ) -> TemporalIntermediateModel: + assert payload_converter is not None + return cls(value=intermediate.workflow_id) + + def _temporal_to_intermediate( + self, *, payload_converter: PayloadConverter | None = None + ) -> temporalio.api.common.v1.WorkflowExecution: + assert payload_converter is not None + return temporalio.api.common.v1.WorkflowExecution( + workflow_id=self.value, + run_id="run-id", + ) + + +class CustomDefaultPayloadConverter(DefaultPayloadConverter): + pass + + +def test_temporal_intermediate_payload_converter_wraps_user_converter(): + data_converter = DataConverter( + payload_converter_class=CustomDefaultPayloadConverter + ) + converter = data_converter.payload_converter + assert isinstance(converter, TemporalIntermediatePayloadConverter) + 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] From 585bc27d62e9dc5f299685289ef239c32bcb215a Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 16 Jul 2026 13:04:40 -0700 Subject: [PATCH 2/2] Support intermediate hooks in system Nexus conversion --- temporalio/converter/_payload_converter.py | 4 +- temporalio/nexus/system/__init__.py | 74 ++++++++++++++++++++-- temporalio/worker/_workflow_instance.py | 4 +- tests/nexus/test_temporal_system_nexus.py | 12 +++- tests/test_converter.py | 8 +-- tests/worker/test_visitor.py | 4 +- 6 files changed, 87 insertions(+), 19 deletions(-) diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index 261ad1033..8cdcc7abe 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -545,7 +545,7 @@ def to_payloads( for value in values: to_intermediate = getattr(value, "_temporal_to_intermediate", None) if to_intermediate is not None: - value = to_intermediate(payload_converter=self._inner_payload_converter) + value = to_intermediate() intermediate_values.append(value) return self._inner_payload_converter.to_payloads(intermediate_values) @@ -570,7 +570,7 @@ def from_payloads( payloads, typing.cast("list[type]", inner_type_hints) ) return [ - from_intermediate(value, payload_converter=self._inner_payload_converter) + from_intermediate(value) if ( from_intermediate := getattr( type_hint, "_temporal_from_intermediate", None diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 21c5a1408..e76756cd7 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -2,22 +2,82 @@ 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 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: + """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 = ( + temporalio.converter.TemporalIntermediatePayloadConverter.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 +93,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 +103,19 @@ 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", + "current_user_payload_converter", "get_payload_converter", "is_system_endpoint", "maybe_visit_payload", "SystemNexusPayloadConverter", + "user_payload_converter_context", ] diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 4d329059b..dd7f4571a 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2093,7 +2093,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/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 2281513f2..0ae2a2464 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -264,16 +264,10 @@ class TemporalIntermediateModel: def _temporal_from_intermediate( cls, intermediate: temporalio.api.common.v1.WorkflowExecution, - *, - payload_converter: PayloadConverter | None = None, ) -> TemporalIntermediateModel: - assert payload_converter is not None return cls(value=intermediate.workflow_id) - def _temporal_to_intermediate( - self, *, payload_converter: PayloadConverter | None = None - ) -> temporalio.api.common.v1.WorkflowExecution: - assert payload_converter is not None + def _temporal_to_intermediate(self) -> temporalio.api.common.v1.WorkflowExecution: return temporalio.api.common.v1.WorkflowExecution( workflow_id=self.value, run_id="run-id", 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")]),