From 003ae3dd4edafa056a8648f17ccffe83c702989a Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 11:05:31 -0700 Subject: [PATCH 1/8] Keep public data converter payload converters unwrapped --- CHANGELOG.md | 4 +- temporalio/activity.py | 2 +- .../openai_agents/_otel_trace_interceptor.py | 2 +- .../openai_agents/_temporal_openai_agents.py | 5 +- .../openai_agents/_trace_interceptor.py | 2 +- .../contrib/workflow_streams/_client.py | 8 +- temporalio/converter/__init__.py | 4 +- temporalio/converter/_data_converter.py | 31 ++++-- temporalio/nexus/system/__init__.py | 2 +- temporalio/testing/_activity.py | 2 +- temporalio/worker/_activity.py | 4 +- temporalio/worker/_nexus.py | 16 ++-- temporalio/worker/_workflow.py | 4 +- tests/test_converter.py | 96 +++++++++++++++---- tests/test_payload_converter_access.py | 64 +++++++++++++ 15 files changed, 198 insertions(+), 48 deletions(-) create mode 100644 tests/test_payload_converter_access.py diff --git a/CHANGELOG.md b/CHANGELOG.md index f016f34ffb..1435259648 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,8 +29,8 @@ to include examples, links to docs, or any other relevant information. ### Fixed -- Current workflow and activity payload converter accessors now return the configured converter - without SDK-internal transfer type conversion. +- `DataConverter.payload_converter` and current workflow and activity payload converter accessors + now return the configured converter without SDK-internal transfer type conversion. ### Security diff --git a/temporalio/activity.py b/temporalio/activity.py index 82678d8a1c..ddf4b73959 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -463,7 +463,7 @@ def payload_converter() -> temporalio.converter.PayloadConverter: This is often used for dynamic activities to convert payloads. """ return _TemporalTransferTypePayloadConverter.unwrap( - _Context.current().payload_converter + _Context.current().payload_converter # raw-payload-converter: Public accessor unwraps the activity context converter. ) diff --git a/temporalio/contrib/openai_agents/_otel_trace_interceptor.py b/temporalio/contrib/openai_agents/_otel_trace_interceptor.py index 63f8f9d830..a186dacb8f 100644 --- a/temporalio/contrib/openai_agents/_otel_trace_interceptor.py +++ b/temporalio/contrib/openai_agents/_otel_trace_interceptor.py @@ -23,7 +23,7 @@ class OTelOpenAIAgentsContextPropagationInterceptor( def __init__( self, otel_id_generator: TemporalIdGenerator, - payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, + payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, # raw-payload-converter: User-configurable tracing header converter. add_temporal_spans: bool = True, ) -> None: """Initialize OTEL-aware context propagation interceptor. diff --git a/temporalio/contrib/openai_agents/_temporal_openai_agents.py b/temporalio/contrib/openai_agents/_temporal_openai_agents.py index 6023ad090c..0b2d1e2c8e 100644 --- a/temporalio/contrib/openai_agents/_temporal_openai_agents.py +++ b/temporalio/contrib/openai_agents/_temporal_openai_agents.py @@ -229,7 +229,10 @@ def _data_converter(converter: DataConverter | None) -> DataConverter: return dataclasses.replace( converter, payload_converter_class=OpenAIPayloadConverter ) - elif not isinstance(converter.payload_converter, OpenAIPayloadConverter): + elif not isinstance( + converter.payload_converter, # raw-payload-converter: Validate the user's converter type. + OpenAIPayloadConverter, + ): raise ValueError( "The payload converter must be of type OpenAIPayloadConverter." ) diff --git a/temporalio/contrib/openai_agents/_trace_interceptor.py b/temporalio/contrib/openai_agents/_trace_interceptor.py index 66297e20b4..3280c99b52 100644 --- a/temporalio/contrib/openai_agents/_trace_interceptor.py +++ b/temporalio/contrib/openai_agents/_trace_interceptor.py @@ -82,7 +82,7 @@ class OpenAIAgentsContextPropagationInterceptor( def __init__( self, - payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, + payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, # raw-payload-converter: User-configurable tracing header converter. add_temporal_spans: bool = True, start_traces: bool = False, ) -> None: diff --git a/temporalio/contrib/workflow_streams/_client.py b/temporalio/contrib/workflow_streams/_client.py index 605bf3f035..dd5bd42d03 100644 --- a/temporalio/contrib/workflow_streams/_client.py +++ b/temporalio/contrib/workflow_streams/_client.py @@ -362,8 +362,12 @@ def _payload_converter(self) -> PayloadConverter: signal/update goes over the wire. See module docstring. """ if self._client is not None: - return self._client.data_converter.payload_converter - return DataConverter.default.payload_converter + return ( + self._client.data_converter.payload_converter # raw-payload-converter: Stream items match workflow.payload_converter(). + ) + return ( + DataConverter.default.payload_converter # raw-payload-converter: Stream items match workflow.payload_converter(). + ) def _encode_buffer(self, entries: list[tuple[str, Any]]) -> list[PublishEntry]: """Convert buffered (topic, value) pairs to wire entries. diff --git a/temporalio/converter/__init__.py b/temporalio/converter/__init__.py index 324b477f25..6f8a27eaed 100644 --- a/temporalio/converter/__init__.py +++ b/temporalio/converter/__init__.py @@ -102,6 +102,8 @@ DataConverter.default = DataConverter() -PayloadConverter.default = DataConverter.default.payload_converter +PayloadConverter.default = ( + DataConverter.default.payload_converter # raw-payload-converter: Public default instance. +) FailureConverter.default = DataConverter.default.failure_converter diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 8604ea1963..62ae2b17b6 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -89,9 +89,15 @@ 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 _get_internal_payload_converter(self) -> PayloadConverter: + """Return the configured converter with SDK transfer type hooks enabled.""" + return _TemporalTransferTypePayloadConverter.wrap( + self.payload_converter # raw-payload-converter: Wrapping boundary. + ) + def _new_payload_converter(self) -> PayloadConverter: """Create a payload converter instance with SDK transfer type hooks enabled.""" return _TemporalTransferTypePayloadConverter.wrap( @@ -113,7 +119,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 +141,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,7 +169,9 @@ 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( @@ -169,11 +179,15 @@ async def decode_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.""" - payload_converter = self.payload_converter + payload_converter = ( + self.payload_converter # raw-payload-converter: Apply context before wrapping. + ) payload_codec = self.payload_codec failure_converter = self.failure_converter external_storage = self.external_storage @@ -188,7 +202,10 @@ def with_context(self, context: SerializationContext) -> Self: if all( new is orig for new, orig in [ - (payload_converter, self.payload_converter), + ( + payload_converter, + self.payload_converter, # raw-payload-converter: Compare the contextual user converter. + ), (payload_codec, self.payload_codec), (failure_converter, self.failure_converter), (external_storage, self.external_storage), diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 1163cae3c4..9c09a41b7a 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -60,7 +60,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 + _current_user_converters().payload_converter # raw-payload-converter: System Nexus model fields use the user converter. ) diff --git a/temporalio/testing/_activity.py b/temporalio/testing/_activity.py index 6f0a9ab5d5..fa5774a1ea 100644 --- a/temporalio/testing/_activity.py +++ b/temporalio/testing/_activity.py @@ -49,7 +49,7 @@ def __init__(self, client: Client | None = None) -> None: self.info = ActivityEnvironment.default_info() self.on_heartbeat: Callable[..., None] = lambda *args: None self.payload_converter = ( - temporalio.converter.DataConverter.default.payload_converter + temporalio.converter.DataConverter.default.payload_converter # raw-payload-converter: Public ActivityEnvironment configuration. ) self.metric_meter = temporalio.common.MetricMeter.noop self._cancelled = False diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index ded3047fcc..8e3b94d141 100644 --- a/temporalio/worker/_activity.py +++ b/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.payload_converter, # raw-payload-converter: Activity context handles wrapping. 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.payload_converter # raw-payload-converter: Activity context handles wrapping. if isinstance(input.executor, concurrent.futures.ThreadPoolExecutor) else self._worker._data_converter.payload_converter_class ) diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index a2f4b8ca75..7dcd56fe65 100644 --- a/temporalio/worker/_nexus.py +++ b/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,10 +601,10 @@ 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.payload_converter, # raw-payload-converter: System Nexus user-model conversion. dc.failure_converter, ) [input] = payload_converter.from_payloads( diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 1b217b4a58..bcf02f71f8 100644 --- a/temporalio/worker/_workflow.py +++ b/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: diff --git a/tests/test_converter.py b/tests/test_converter.py index 4991520965..be65f34fc3 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -48,16 +48,16 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadCodec, + PayloadConverter, + SerializationContext, TransferTypeConverter, + WorkflowSerializationContext, create_payload_validation_error, decode_search_attributes, encode_search_attribute_values, transfer_type_convertible, value_to_type, ) -from temporalio.converter._payload_converter import ( - _TemporalTransferTypePayloadConverter, -) from temporalio.exceptions import ( ApplicationError, FailureError, @@ -388,15 +388,69 @@ class CustomDefaultPayloadConverter(DefaultPayloadConverter): pass -def test_temporal_transfer_type_payload_converter_wraps_user_converter(): +class ContextualTransferPayloadConverter(CustomDefaultPayloadConverter): + def __init__(self) -> None: + super().__init__() + self.context: SerializationContext | None = None + self.encoding_contexts: list[SerializationContext | None] = [] + + def with_context( + self, context: SerializationContext + ) -> ContextualTransferPayloadConverter: + converter = ContextualTransferPayloadConverter() + converter.context = context + return converter + + def to_payloads(self, values: Sequence[Any]) -> list[Payload]: + self.encoding_contexts.append(self.context) + return super().to_payloads(values) + + +async def test_transfer_type_conversion_preserves_user_converter_context(): + original = DataConverter(payload_converter_class=ContextualTransferPayloadConverter) + context = WorkflowSerializationContext(namespace="test", workflow_id="workflow-id") + converter = original.with_context(context) + user_converter = converter.payload_converter + assert isinstance(user_converter, ContextualTransferPayloadConverter) + assert user_converter.context is context + assert user_converter is not original.payload_converter + value = TemporalTransferTypeValue("workflow-id") + + payloads = await converter.encode([value]) + + assert user_converter.encoding_contexts == [context] + assert await converter.decode(payloads, [TemporalTransferTypeValue]) == [value] + assert converter.payload_converter is user_converter + + +async def test_transfer_type_conversion_in_failure_details(): + converter = DataConverter(payload_converter_class=CustomDefaultPayloadConverter) + failure = Failure() + + await converter.encode_failure( + ApplicationError("test failure", TemporalTransferTypeValue("workflow-id")), + failure, + ) + converted = await converter.decode_failure(failure) + + assert isinstance(converted, ApplicationError) + assert converted.details == ( + temporalio.api.common.v1.WorkflowExecution( + workflow_id="workflow-id", run_id="run-id" + ), + ) + assert isinstance(converter.payload_converter, CustomDefaultPayloadConverter) + + +async def test_temporal_transfer_type_payload_converter_wraps_user_converter(): data_converter = DataConverter( payload_converter_class=CustomDefaultPayloadConverter ) - converter = data_converter.payload_converter - assert isinstance(converter, _TemporalTransferTypePayloadConverter) + assert isinstance(data_converter.payload_converter, CustomDefaultPayloadConverter) + assert isinstance(PayloadConverter.default, DefaultPayloadConverter) value = TemporalTransferTypeValue("workflow-id") - payload = converter.to_payload(value) + [payload] = await data_converter.encode([value]) assert payload.metadata["encoding"] == b"json/protobuf" assert ( @@ -404,27 +458,29 @@ def test_temporal_transfer_type_payload_converter_wraps_user_converter(): ) 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, TemporalTransferTypeValue) == value + assert await data_converter.decode([payload], [TemporalTransferTypeValue]) == [ + value + ] - plain_proto_payload = converter.to_payload( - temporalio.api.common.v1.WorkflowExecution(workflow_id="id1", run_id="id2") + [plain_proto_payload] = await data_converter.encode( + [temporalio.api.common.v1.WorkflowExecution(workflow_id="id1", run_id="id2")] ) assert plain_proto_payload.metadata["encoding"] == b"json/protobuf" -def test_temporal_transfer_type_payload_converter_without_transfer_type_hint(): - converter = DataConverter.default.payload_converter +async def test_temporal_transfer_type_payload_converter_without_transfer_type_hint(): + converter = DataConverter.default value = TemporalTransferTypeValueWithoutHint("workflow-id") - payload = converter.to_payload(value) + [payload] = await converter.encode([value]) assert payload.metadata["encoding"] == b"json/protobuf" assert ( payload.metadata["messageType"] == b"temporal.api.common.v1.WorkflowExecution" ) - assert ( - converter.from_payload(payload, TemporalTransferTypeValueWithoutHint) == value - ) + assert await converter.decode( + [payload], [TemporalTransferTypeValueWithoutHint] + ) == [value] @pytest.mark.parametrize( @@ -440,15 +496,15 @@ def test_temporal_transfer_type_payload_converter_without_transfer_type_hint(): ), ], ) -def test_temporal_transfer_type_payload_converter_with_generic_value( +async def test_temporal_transfer_type_payload_converter_with_generic_value( value: TemporalTransferTypeGenericValue[T], type_hint: type[TemporalTransferTypeGenericValue[T]], ): - converter = DataConverter.default.payload_converter + converter = DataConverter.default - payload = converter.to_payload(value) + [payload] = await converter.encode([value]) - assert converter.from_payload(payload, type_hint) == value + assert await converter.decode([payload], [type_hint]) == [value] def test_transfer_type_convertible_rejects_existing_converter(): diff --git a/tests/test_payload_converter_access.py b/tests/test_payload_converter_access.py new file mode 100644 index 0000000000..f59c940cc7 --- /dev/null +++ b/tests/test_payload_converter_access.py @@ -0,0 +1,64 @@ +"""Keep raw payload conversion confined to intentional SDK boundaries.""" + +import ast +import os +from pathlib import Path + +# These expressions belong to public APIs, other converter owners, or restricted value domains. +_ALLOWED_READS = { + ( + "contrib/workflow_streams/_stream.py", + "workflow.payload_converter", + ): "Workflow Streams uses the public current-converter API for user items.", + ( + "contrib/opentelemetry/_interceptor.py", + "self.payload_converter", + ): "Tracing interceptors own a converter for header dictionaries.", + ( + "converter/_search_attributes.py", + "default().payload_converter", + ): "Search attributes only support their own restricted set of primitive values.", + ( + "testing/_activity.py", + "env.payload_converter", + ): "ActivityEnvironment exposes its own configurable converter to activity context.", +} + + +def test_sdk_payload_converter_access() -> None: + root = Path(__file__).resolve().parents[1] / "temporalio" + violations = [] + for directory, subdirectories, filenames in os.walk(root): + subdirectories[:] = [ + name + for name in subdirectories + if name not in {"__pycache__", "target", "sdk-core"} + ] + for filename in filenames: + if not filename.endswith(".py"): + continue + path = Path(directory) / filename + source = path.read_text(encoding="utf-8") + lines = source.splitlines() + tree = ast.parse(source, filename=str(path)) + for node in ast.walk(tree): + if not ( + isinstance(node, ast.Attribute) + and node.attr == "payload_converter" + and isinstance(node.ctx, ast.Load) + ): + continue + relative_path = path.relative_to(root).as_posix() + if (relative_path, ast.unparse(node)) in _ALLOWED_READS: + continue + _, marker, reason = lines[node.lineno - 1].partition( + "# raw-payload-converter:" + ) + if marker and reason.strip(): + continue + violations.append(f"{relative_path}:{node.lineno}: {ast.unparse(node)}") + assert not violations, ( + "Use DataConverter._get_internal_payload_converter() for SDK serialization. " + "Intentional raw access requires '# raw-payload-converter: '.\n" + + "\n".join(sorted(violations)) + ) From 5351f17604d23ef79681bb4c14b115164616c1a8 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 11:15:59 -0700 Subject: [PATCH 2/8] Move activity converter wrapping into the worker --- temporalio/activity.py | 8 ++------ temporalio/worker/_activity.py | 4 ++-- tests/testing/test_activity.py | 2 ++ tests/worker/test_activity.py | 2 ++ 4 files changed, 8 insertions(+), 8 deletions(-) diff --git a/temporalio/activity.py b/temporalio/activity.py index ddf4b73959..651ae62403 100644 --- a/temporalio/activity.py +++ b/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 diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index 8e3b94d141..f72df9cf4b 100644 --- a/temporalio/worker/_activity.py +++ b/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, # raw-payload-converter: Activity context handles wrapping. + 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 # raw-payload-converter: Activity context handles wrapping. + self._worker._data_converter._get_internal_payload_converter() if isinstance(input.executor, concurrent.futures.ThreadPoolExecutor) else self._worker._data_converter.payload_converter_class ) diff --git a/tests/testing/test_activity.py b/tests/testing/test_activity.py index 71ba7f5907..74e14b99ff 100644 --- a/tests/testing/test_activity.py +++ b/tests/testing/test_activity.py @@ -16,6 +16,7 @@ async def test_activity_env_async(): waiting = asyncio.Event() async def do_stuff(param: str) -> str: + assert activity.payload_converter() is env.payload_converter activity.heartbeat(f"param: {param}") # Ensure it works across create_task @@ -57,6 +58,7 @@ def test_activity_env_sync(): properly_cancelled = False def do_stuff(param: str) -> None: + assert activity.payload_converter() is env.payload_converter activity.heartbeat(f"param: {param}") # Ensure it works across thread diff --git a/tests/worker/test_activity.py b/tests/worker/test_activity.py index 64691a93ff..c2add47699 100644 --- a/tests/worker/test_activity.py +++ b/tests/worker/test_activity.py @@ -30,6 +30,7 @@ WorkflowHandle, ) from temporalio.common import RawValue, RetryPolicy +from temporalio.converter import DefaultPayloadConverter from temporalio.exceptions import ( ActivityError, ApplicationError, @@ -1616,6 +1617,7 @@ class DynActivityValue: @activity.defn(dynamic=True) def sync_dyn_activity(args: Sequence[RawValue]) -> DynActivityValue: assert len(args) == 2 + assert isinstance(activity.payload_converter(), DefaultPayloadConverter) arg1 = activity.payload_converter().from_payload(args[0].payload, DynActivityValue) assert isinstance(arg1, DynActivityValue) arg2 = activity.payload_converter().from_payload(args[1].payload, DynActivityValue) From 96d37ae0a034e8464088d483b83110f664525ae3 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 11:21:05 -0700 Subject: [PATCH 3/8] Pass internal payload converters to System Nexus consistently --- temporalio/nexus/system/__init__.py | 2 +- temporalio/worker/_nexus.py | 2 +- tests/nexus/test_temporal_system_nexus.py | 18 +++++++++--------- tests/worker/test_visitor.py | 9 +++++---- 4 files changed, 16 insertions(+), 15 deletions(-) diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 9c09a41b7a..9fe3eb07db 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -174,7 +174,7 @@ def _get_payload_converter( # pyright: ignore[reportUnusedFunction] user_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 the system envelope converter given an internal, wrapped payload converter.""" return _SystemNexusPayloadConverter(user_payload_converter, user_failure_converter) diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index 7dcd56fe65..f68a658d56 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -604,7 +604,7 @@ async def deserialize( 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, # raw-payload-converter: System Nexus user-model conversion. + payload_converter, dc.failure_converter, ) [input] = payload_converter.from_payloads( diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index a97a821dc1..13ed52b1ed 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -435,7 +435,7 @@ def _new_system_nexus_request_payload() -> temporalio.api.common.v1.Payload: request = workflowservice_pb2.SignalWithStartWorkflowExecutionRequest() request.input.payloads.add().CopyFrom(nested_payload) payload = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).to_payload(request) assert payload is not None @@ -461,7 +461,7 @@ async def test_nexus_payload_serializer_decodes_system_input() -> None: headers={"test-header": "header-value"}, ) payload = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).to_payload(request) assert payload is not None @@ -495,7 +495,7 @@ async def test_nexus_payload_serializer_codec_skips_outer_envelope() -> None: namespace="target-namespace", ) payload = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).to_payload(request) assert payload is not None @@ -543,7 +543,7 @@ async def test_schedule_marked_system_nexus_payload_ignores_endpoint() -> None: schedule = completion.successful.commands[0].schedule_nexus_operation data_converter = temporalio.converter.default() decoded = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).from_payload(schedule.input) assert isinstance( @@ -570,7 +570,7 @@ async def test_schedule_unmarked_system_nexus_payload_visits_input_as_regular_pa assert schedule.input.metadata["visited"] == b"true" data_converter = temporalio.converter.default() decoded = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).from_payload(schedule.input) assert isinstance( @@ -693,7 +693,7 @@ def _field_is_repeated(field: FieldDescriptor) -> bool: def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None: data_converter = temporalio.converter.default() payload_converter = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ) proto_value = _build_proto_sample(message_type) @@ -708,7 +708,7 @@ def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None: def test_system_nexus_uses_user_failure_converter() -> None: - payload_converter = temporalio.converter.default().payload_converter + payload_converter = temporalio.converter.default()._get_internal_payload_converter() failure_converter = _TrackingFailureConverter(DefaultPayloadConverter) system_converter = nexus_system._get_payload_converter( payload_converter, failure_converter @@ -734,7 +734,7 @@ def test_system_nexus_uses_user_failure_converter() -> None: def test_system_nexus_payload_converter_restores_user_context_on_failure() -> None: outer_data_converter = temporalio.converter.default() outer_converters = nexus_system._SystemNexusUserConverters( - outer_data_converter.payload_converter, + outer_data_converter._get_internal_payload_converter(), outer_data_converter.failure_converter, ) inner_data_converter = temporalio.converter.DataConverter() @@ -750,7 +750,7 @@ def to_failure( raise ValueError("conversion failed") inner_system_converter = nexus_system._get_payload_converter( - inner_data_converter.payload_converter, + inner_data_converter._get_internal_payload_converter(), RaisingFailureConverter(), ) diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py index 362288f348..0c6789f78b 100644 --- a/tests/worker/test_visitor.py +++ b/tests/worker/test_visitor.py @@ -236,7 +236,7 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: ) data_converter = temporalio.converter.default() payload_converter = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ) system_payload = payload_converter.to_payload(system_request) @@ -281,7 +281,7 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: ) data_converter = temporalio.converter.default() payload_converter = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ) system_payload = payload_converter.to_payload(response) @@ -308,7 +308,7 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: async def test_unknown_system_nexus_payload_raises_application_error(): data_converter = temporalio.converter.default() payload_converter = nexus_system._get_payload_converter( - data_converter.payload_converter, + data_converter._get_internal_payload_converter(), data_converter.failure_converter, ) system_payload = payload_converter.to_payload( @@ -483,7 +483,8 @@ async def _visit(self) -> None: data_converter = temporalio.converter.default() payload_converter = nexus_system._get_payload_converter( - data_converter.payload_converter, data_converter.failure_converter + data_converter._get_internal_payload_converter(), + data_converter.failure_converter, ) system_request = workflowservice_pb2.SignalWithStartWorkflowExecutionRequest( input=Payloads(payloads=[Payload(data=b"workflow-input")]), From db030fcd253d49bde46440e44f0695e2e0ff8a61 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 11:28:41 -0700 Subject: [PATCH 4/8] Remove raw payload converter access check --- temporalio/activity.py | 2 +- .../openai_agents/_otel_trace_interceptor.py | 2 +- .../openai_agents/_temporal_openai_agents.py | 5 +- .../openai_agents/_trace_interceptor.py | 2 +- .../contrib/workflow_streams/_client.py | 8 +-- temporalio/converter/__init__.py | 4 +- temporalio/converter/_data_converter.py | 13 +--- temporalio/nexus/system/__init__.py | 2 +- temporalio/testing/_activity.py | 2 +- tests/test_payload_converter_access.py | 64 ------------------- 10 files changed, 12 insertions(+), 92 deletions(-) delete mode 100644 tests/test_payload_converter_access.py diff --git a/temporalio/activity.py b/temporalio/activity.py index 651ae62403..768c8d108b 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -459,7 +459,7 @@ def payload_converter() -> temporalio.converter.PayloadConverter: This is often used for dynamic activities to convert payloads. """ return _TemporalTransferTypePayloadConverter.unwrap( - _Context.current().payload_converter # raw-payload-converter: Public accessor unwraps the activity context converter. + _Context.current().payload_converter ) diff --git a/temporalio/contrib/openai_agents/_otel_trace_interceptor.py b/temporalio/contrib/openai_agents/_otel_trace_interceptor.py index a186dacb8f..63f8f9d830 100644 --- a/temporalio/contrib/openai_agents/_otel_trace_interceptor.py +++ b/temporalio/contrib/openai_agents/_otel_trace_interceptor.py @@ -23,7 +23,7 @@ class OTelOpenAIAgentsContextPropagationInterceptor( def __init__( self, otel_id_generator: TemporalIdGenerator, - payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, # raw-payload-converter: User-configurable tracing header converter. + payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, add_temporal_spans: bool = True, ) -> None: """Initialize OTEL-aware context propagation interceptor. diff --git a/temporalio/contrib/openai_agents/_temporal_openai_agents.py b/temporalio/contrib/openai_agents/_temporal_openai_agents.py index 0b2d1e2c8e..6023ad090c 100644 --- a/temporalio/contrib/openai_agents/_temporal_openai_agents.py +++ b/temporalio/contrib/openai_agents/_temporal_openai_agents.py @@ -229,10 +229,7 @@ def _data_converter(converter: DataConverter | None) -> DataConverter: return dataclasses.replace( converter, payload_converter_class=OpenAIPayloadConverter ) - elif not isinstance( - converter.payload_converter, # raw-payload-converter: Validate the user's converter type. - OpenAIPayloadConverter, - ): + elif not isinstance(converter.payload_converter, OpenAIPayloadConverter): raise ValueError( "The payload converter must be of type OpenAIPayloadConverter." ) diff --git a/temporalio/contrib/openai_agents/_trace_interceptor.py b/temporalio/contrib/openai_agents/_trace_interceptor.py index 3280c99b52..66297e20b4 100644 --- a/temporalio/contrib/openai_agents/_trace_interceptor.py +++ b/temporalio/contrib/openai_agents/_trace_interceptor.py @@ -82,7 +82,7 @@ class OpenAIAgentsContextPropagationInterceptor( def __init__( self, - payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, # raw-payload-converter: User-configurable tracing header converter. + payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter, add_temporal_spans: bool = True, start_traces: bool = False, ) -> None: diff --git a/temporalio/contrib/workflow_streams/_client.py b/temporalio/contrib/workflow_streams/_client.py index dd5bd42d03..605bf3f035 100644 --- a/temporalio/contrib/workflow_streams/_client.py +++ b/temporalio/contrib/workflow_streams/_client.py @@ -362,12 +362,8 @@ def _payload_converter(self) -> PayloadConverter: signal/update goes over the wire. See module docstring. """ if self._client is not None: - return ( - self._client.data_converter.payload_converter # raw-payload-converter: Stream items match workflow.payload_converter(). - ) - return ( - DataConverter.default.payload_converter # raw-payload-converter: Stream items match workflow.payload_converter(). - ) + return self._client.data_converter.payload_converter + return DataConverter.default.payload_converter def _encode_buffer(self, entries: list[tuple[str, Any]]) -> list[PublishEntry]: """Convert buffered (topic, value) pairs to wire entries. diff --git a/temporalio/converter/__init__.py b/temporalio/converter/__init__.py index 6f8a27eaed..324b477f25 100644 --- a/temporalio/converter/__init__.py +++ b/temporalio/converter/__init__.py @@ -102,8 +102,6 @@ DataConverter.default = DataConverter() -PayloadConverter.default = ( - DataConverter.default.payload_converter # raw-payload-converter: Public default instance. -) +PayloadConverter.default = DataConverter.default.payload_converter FailureConverter.default = DataConverter.default.failure_converter diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 62ae2b17b6..d30cb3b2b9 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -94,9 +94,7 @@ def __post_init__(self) -> None: # noqa: D105 def _get_internal_payload_converter(self) -> PayloadConverter: """Return the configured converter with SDK transfer type hooks enabled.""" - return _TemporalTransferTypePayloadConverter.wrap( - self.payload_converter # raw-payload-converter: Wrapping boundary. - ) + return _TemporalTransferTypePayloadConverter.wrap(self.payload_converter) def _new_payload_converter(self) -> PayloadConverter: """Create a payload converter instance with SDK transfer type hooks enabled.""" @@ -185,9 +183,7 @@ async def decode_failure( def with_context(self, context: SerializationContext) -> Self: """Return an instance with context set on the component converters.""" - payload_converter = ( - self.payload_converter # raw-payload-converter: Apply context before wrapping. - ) + payload_converter = self.payload_converter payload_codec = self.payload_codec failure_converter = self.failure_converter external_storage = self.external_storage @@ -202,10 +198,7 @@ def with_context(self, context: SerializationContext) -> Self: if all( new is orig for new, orig in [ - ( - payload_converter, - self.payload_converter, # raw-payload-converter: Compare the contextual user converter. - ), + (payload_converter, self.payload_converter), (payload_codec, self.payload_codec), (failure_converter, self.failure_converter), (external_storage, self.external_storage), diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 9fe3eb07db..31dadac8f1 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -60,7 +60,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 # raw-payload-converter: System Nexus model fields use the user converter. + _current_user_converters().payload_converter ) diff --git a/temporalio/testing/_activity.py b/temporalio/testing/_activity.py index fa5774a1ea..6f0a9ab5d5 100644 --- a/temporalio/testing/_activity.py +++ b/temporalio/testing/_activity.py @@ -49,7 +49,7 @@ def __init__(self, client: Client | None = None) -> None: self.info = ActivityEnvironment.default_info() self.on_heartbeat: Callable[..., None] = lambda *args: None self.payload_converter = ( - temporalio.converter.DataConverter.default.payload_converter # raw-payload-converter: Public ActivityEnvironment configuration. + temporalio.converter.DataConverter.default.payload_converter ) self.metric_meter = temporalio.common.MetricMeter.noop self._cancelled = False diff --git a/tests/test_payload_converter_access.py b/tests/test_payload_converter_access.py deleted file mode 100644 index f59c940cc7..0000000000 --- a/tests/test_payload_converter_access.py +++ /dev/null @@ -1,64 +0,0 @@ -"""Keep raw payload conversion confined to intentional SDK boundaries.""" - -import ast -import os -from pathlib import Path - -# These expressions belong to public APIs, other converter owners, or restricted value domains. -_ALLOWED_READS = { - ( - "contrib/workflow_streams/_stream.py", - "workflow.payload_converter", - ): "Workflow Streams uses the public current-converter API for user items.", - ( - "contrib/opentelemetry/_interceptor.py", - "self.payload_converter", - ): "Tracing interceptors own a converter for header dictionaries.", - ( - "converter/_search_attributes.py", - "default().payload_converter", - ): "Search attributes only support their own restricted set of primitive values.", - ( - "testing/_activity.py", - "env.payload_converter", - ): "ActivityEnvironment exposes its own configurable converter to activity context.", -} - - -def test_sdk_payload_converter_access() -> None: - root = Path(__file__).resolve().parents[1] / "temporalio" - violations = [] - for directory, subdirectories, filenames in os.walk(root): - subdirectories[:] = [ - name - for name in subdirectories - if name not in {"__pycache__", "target", "sdk-core"} - ] - for filename in filenames: - if not filename.endswith(".py"): - continue - path = Path(directory) / filename - source = path.read_text(encoding="utf-8") - lines = source.splitlines() - tree = ast.parse(source, filename=str(path)) - for node in ast.walk(tree): - if not ( - isinstance(node, ast.Attribute) - and node.attr == "payload_converter" - and isinstance(node.ctx, ast.Load) - ): - continue - relative_path = path.relative_to(root).as_posix() - if (relative_path, ast.unparse(node)) in _ALLOWED_READS: - continue - _, marker, reason = lines[node.lineno - 1].partition( - "# raw-payload-converter:" - ) - if marker and reason.strip(): - continue - violations.append(f"{relative_path}:{node.lineno}: {ast.unparse(node)}") - assert not violations, ( - "Use DataConverter._get_internal_payload_converter() for SDK serialization. " - "Intentional raw access requires '# raw-payload-converter: '.\n" - + "\n".join(sorted(violations)) - ) From 30bb96dfa766ea7169bafe477142d67c7405d1ad Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 11:38:43 -0700 Subject: [PATCH 5/8] Expand transfer type integration coverage --- tests/test_transfer_types.py | 434 +++++++++++++++++++++++++++++++++++ 1 file changed, 434 insertions(+) create mode 100644 tests/test_transfer_types.py diff --git a/tests/test_transfer_types.py b/tests/test_transfer_types.py new file mode 100644 index 0000000000..253334ae57 --- /dev/null +++ b/tests/test_transfer_types.py @@ -0,0 +1,434 @@ +from __future__ import annotations + +import concurrent.futures +import uuid +from datetime import timedelta + +import nexusrpc +import nexusrpc.handler +import pytest + +import temporalio.activity as activity +import temporalio.client +import temporalio.common +import temporalio.converter +import temporalio.exceptions +import temporalio.nexus +import temporalio.testing +import temporalio.worker +import temporalio.workflow as workflow +from tests.helpers import new_worker +from tests.helpers.nexus import make_nexus_endpoint_name + + +class TransferValueConverter( + temporalio.converter.TransferTypeConverter["TransferValue", str] +): + transfer_type = str + + def to_transfer_type(self, value: TransferValue) -> str: + return f"transfer:{value.text}" + + def from_transfer_type( + self, value: str, type_hint: type[TransferValue] + ) -> TransferValue: + assert type_hint is TransferValue + assert value.startswith("transfer:") + return TransferValue(value.removeprefix("transfer:")) + + +@temporalio.converter.transfer_type_convertible(TransferValueConverter) +class TransferValue: + # A non-dataclass ensures missing transfer conversion cannot silently fall + # back to the default converter's dataclass JSON support. + def __init__(self, text: str) -> None: + self.text = text + + def append(self, step: str) -> TransferValue: + return TransferValue(f"{self.text}/{step}") + + +def test_transfer_value_requires_transfer_conversion(): + with pytest.raises(TypeError, match="not JSON serializable"): + temporalio.converter.DefaultPayloadConverter().to_payload( + TransferValue("input") + ) + + +@workflow.defn +class TransferWorkflow: + @workflow.run + async def run(self, value: TransferValue, operation: str) -> TransferValue: + if operation == "child": + result = await workflow.execute_child_workflow( + TransferWorkflow.run, + args=[value.append("child-input"), "return"], + id=f"{workflow.info().workflow_id}-child", + ) + return result.append("parent") + if operation == "continue-as-new": + workflow.continue_as_new(args=[value.append("continued"), "return"]) + assert operation == "return" + return value.append("workflow") + + +@pytest.mark.parametrize( + "operation, expected", + [ + ("return", "input/workflow"), + ("child", "input/child-input/workflow/parent"), + ("continue-as-new", "input/continued/workflow"), + ], +) +async def test_transfer_types_workflow( + client: temporalio.client.Client, operation: str, expected: str +): + async with new_worker( + client, TransferWorkflow, workflow_failure_exception_types=[Exception] + ) as worker: + result = await client.execute_workflow( + TransferWorkflow.run, + args=[TransferValue("input"), operation], + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=30), + ) + assert result.text == expected + + +@activity.defn +async def transfer_activity(value: TransferValue) -> TransferValue: + return value.append("activity") + + +@activity.defn +def transfer_sync_activity(value: TransferValue) -> TransferValue: + return value.append("activity") + + +@workflow.defn +class TransferActivityWorkflow: + @workflow.run + async def run( + self, value: TransferValue, local: bool, synchronous: bool + ) -> TransferValue: + activity_fn = transfer_sync_activity if synchronous else transfer_activity + if local: + result = await workflow.execute_local_activity( + activity_fn, + value.append("activity-input"), + start_to_close_timeout=timedelta(seconds=10), + retry_policy=temporalio.common.RetryPolicy(maximum_attempts=1), + ) + else: + result = await workflow.execute_activity( + activity_fn, + value.append("activity-input"), + start_to_close_timeout=timedelta(seconds=10), + retry_policy=temporalio.common.RetryPolicy(maximum_attempts=1), + ) + assert isinstance(result, TransferValue) + return result.append("workflow") + + +@pytest.mark.parametrize("local", [False, True], ids=["remote", "local"]) +@pytest.mark.parametrize("execution", ["async", "thread", "process"]) +async def test_transfer_types_activity( + client: temporalio.client.Client, + shared_state_manager: temporalio.worker.SharedStateManager, + local: bool, + execution: str, +): + executor_cls = ( + concurrent.futures.ProcessPoolExecutor + if execution == "process" + else concurrent.futures.ThreadPoolExecutor + ) + with executor_cls(max_workers=2) as executor: + async with new_worker( + client, + TransferActivityWorkflow, + activities=[transfer_activity, transfer_sync_activity], + activity_executor=executor, + shared_state_manager=shared_state_manager, + workflow_failure_exception_types=[Exception], + ) as worker: + result = await client.execute_workflow( + TransferActivityWorkflow.run, + args=[TransferValue("input"), local, execution != "async"], + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=30), + ) + assert result.text == "input/activity-input/activity/workflow" + + +@workflow.defn +class TransferMessagesWorkflow: + def __init__(self) -> None: + self.value = TransferValue("unset") + self.finished = False + + @workflow.run + async def run(self, value: TransferValue) -> TransferValue: + await workflow.wait_condition(lambda: self.finished) + return self.value.append(value.text) + + @workflow.signal + def signal(self, value: TransferValue) -> None: + self.value = value.append("signal") + + @workflow.query + def query(self, value: TransferValue) -> TransferValue: + return self.value.append(value.text) + + @workflow.update + async def update(self, value: TransferValue) -> TransferValue: + self.value = value.append("update") + return self.value.append("result") + + @update.validator + def validate_update(self, value: TransferValue) -> None: + assert value.text + + @workflow.signal + def finish(self) -> None: + self.finished = True + + +@pytest.mark.parametrize( + "start", ["workflow", "signal-with-start", "update-with-start"] +) +async def test_transfer_types_messages(client: temporalio.client.Client, start: str): + async with new_worker( + client, TransferMessagesWorkflow, workflow_failure_exception_types=[Exception] + ) as worker: + if start == "update-with-start": + start_op = temporalio.client.WithStartWorkflowOperation( + TransferMessagesWorkflow.run, + TransferValue("workflow-input"), + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=30), + id_conflict_policy=temporalio.common.WorkflowIDConflictPolicy.FAIL, + ) + result = await client.execute_update_with_start_workflow( + TransferMessagesWorkflow.update, + TransferValue("start-update-input"), + start_workflow_operation=start_op, + ) + assert result.text == "start-update-input/update/result" + handle = await start_op.workflow_handle() + else: + handle = await client.start_workflow( + TransferMessagesWorkflow.run, + TransferValue("workflow-input"), + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=30), + start_signal="signal" if start == "signal-with-start" else None, + start_signal_args=[TransferValue("signal-input")] + if start == "signal-with-start" + else [], + ) + if start != "signal-with-start": + await handle.signal( + TransferMessagesWorkflow.signal, TransferValue("signal-input") + ) + result = await handle.query( + TransferMessagesWorkflow.query, TransferValue("query-input") + ) + assert result.text == "signal-input/signal/query-input" + + result = await handle.execute_update( + TransferMessagesWorkflow.update, TransferValue("update-input") + ) + assert result.text == "update-input/update/result" + await handle.signal(TransferMessagesWorkflow.finish) + assert (await handle.result()).text == "update-input/update/workflow-input" + + +@workflow.defn +class TransferSignalWorkflow: + @workflow.run + async def run(self, value: TransferValue, external: bool) -> TransferValue: + child = await workflow.start_child_workflow( + TransferMessagesWorkflow.run, + value.append("child-input"), + id=f"{workflow.info().workflow_id}-child", + ) + if external: + handle = workflow.get_external_workflow_handle_for( + TransferMessagesWorkflow.run, child.id + ) + await handle.signal( + TransferMessagesWorkflow.signal, value.append("external") + ) + await handle.signal(TransferMessagesWorkflow.finish) + else: + await child.signal(TransferMessagesWorkflow.signal, value.append("child")) + await child.signal(TransferMessagesWorkflow.finish) + return (await child).append("parent") + + +@pytest.mark.parametrize("external", [False, True], ids=["child", "external"]) +async def test_transfer_types_workflow_signal( + client: temporalio.client.Client, external: bool +): + async with new_worker( + client, + TransferSignalWorkflow, + TransferMessagesWorkflow, + workflow_failure_exception_types=[Exception], + ) as worker: + result = await client.execute_workflow( + TransferSignalWorkflow.run, + args=[TransferValue("input"), external], + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=30), + ) + signal_target = "external" if external else "child" + assert result.text == f"input/{signal_target}/signal/input/child-input/parent" + + +@activity.defn +async def transfer_failure_activity(value: TransferValue) -> None: + raise temporalio.exceptions.ApplicationError("failure", value.append("activity")) + + +@workflow.defn +class TransferFailureWorkflow: + @workflow.run + async def run(self, value: TransferValue, operation: str) -> None: + if operation == "workflow": + raise temporalio.exceptions.ApplicationError( + "failure", value.append("workflow") + ) + try: + if operation == "local-activity": + await workflow.execute_local_activity( + transfer_failure_activity, + value, + start_to_close_timeout=timedelta(seconds=10), + retry_policy=temporalio.common.RetryPolicy(maximum_attempts=1), + ) + else: + await workflow.execute_activity( + transfer_failure_activity, + value, + start_to_close_timeout=timedelta(seconds=10), + retry_policy=temporalio.common.RetryPolicy(maximum_attempts=1), + ) + except temporalio.exceptions.FailureError as err: + cause = ( + err.cause + if isinstance(err, temporalio.exceptions.ActivityError) + else err + ) + assert isinstance(cause, temporalio.exceptions.ApplicationError) + # Failure details have no type hints, so decoding yields the transfer type. + assert cause.details == ("transfer:input/activity",) + raise + + +@pytest.mark.parametrize("operation", ["workflow", "activity", "local-activity"]) +async def test_transfer_types_failure_details( + client: temporalio.client.Client, operation: str +): + async with new_worker( + client, + TransferFailureWorkflow, + activities=[transfer_failure_activity], + workflow_failure_exception_types=[Exception], + ) as worker: + with pytest.raises(temporalio.client.WorkflowFailureError) as err: + await client.execute_workflow( + TransferFailureWorkflow.run, + args=[TransferValue("input"), operation], + id=str(uuid.uuid4()), + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=30), + ) + cause: BaseException | None = err.value.cause + if isinstance(cause, temporalio.exceptions.ActivityError): + cause = cause.cause + assert isinstance(cause, temporalio.exceptions.ApplicationError) + source = "workflow" if operation == "workflow" else "activity" + assert cause.details == (f"transfer:input/{source}",) + + +@nexusrpc.service +class TransferService: + sync: nexusrpc.Operation[TransferValue, TransferValue] + async_: nexusrpc.Operation[TransferValue, TransferValue] + + +@nexusrpc.handler.service_handler(service=TransferService) +class TransferServiceHandler: + @nexusrpc.handler.sync_operation + async def sync( + self, _ctx: nexusrpc.handler.StartOperationContext, input: TransferValue + ) -> TransferValue: + return input.append("nexus-sync") + + @temporalio.nexus.workflow_run_operation + async def async_( + self, ctx: temporalio.nexus.WorkflowRunOperationContext, input: TransferValue + ) -> temporalio.nexus.WorkflowHandle[TransferValue]: + return await ctx.start_workflow( + TransferWorkflow.run, + args=[input.append("nexus-async"), "return"], + id=str(uuid.uuid4()), + ) + + +@workflow.defn +class TransferNexusWorkflow: + @workflow.run + async def run( + self, value: TransferValue, endpoint: str, synchronous: bool + ) -> TransferValue: + client = workflow.create_nexus_client( + service=TransferService, endpoint=endpoint + ) + result = await client.execute_operation( + TransferService.sync if synchronous else TransferService.async_, + value.append("nexus-input"), + schedule_to_close_timeout=timedelta(seconds=20), + ) + return result.append("caller") + + +# Cloud CI credentials cannot manage Nexus endpoints. +@pytest.mark.requires_local_server +@pytest.mark.parametrize("synchronous", [False, True], ids=["async", "sync"]) +async def test_transfer_types_nexus( + env: temporalio.testing.WorkflowEnvironment, synchronous: bool +): + if env.supports_time_skipping: + pytest.skip("Nexus requires the dev server") + task_queue = str(uuid.uuid4()) + endpoint = await env.create_nexus_endpoint( + make_nexus_endpoint_name(task_queue), task_queue + ) + try: + async with new_worker( + env.client, + TransferNexusWorkflow, + TransferWorkflow, + task_queue=task_queue, + nexus_service_handlers=[TransferServiceHandler()], + workflow_failure_exception_types=[Exception], + ): + result = await env.client.execute_workflow( + TransferNexusWorkflow.run, + args=[TransferValue("input"), endpoint.spec.name, synchronous], + id=str(uuid.uuid4()), + task_queue=task_queue, + execution_timeout=timedelta(seconds=30), + ) + expected = "nexus-sync" if synchronous else "nexus-async/workflow" + assert result.text == f"input/nexus-input/{expected}/caller" + finally: + await env.delete_nexus_endpoint(endpoint) From c746083b77d279d1b7416d51051c54cd8a92f7ee Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 13:11:59 -0700 Subject: [PATCH 6/8] Consolidate transfer type tests and clarify internal factory --- temporalio/converter/_data_converter.py | 2 +- temporalio/worker/_workflow.py | 2 +- temporalio/worker/workflow_sandbox/_runner.py | 2 +- tests/test_converter.py | 238 ---------------- tests/test_transfer_types.py | 258 +++++++++++++++++- 5 files changed, 259 insertions(+), 243 deletions(-) diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index d30cb3b2b9..3c2514ca4a 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -96,7 +96,7 @@ def _get_internal_payload_converter(self) -> PayloadConverter: """Return the configured converter with SDK transfer type hooks enabled.""" return _TemporalTransferTypePayloadConverter.wrap(self.payload_converter) - def _new_payload_converter(self) -> PayloadConverter: + def _new_internal_payload_converter(self) -> PayloadConverter: """Create a payload converter instance with SDK transfer type hooks enabled.""" return _TemporalTransferTypePayloadConverter.wrap( self.payload_converter_class() diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index bcf02f71f8..c9bc7ab34c 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -727,7 +727,7 @@ def _create_workflow_instance( # Create instance from details det = WorkflowInstanceDetails( - payload_converter_factory=self._data_converter._new_payload_converter, + 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, diff --git a/temporalio/worker/workflow_sandbox/_runner.py b/temporalio/worker/workflow_sandbox/_runner.py index 7f06bfcd6c..37c8e0744f 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_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, diff --git a/tests/test_converter.py b/tests/test_converter.py index be65f34fc3..4c64e25ba4 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -6,7 +6,6 @@ import logging import sys import traceback -import typing from collections import deque from collections.abc import Iterable, Mapping, MutableMapping, Sequence from dataclasses import dataclass @@ -15,11 +14,8 @@ from typing import ( Any, Dict, # type:ignore[reportDeprecated] - Generic, Literal, NewType, - TypeVar, - cast, get_args, get_type_hints, ) @@ -48,14 +44,9 @@ JSONTypeConverter, JSONTypeConverterUnhandled, PayloadCodec, - PayloadConverter, - SerializationContext, - TransferTypeConverter, - WorkflowSerializationContext, create_payload_validation_error, decode_search_attributes, encode_search_attribute_values, - transfer_type_convertible, value_to_type, ) from temporalio.exceptions import ( @@ -285,235 +276,6 @@ def test_binary_proto(): assert decoded == proto -class TemporalTransferTypeValueConverter( - TransferTypeConverter[ - "TemporalTransferTypeValue", - temporalio.api.common.v1.WorkflowExecution, - ] -): - transfer_type = temporalio.api.common.v1.WorkflowExecution - - def to_transfer_type( - self, value: TemporalTransferTypeValue - ) -> temporalio.api.common.v1.WorkflowExecution: - return temporalio.api.common.v1.WorkflowExecution( - workflow_id=value.value, - run_id="run-id", - ) - - def from_transfer_type( - self, - value: temporalio.api.common.v1.WorkflowExecution, - type_hint: type[TemporalTransferTypeValue], - ) -> TemporalTransferTypeValue: - return TemporalTransferTypeValue(value=value.workflow_id) - - -@transfer_type_convertible(TemporalTransferTypeValueConverter) -@dataclass -class TemporalTransferTypeValue: - value: str - - -class TemporalTransferTypeValueWithoutHintConverter( - TransferTypeConverter[ - "TemporalTransferTypeValueWithoutHint", - temporalio.api.common.v1.WorkflowExecution, - ] -): - def to_transfer_type( - self, value: TemporalTransferTypeValueWithoutHint - ) -> temporalio.api.common.v1.WorkflowExecution: - return temporalio.api.common.v1.WorkflowExecution( - workflow_id=value.value, - run_id="run-id", - ) - - def from_transfer_type( - self, - value: temporalio.api.common.v1.WorkflowExecution, - type_hint: type[TemporalTransferTypeValueWithoutHint], - ) -> TemporalTransferTypeValueWithoutHint: - return TemporalTransferTypeValueWithoutHint(value=value.workflow_id) - - -@transfer_type_convertible(TemporalTransferTypeValueWithoutHintConverter) -@dataclass -class TemporalTransferTypeValueWithoutHint: - value: str - - -T = TypeVar("T") - - -@dataclass -class TemporalTransferTypeGenericValue(Generic[T]): - value: T - - -class TemporalTransferTypeGenericValueConverter( - TransferTypeConverter[ - TemporalTransferTypeGenericValue[T], - temporalio.api.common.v1.WorkflowExecution, - ] -): - transfer_type = temporalio.api.common.v1.WorkflowExecution - - def to_transfer_type( - self, value: TemporalTransferTypeGenericValue[T] - ) -> temporalio.api.common.v1.WorkflowExecution: - return temporalio.api.common.v1.WorkflowExecution( - workflow_id=str(value.value), - run_id="run-id", - ) - - def from_transfer_type( - self, - value: temporalio.api.common.v1.WorkflowExecution, - type_hint: type[TemporalTransferTypeGenericValue[T]], - ) -> TemporalTransferTypeGenericValue[T]: - converted_value: str | int = value.workflow_id - if typing.get_args(type_hint)[0] is int: - converted_value = int(converted_value) - return TemporalTransferTypeGenericValue(value=cast(T, converted_value)) - - -# Register after both classes are defined so the generic type can be resolved. -transfer_type_convertible(TemporalTransferTypeGenericValueConverter)( - TemporalTransferTypeGenericValue -) - - -class CustomDefaultPayloadConverter(DefaultPayloadConverter): - pass - - -class ContextualTransferPayloadConverter(CustomDefaultPayloadConverter): - def __init__(self) -> None: - super().__init__() - self.context: SerializationContext | None = None - self.encoding_contexts: list[SerializationContext | None] = [] - - def with_context( - self, context: SerializationContext - ) -> ContextualTransferPayloadConverter: - converter = ContextualTransferPayloadConverter() - converter.context = context - return converter - - def to_payloads(self, values: Sequence[Any]) -> list[Payload]: - self.encoding_contexts.append(self.context) - return super().to_payloads(values) - - -async def test_transfer_type_conversion_preserves_user_converter_context(): - original = DataConverter(payload_converter_class=ContextualTransferPayloadConverter) - context = WorkflowSerializationContext(namespace="test", workflow_id="workflow-id") - converter = original.with_context(context) - user_converter = converter.payload_converter - assert isinstance(user_converter, ContextualTransferPayloadConverter) - assert user_converter.context is context - assert user_converter is not original.payload_converter - value = TemporalTransferTypeValue("workflow-id") - - payloads = await converter.encode([value]) - - assert user_converter.encoding_contexts == [context] - assert await converter.decode(payloads, [TemporalTransferTypeValue]) == [value] - assert converter.payload_converter is user_converter - - -async def test_transfer_type_conversion_in_failure_details(): - converter = DataConverter(payload_converter_class=CustomDefaultPayloadConverter) - failure = Failure() - - await converter.encode_failure( - ApplicationError("test failure", TemporalTransferTypeValue("workflow-id")), - failure, - ) - converted = await converter.decode_failure(failure) - - assert isinstance(converted, ApplicationError) - assert converted.details == ( - temporalio.api.common.v1.WorkflowExecution( - workflow_id="workflow-id", run_id="run-id" - ), - ) - assert isinstance(converter.payload_converter, CustomDefaultPayloadConverter) - - -async def test_temporal_transfer_type_payload_converter_wraps_user_converter(): - data_converter = DataConverter( - payload_converter_class=CustomDefaultPayloadConverter - ) - assert isinstance(data_converter.payload_converter, CustomDefaultPayloadConverter) - assert isinstance(PayloadConverter.default, DefaultPayloadConverter) - value = TemporalTransferTypeValue("workflow-id") - - [payload] = await data_converter.encode([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 await data_converter.decode([payload], [TemporalTransferTypeValue]) == [ - value - ] - - [plain_proto_payload] = await data_converter.encode( - [temporalio.api.common.v1.WorkflowExecution(workflow_id="id1", run_id="id2")] - ) - assert plain_proto_payload.metadata["encoding"] == b"json/protobuf" - - -async def test_temporal_transfer_type_payload_converter_without_transfer_type_hint(): - converter = DataConverter.default - value = TemporalTransferTypeValueWithoutHint("workflow-id") - - [payload] = await converter.encode([value]) - - assert payload.metadata["encoding"] == b"json/protobuf" - assert ( - payload.metadata["messageType"] == b"temporal.api.common.v1.WorkflowExecution" - ) - assert await converter.decode( - [payload], [TemporalTransferTypeValueWithoutHint] - ) == [value] - - -@pytest.mark.parametrize( - ("value", "type_hint"), - [ - ( - TemporalTransferTypeGenericValue("workflow-id"), - TemporalTransferTypeGenericValue[str], - ), - ( - TemporalTransferTypeGenericValue(123), - TemporalTransferTypeGenericValue[int], - ), - ], -) -async def test_temporal_transfer_type_payload_converter_with_generic_value( - value: TemporalTransferTypeGenericValue[T], - type_hint: type[TemporalTransferTypeGenericValue[T]], -): - converter = DataConverter.default - - [payload] = await converter.encode([value]) - - assert await converter.decode([payload], [type_hint]) == [value] - - -def test_transfer_type_convertible_rejects_existing_converter(): - with pytest.raises(TypeError, match="already has a transfer type converter"): - transfer_type_convertible(TemporalTransferTypeValueConverter)( - TemporalTransferTypeValue - ) - - 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/test_transfer_types.py b/tests/test_transfer_types.py index 253334ae57..7b7ca33fcd 100644 --- a/tests/test_transfer_types.py +++ b/tests/test_transfer_types.py @@ -1,14 +1,20 @@ from __future__ import annotations import concurrent.futures +import typing import uuid +from collections.abc import Sequence +from dataclasses import dataclass from datetime import timedelta +from typing import Any, Generic, TypeVar, cast import nexusrpc import nexusrpc.handler import pytest import temporalio.activity as activity +import temporalio.api.common.v1 +import temporalio.api.failure.v1 import temporalio.client import temporalio.common import temporalio.converter @@ -21,6 +27,252 @@ from tests.helpers.nexus import make_nexus_endpoint_name +class TemporalTransferTypeValueConverter( + temporalio.converter.TransferTypeConverter[ + "TemporalTransferTypeValue", + temporalio.api.common.v1.WorkflowExecution, + ] +): + transfer_type = temporalio.api.common.v1.WorkflowExecution + + def to_transfer_type( + self, value: TemporalTransferTypeValue + ) -> temporalio.api.common.v1.WorkflowExecution: + return temporalio.api.common.v1.WorkflowExecution( + workflow_id=value.value, + run_id="run-id", + ) + + def from_transfer_type( + self, + value: temporalio.api.common.v1.WorkflowExecution, + type_hint: type[TemporalTransferTypeValue], + ) -> TemporalTransferTypeValue: + return TemporalTransferTypeValue(value=value.workflow_id) + + +@temporalio.converter.transfer_type_convertible(TemporalTransferTypeValueConverter) +@dataclass +class TemporalTransferTypeValue: + value: str + + +class TemporalTransferTypeValueWithoutHintConverter( + temporalio.converter.TransferTypeConverter[ + "TemporalTransferTypeValueWithoutHint", + temporalio.api.common.v1.WorkflowExecution, + ] +): + def to_transfer_type( + self, value: TemporalTransferTypeValueWithoutHint + ) -> temporalio.api.common.v1.WorkflowExecution: + return temporalio.api.common.v1.WorkflowExecution( + workflow_id=value.value, + run_id="run-id", + ) + + def from_transfer_type( + self, + value: temporalio.api.common.v1.WorkflowExecution, + type_hint: type[TemporalTransferTypeValueWithoutHint], + ) -> TemporalTransferTypeValueWithoutHint: + return TemporalTransferTypeValueWithoutHint(value=value.workflow_id) + + +@temporalio.converter.transfer_type_convertible( + TemporalTransferTypeValueWithoutHintConverter +) +@dataclass +class TemporalTransferTypeValueWithoutHint: + value: str + + +T = TypeVar("T") + + +@dataclass +class TemporalTransferTypeGenericValue(Generic[T]): + value: T + + +class TemporalTransferTypeGenericValueConverter( + temporalio.converter.TransferTypeConverter[ + TemporalTransferTypeGenericValue[T], + temporalio.api.common.v1.WorkflowExecution, + ] +): + transfer_type = temporalio.api.common.v1.WorkflowExecution + + def to_transfer_type( + self, value: TemporalTransferTypeGenericValue[T] + ) -> temporalio.api.common.v1.WorkflowExecution: + return temporalio.api.common.v1.WorkflowExecution( + workflow_id=str(value.value), + run_id="run-id", + ) + + def from_transfer_type( + self, + value: temporalio.api.common.v1.WorkflowExecution, + type_hint: type[TemporalTransferTypeGenericValue[T]], + ) -> TemporalTransferTypeGenericValue[T]: + converted_value: str | int = value.workflow_id + if typing.get_args(type_hint)[0] is int: + converted_value = int(converted_value) + return TemporalTransferTypeGenericValue(value=cast(T, converted_value)) + + +# Register after both classes are defined so the generic type can be resolved. +temporalio.converter.transfer_type_convertible( + TemporalTransferTypeGenericValueConverter +)(TemporalTransferTypeGenericValue) + + +class CustomDefaultPayloadConverter(temporalio.converter.DefaultPayloadConverter): + pass + + +class ContextualTransferPayloadConverter(CustomDefaultPayloadConverter): + def __init__(self) -> None: + super().__init__() + self.context: temporalio.converter.SerializationContext | None = None + self.encoding_contexts: list[ + temporalio.converter.SerializationContext | None + ] = [] + + def with_context( + self, context: temporalio.converter.SerializationContext + ) -> ContextualTransferPayloadConverter: + converter = ContextualTransferPayloadConverter() + converter.context = context + return converter + + def to_payloads( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + self.encoding_contexts.append(self.context) + return super().to_payloads(values) + + +async def test_transfer_type_conversion_preserves_user_converter_context(): + original = temporalio.converter.DataConverter( + payload_converter_class=ContextualTransferPayloadConverter + ) + context = temporalio.converter.WorkflowSerializationContext( + namespace="test", workflow_id="workflow-id" + ) + converter = original.with_context(context) + user_converter = converter.payload_converter + assert isinstance(user_converter, ContextualTransferPayloadConverter) + assert user_converter.context is context + assert user_converter is not original.payload_converter + value = TemporalTransferTypeValue("workflow-id") + + payloads = await converter.encode([value]) + + assert user_converter.encoding_contexts == [context] + assert await converter.decode(payloads, [TemporalTransferTypeValue]) == [value] + assert converter.payload_converter is user_converter + + +async def test_transfer_type_conversion_in_failure_details(): + converter = temporalio.converter.DataConverter( + payload_converter_class=CustomDefaultPayloadConverter + ) + failure = temporalio.api.failure.v1.Failure() + + await converter.encode_failure( + temporalio.exceptions.ApplicationError( + "test failure", TemporalTransferTypeValue("workflow-id") + ), + failure, + ) + converted = await converter.decode_failure(failure) + + assert isinstance(converted, temporalio.exceptions.ApplicationError) + assert converted.details == ( + temporalio.api.common.v1.WorkflowExecution( + workflow_id="workflow-id", run_id="run-id" + ), + ) + assert isinstance(converter.payload_converter, CustomDefaultPayloadConverter) + + +async def test_temporal_transfer_type_payload_converter_wraps_user_converter(): + data_converter = temporalio.converter.DataConverter( + payload_converter_class=CustomDefaultPayloadConverter + ) + assert isinstance(data_converter.payload_converter, CustomDefaultPayloadConverter) + assert isinstance( + temporalio.converter.PayloadConverter.default, + temporalio.converter.DefaultPayloadConverter, + ) + value = TemporalTransferTypeValue("workflow-id") + + [payload] = await data_converter.encode([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 await data_converter.decode([payload], [TemporalTransferTypeValue]) == [ + value + ] + + [plain_proto_payload] = await data_converter.encode( + [temporalio.api.common.v1.WorkflowExecution(workflow_id="id1", run_id="id2")] + ) + assert plain_proto_payload.metadata["encoding"] == b"json/protobuf" + + +async def test_temporal_transfer_type_payload_converter_without_transfer_type_hint(): + converter = temporalio.converter.DataConverter.default + value = TemporalTransferTypeValueWithoutHint("workflow-id") + + [payload] = await converter.encode([value]) + + assert payload.metadata["encoding"] == b"json/protobuf" + assert ( + payload.metadata["messageType"] == b"temporal.api.common.v1.WorkflowExecution" + ) + assert await converter.decode( + [payload], [TemporalTransferTypeValueWithoutHint] + ) == [value] + + +@pytest.mark.parametrize( + ("value", "type_hint"), + [ + ( + TemporalTransferTypeGenericValue("workflow-id"), + TemporalTransferTypeGenericValue[str], + ), + ( + TemporalTransferTypeGenericValue(123), + TemporalTransferTypeGenericValue[int], + ), + ], +) +async def test_temporal_transfer_type_payload_converter_with_generic_value( + value: TemporalTransferTypeGenericValue[T], + type_hint: type[TemporalTransferTypeGenericValue[T]], +): + converter = temporalio.converter.DataConverter.default + + [payload] = await converter.encode([value]) + + assert await converter.decode([payload], [type_hint]) == [value] + + +def test_transfer_type_convertible_rejects_existing_converter(): + with pytest.raises(TypeError, match="already has a transfer type converter"): + temporalio.converter.transfer_type_convertible( + TemporalTransferTypeValueConverter + )(TemporalTransferTypeValue) + + class TransferValueConverter( temporalio.converter.TransferTypeConverter["TransferValue", str] ): @@ -258,8 +510,10 @@ async def run(self, value: TransferValue, external: bool) -> TransferValue: id=f"{workflow.info().workflow_id}-child", ) if external: - handle = workflow.get_external_workflow_handle_for( - TransferMessagesWorkflow.run, child.id + handle: workflow.ExternalWorkflowHandle[TransferMessagesWorkflow] = ( + workflow.get_external_workflow_handle_for( + TransferMessagesWorkflow.run, child.id + ) ) await handle.signal( TransferMessagesWorkflow.signal, value.append("external") From 71755bf91a6d41c92ead984af58dec0c1cf3d7c2 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 17 Sep 2026 10:36:57 -0700 Subject: [PATCH 7/8] Explain internal payload converter instance lifetimes --- temporalio/converter/_data_converter.py | 4 ++++ temporalio/worker/_workflow.py | 2 ++ 2 files changed, 6 insertions(+) diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 3c2514ca4a..540c9bcd63 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -94,10 +94,14 @@ def __post_init__(self) -> None: # noqa: D105 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() ) diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index c9bc7ab34c..2cc7a7ab84 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -727,6 +727,8 @@ 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, failure_converter_class=self._data_converter.failure_converter_class, interceptor_classes=self._interceptor_classes, From 4cff05485c4625514f98dd8e29ad8659577c0248 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 17 Sep 2026 12:34:40 -0700 Subject: [PATCH 8/8] Clarify System Nexus payload converter naming --- temporalio/nexus/system/__init__.py | 12 +++++---- temporalio/worker/_nexus.py | 8 +++--- temporalio/worker/_workflow_instance.py | 20 ++++++++------ tests/nexus/test_temporal_system_nexus.py | 32 +++++++++++++---------- tests/worker/test_visitor.py | 8 +++--- 5 files changed, 46 insertions(+), 34 deletions(-) diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 31dadac8f1..52b36e9153 100644 --- a/temporalio/nexus/system/__init__.py +++ b/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 system envelope converter given an internal, wrapped payload converter.""" - return _SystemNexusPayloadConverter(user_payload_converter, user_failure_converter) + return _SystemNexusPayloadConverter( + internal_payload_converter, user_failure_converter + ) def _get_serialization_context( # pyright: ignore[reportUnusedFunction] diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index f68a658d56..38244e99fc 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -603,9 +603,11 @@ async def deserialize( try: payload_converter = dc._get_internal_payload_converter() if temporalio.nexus.system._is_system_payload(payload): - payload_converter = temporalio.nexus.system._get_payload_converter( - 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], diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 1084310218..430fe59e21 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/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: diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index 13ed52b1ed..9745b0328e 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -434,7 +434,7 @@ 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( + payload = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).to_payload(request) @@ -460,7 +460,7 @@ async def test_nexus_payload_serializer_decodes_system_input() -> None: namespace="target-namespace", headers={"test-header": "header-value"}, ) - payload = nexus_system._get_payload_converter( + payload = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).to_payload(request) @@ -494,7 +494,7 @@ async def test_nexus_payload_serializer_codec_skips_outer_envelope() -> None: signal="test-signal", namespace="target-namespace", ) - payload = nexus_system._get_payload_converter( + payload = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).to_payload(request) @@ -542,7 +542,7 @@ async def test_schedule_marked_system_nexus_payload_ignores_endpoint() -> None: schedule = completion.successful.commands[0].schedule_nexus_operation data_converter = temporalio.converter.default() - decoded = nexus_system._get_payload_converter( + decoded = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).from_payload(schedule.input) @@ -569,7 +569,7 @@ async def test_schedule_unmarked_system_nexus_payload_visits_input_as_regular_pa schedule = completion.successful.commands[0].schedule_nexus_operation assert schedule.input.metadata["visited"] == b"true" data_converter = temporalio.converter.default() - decoded = nexus_system._get_payload_converter( + decoded = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ).from_payload(schedule.input) @@ -692,7 +692,7 @@ def _field_is_repeated(field: FieldDescriptor) -> bool: ) def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None: data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_payload_converter( + payload_converter = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ) @@ -710,7 +710,7 @@ def test_system_nexus_proto_roundtrip(message_type: type[Message]) -> None: def test_system_nexus_uses_user_failure_converter() -> None: payload_converter = temporalio.converter.default()._get_internal_payload_converter() failure_converter = _TrackingFailureConverter(DefaultPayloadConverter) - system_converter = nexus_system._get_payload_converter( + system_converter = nexus_system._get_system_nexus_payload_converter( payload_converter, failure_converter ) @@ -749,7 +749,7 @@ def to_failure( assert isinstance(payload_converter, DefaultPayloadConverter) raise ValueError("conversion failed") - inner_system_converter = nexus_system._get_payload_converter( + inner_system_converter = nexus_system._get_system_nexus_payload_converter( inner_data_converter._get_internal_payload_converter(), RaisingFailureConverter(), ) @@ -833,20 +833,24 @@ async def test_signal_with_start_uses_target_workflow_serialization_context( captured_contexts: list[SerializationContext | None] = [] system_payload_converter_wrap_count = 0 - original_get_payload_converter = nexus_system._get_payload_converter + original_get_system_nexus_payload_converter = ( + nexus_system._get_system_nexus_payload_converter + ) - def capture_get_payload_converter( - user_payload_converter: temporalio.converter.PayloadConverter, + def capture_get_system_nexus_payload_converter( + internal_payload_converter: temporalio.converter.PayloadConverter, user_failure_converter: temporalio.converter.FailureConverter, ) -> temporalio.converter.PayloadConverter: nonlocal system_payload_converter_wrap_count system_payload_converter_wrap_count += 1 - return original_get_payload_converter( - user_payload_converter, user_failure_converter + return original_get_system_nexus_payload_converter( + internal_payload_converter, user_failure_converter ) monkeypatch.setattr( - nexus_system, "_get_payload_converter", capture_get_payload_converter + nexus_system, + "_get_system_nexus_payload_converter", + capture_get_system_nexus_payload_converter, ) caller_config = env.client.config() caller_config["data_converter"] = dataclasses.replace( diff --git a/tests/worker/test_visitor.py b/tests/worker/test_visitor.py index 0c6789f78b..320f537aa9 100644 --- a/tests/worker/test_visitor.py +++ b/tests/worker/test_visitor.py @@ -235,7 +235,7 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: input=Payloads(payloads=[Payload(data=b"workflow-input")]), ) data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_payload_converter( + payload_converter = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ) @@ -280,7 +280,7 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: run_id="test-run-id" ) data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_payload_converter( + payload_converter = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ) @@ -307,7 +307,7 @@ async def visit_system_nexus_envelope(self, payload: Payload) -> None: async def test_unknown_system_nexus_payload_raises_application_error(): data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_payload_converter( + payload_converter = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, ) @@ -482,7 +482,7 @@ async def _visit(self) -> None: active_visits -= 1 data_converter = temporalio.converter.default() - payload_converter = nexus_system._get_payload_converter( + payload_converter = nexus_system._get_system_nexus_payload_converter( data_converter._get_internal_payload_converter(), data_converter.failure_converter, )