From 32717a375053b40e50b8fae720f304f0d9da72ed Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 12:31:15 -0700 Subject: [PATCH 1/4] Use declared types for transfer serialization --- CHANGELOG.md | 4 + temporalio/client/_client.py | 17 + temporalio/client/_impl.py | 20 +- temporalio/client/_interceptor.py | 8 + temporalio/client/_nexus.py | 13 +- temporalio/client/_schedule.py | 14 +- temporalio/client/_workflow.py | 19 + temporalio/converter/_data_converter.py | 18 + temporalio/converter/_payload_converter.py | 64 +++- temporalio/worker/_activity.py | 7 +- temporalio/worker/_interceptor.py | 5 + temporalio/worker/_nexus.py | 9 +- temporalio/worker/_workflow_instance.py | 56 ++- tests/test_serialization_type_hints.py | 389 +++++++++++++++++++++ 14 files changed, 610 insertions(+), 33 deletions(-) create mode 100644 tests/test_serialization_type_hints.py diff --git a/CHANGELOG.md b/CHANGELOG.md index e8597d2c2..cfa2cf757 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,6 +29,10 @@ to include examples, links to docs, or any other relevant information. ### Fixed +- Use declared argument and result types, when available, to select transfer type + converters during serialization, while preserving existing data and payload + converter method signatures. + ### Security ## [1.33.0] - 2026-09-14 diff --git a/temporalio/client/_client.py b/temporalio/client/_client.py index b5db0c4b7..120b9a093 100644 --- a/temporalio/client/_client.py +++ b/temporalio/client/_client.py @@ -625,6 +625,11 @@ async def start_workflow( return await self._impl.start_workflow( StartWorkflowInput( workflow=name, + arg_types=( + temporalio.workflow._Definition.must_from_run_fn(workflow).arg_types + if callable(workflow) + else None + ), args=temporalio.common._arg_or_args(arg, args), id=id, task_queue=task_queue, @@ -1187,6 +1192,11 @@ async def _start_update_with_start( update_input = UpdateWithStartUpdateWorkflowInput( update_id=id, update=update_name, + arg_types=( + update._defn.arg_types + if isinstance(update, temporalio.workflow.UpdateMethodMultiParam) + else None + ), args=temporalio.common._arg_or_args(arg, args), headers={}, ret_type=result_type or result_type_from_type_hint, @@ -1521,6 +1531,13 @@ async def start_activity( return await self._impl.start_activity( StartActivityInput( activity_type=name, + arg_types=( + temporalio.activity._Definition.must_from_callable( + activity + ).arg_types + if callable(activity) + else None + ), args=temporalio.common._arg_or_args(arg, args), id=id, task_queue=task_queue, diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index 848975dc9..6e411321e 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -284,7 +284,9 @@ async def _populate_start_workflow_execution_request( req.workflow_type.name = input.workflow req.task_queue.name = input.task_queue if input.args: - req.input.payloads.extend(await data_converter.encode(input.args)) + req.input.payloads.extend( + await data_converter.encode_with_type_hints(input.args, input.arg_types) + ) if input.execution_timeout is not None: req.workflow_execution_timeout.FromTimedelta(input.execution_timeout) if input.run_timeout is not None: @@ -422,7 +424,7 @@ async def query_workflow(self, input: QueryWorkflowInput) -> Any: req.query.query_type = input.query if input.args: req.query.query_args.payloads.extend( - await data_converter.encode(input.args) + await data_converter.encode_with_type_hints(input.args, input.arg_types) ) if input.headers is not None: # type:ignore[reportUnnecessaryComparison] await self._apply_headers(input.headers, req.query.header.fields) @@ -484,7 +486,9 @@ async def signal_workflow(self, input: SignalWorkflowInput) -> None: request_id=str(uuid.uuid4()), ) if input.args: - req.input.payloads.extend(await data_converter.encode(input.args)) + req.input.payloads.extend( + await data_converter.encode_with_type_hints(input.args, input.arg_types) + ) if input.headers is not None: # type:ignore[reportUnnecessaryComparison] await self._apply_headers(input.headers, req.header.fields) temporalio.nexus._operation_context._apply_nexus_context_to_signal_workflow_request( @@ -626,7 +630,9 @@ async def _build_start_activity_execution_request( # Set input payloads if input.args: - req.input.payloads.extend(await data_converter.encode(input.args)) + req.input.payloads.extend( + await data_converter.encode_with_type_hints(input.args, input.arg_types) + ) # Set search attributes if input.search_attributes is not None: @@ -939,7 +945,7 @@ async def _build_update_workflow_execution_request( ) if input.args: req.request.input.args.payloads.extend( - await data_converter.encode(input.args) + await data_converter.encode_with_type_hints(input.args, input.arg_types) ) if input.headers is not None: # type:ignore[reportUnnecessaryComparison] await self._apply_headers(input.headers, req.request.input.header.fields) @@ -1581,7 +1587,9 @@ async def start_nexus_operation( req.start_to_close_timeout.FromTimedelta(input.start_to_close_timeout) # Set input payload - encoded = await data_converter.encode([input.arg]) + encoded = await data_converter.encode_with_type_hints( + [input.arg], [input.input_type] + ) if encoded: req.input.CopyFrom(encoded[0]) diff --git a/temporalio/client/_interceptor.py b/temporalio/client/_interceptor.py index 68077ebc9..d79d6d766 100644 --- a/temporalio/client/_interceptor.py +++ b/temporalio/client/_interceptor.py @@ -96,6 +96,7 @@ class StartWorkflowInput: request_eager_start: bool priority: temporalio.common.Priority versioning_override: temporalio.common.VersioningOverride | None = None + arg_types: list[type] | None = None @dataclass @@ -170,6 +171,7 @@ class QueryWorkflowInput: ret_type: type | None rpc_metadata: Mapping[str, str | bytes] rpc_timeout: timedelta | None + arg_types: list[type] | None = None @dataclass @@ -183,6 +185,7 @@ class SignalWorkflowInput: headers: Mapping[str, temporalio.api.common.v1.Payload] rpc_metadata: Mapping[str, str | bytes] rpc_timeout: timedelta | None + arg_types: list[type] | None = None @dataclass @@ -221,6 +224,7 @@ class StartActivityInput: headers: Mapping[str, temporalio.api.common.v1.Payload] rpc_metadata: Mapping[str, str | bytes] rpc_timeout: timedelta | None + arg_types: list[type] | None = None @dataclass @@ -342,6 +346,7 @@ class StartWorkflowUpdateInput: ret_type: type | None rpc_metadata: Mapping[str, str | bytes] rpc_timeout: timedelta | None + arg_types: list[type] | None = None @dataclass @@ -354,6 +359,7 @@ class UpdateWithStartUpdateWorkflowInput: wait_for_stage: WorkflowUpdateStage headers: Mapping[str, temporalio.api.common.v1.Payload] ret_type: type | None + arg_types: list[type] | None = None @dataclass @@ -386,6 +392,7 @@ class UpdateWithStartStartWorkflowInput: ret_type: type | None priority: temporalio.common.Priority versioning_override: temporalio.common.VersioningOverride | None = None + arg_types: list[type] | None = None @dataclass @@ -605,6 +612,7 @@ class StartNexusOperationInput: headers: Mapping[str, str] rpc_metadata: Mapping[str, str | bytes] rpc_timeout: timedelta | None + input_type: type | None = None @dataclass diff --git a/temporalio/client/_nexus.py b/temporalio/client/_nexus.py index 1f0a7338e..172ede348 100644 --- a/temporalio/client/_nexus.py +++ b/temporalio/client/_nexus.py @@ -940,16 +940,16 @@ def __init__( def _resolve_operation( self, operation: nexusrpc.Operation[Any, Any] | str | Callable[..., Any], - ) -> tuple[str, type | None]: - """Resolve an operation to its name and output type.""" + ) -> tuple[str, type | None, type | None]: + """Resolve an operation to its name, input type, and output type.""" if isinstance(operation, str): - return operation, None + return operation, None, None elif isinstance(operation, nexusrpc.Operation): - return operation.name, operation.output_type + return operation.name, operation.input_type, operation.output_type elif callable(operation): _, op = temporalio.nexus._util.get_operation_factory(operation) if isinstance(op, nexusrpc.Operation): - return op.name, op.output_type + return op.name, op.input_type, op.output_type else: raise ValueError( f"Operation callable is not a Nexus operation: {operation}" @@ -982,7 +982,7 @@ async def start_operation( .. warning:: This API is experimental and unstable. """ - op_name, output_type = self._resolve_operation(operation) + op_name, input_type, output_type = self._resolve_operation(operation) final_result_type: type | None = ( result_type if isinstance(operation, str) else output_type ) @@ -990,6 +990,7 @@ async def start_operation( return await self._client._impl.start_nexus_operation( StartNexusOperationInput( operation=op_name, + input_type=input_type, arg=arg, id=id, endpoint=self._endpoint, diff --git a/temporalio/client/_schedule.py b/temporalio/client/_schedule.py index 15946cfe8..bc3f28ae4 100644 --- a/temporalio/client/_schedule.py +++ b/temporalio/client/_schedule.py @@ -557,6 +557,7 @@ class ScheduleActionStartWorkflow(ScheduleAction): Headers may still be encoded by the payload codec if present. """ _from_raw: bool = dataclasses.field(compare=False, init=False) + _arg_types: list[type] | None = dataclasses.field(compare=False, init=False) @staticmethod def _from_proto( # pyright: ignore @@ -682,6 +683,7 @@ def __init__( values. """ super().__init__() + self._arg_types = None if raw_info: self._from_raw = True # Ignore other fields @@ -753,6 +755,7 @@ def __init__( defn = temporalio.workflow._Definition.must_from_run_fn(workflow) if not defn.name: raise ValueError("Cannot schedule dynamic workflow explicitly") + self._arg_types = defn.arg_types workflow = defn.name elif not isinstance(workflow, str): raise TypeError("Workflow must be a string or callable") # type:ignore[reportUnreachable] @@ -815,8 +818,15 @@ async def _to_proto( payloads=[ a if isinstance(a, temporalio.api.common.v1.Payload) - else (await data_converter.encode([a]))[0] - for a in self.args + else ( + await data_converter.encode_with_type_hints( + [a], + [self._arg_types[index]] + if self._arg_types and index < len(self._arg_types) + else None, + ) + )[0] + for index, a in enumerate(self.args) ] ) if self.args diff --git a/temporalio/client/_workflow.py b/temporalio/client/_workflow.py index 0607ade0a..0aa3e6f5c 100644 --- a/temporalio/client/_workflow.py +++ b/temporalio/client/_workflow.py @@ -580,6 +580,7 @@ async def query( """ query_name: str ret_type = result_type + arg_types: list[type] | None = None if callable(query): defn = temporalio.workflow._QueryDefinition.from_fn(query) if not defn: @@ -592,6 +593,7 @@ async def query( # TODO(cretz): Check count/type of args at runtime? query_name = defn.name ret_type = defn.ret_type + arg_types = defn.arg_types else: query_name = str(query) @@ -600,6 +602,7 @@ async def query( id=self._id, run_id=self._run_id, query=query_name, + arg_types=arg_types, args=temporalio.common._arg_or_args(arg, args), reject_condition=reject_condition or self._client._config["default_workflow_query_reject_condition"], @@ -692,6 +695,12 @@ async def signal( signal=temporalio.workflow._SignalDefinition.must_name_from_fn_or_str( signal ), + arg_types=( + defn.arg_types + if callable(signal) + and (defn := temporalio.workflow._SignalDefinition.from_fn(signal)) + else None + ), args=temporalio.common._arg_or_args(arg, args), headers={}, rpc_metadata=rpc_metadata, @@ -970,6 +979,11 @@ async def _start_update( first_execution_run_id=self._first_execution_run_id, update_id=id, update=update_name, + arg_types=( + update._defn.arg_types + if isinstance(update, temporalio.workflow.UpdateMethodMultiParam) + else None + ), args=temporalio.common._arg_or_args(arg, args), headers={}, ret_type=result_type or result_type_from_type_hint, @@ -1201,6 +1215,11 @@ def __init__( self._start_workflow_input = UpdateWithStartStartWorkflowInput( workflow=name, + arg_types=( + temporalio.workflow._Definition.must_from_run_fn(workflow).arg_types + if callable(workflow) + else None + ), args=temporalio.common._arg_or_args(arg, args), id=id, task_queue=task_queue, diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 8604ea196..4c02fd2b4 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -29,6 +29,7 @@ from temporalio.converter._payload_converter import ( PayloadConverter, _TemporalTransferTypePayloadConverter, + _TypeHintedValues, ) from temporalio.converter._serialization_context import ( SerializationContext, @@ -98,6 +99,23 @@ def _new_payload_converter(self) -> PayloadConverter: self.payload_converter_class() ) + async def encode_with_type_hints( + self, + values: Sequence[Any], + type_hints: Sequence[type | None] | None = None, + ) -> list[temporalio.api.common.v1.Payload]: + """Encode values using declared types for transfer converter selection. + + Hints correspond to values by position. A missing or None hint uses the + value's runtime type; hints for omitted arguments are ignored. + Existing :py:meth:`encode` overrides are invoked unchanged; overrides + should forward the original sequence to preserve hints when delegating + to payload conversion. + """ + return await self.encode( + _TypeHintedValues(values, type_hints) if type_hints is not None else values + ) + async def encode( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index a8bc35e28..a15cf4532 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -57,9 +57,22 @@ _TRANSFER_TYPE_CONVERTER_ATTR = "__temporal_transfer_type_converter" +class _TypeHintedValues(list[Any]): + """Carry hints through legacy overrides without changing their signatures.""" + + def __init__( + self, values: Sequence[Any], type_hints: Sequence[type | None] + ) -> None: + super().__init__(values) + self.type_hints = tuple(type_hints) + + class TransferTypeConverter(Generic[ValueT, TransferTypeT], ABC): """Converter between a user-facing value and a transfer type value. + When available, the declared type determines which converter is used for + serialization. Otherwise, the value's runtime type is used. + .. warning:: This API is experimental and subject to change. """ @@ -80,6 +93,19 @@ def to_transfer_type(self, value: ValueT) -> TransferTypeT: """ raise NotImplementedError + def to_transfer_type_with_type_hint( + self, + value: ValueT, + type_hint: type[ValueT] | None, # type: ignore[reportUnusedParameter] + ) -> TransferTypeT: + """Convert a value with its declared type, including generic arguments. + + The default implementation delegates to :py:meth:`to_transfer_type`. + Override this method when conversion needs the declared type. The hint + is None when the caller has no declared type available. + """ + return self.to_transfer_type(value) + @abstractmethod def from_transfer_type( self, value: TransferTypeT, type_hint: type[ValueT] @@ -122,6 +148,9 @@ def transfer_type_convertible( def _get_transfer_type_converter( value_type: object, ) -> TransferTypeConverter[Any, Any] | None: + while typing.get_origin(value_type) is typing.Annotated: + value_type = typing.get_args(value_type)[0] + value_type = typing.get_origin(value_type) or value_type converter = getattr(value_type, _TRANSFER_TYPE_CONVERTER_ATTR, None) if isinstance(converter, TransferTypeConverter): return converter @@ -134,6 +163,23 @@ class PayloadConverter(ABC): default: ClassVar[PayloadConverter] """Default payload converter.""" + def to_payloads_with_type_hints( + self, + values: Sequence[Any], + type_hints: Sequence[type | None] | None = None, + ) -> list[temporalio.api.common.v1.Payload]: + """Convert values using declared types for transfer converter selection. + + Hints correspond to values by position. A missing or None hint uses the + value's runtime type; hints for omitted arguments are ignored. + Existing :py:meth:`to_payloads` overrides are invoked unchanged; + overrides should forward the original sequence to preserve hints when + delegating to another payload converter. + """ + return self.to_payloads( + _TypeHintedValues(values, type_hints) if type_hints is not None else values + ) + @abstractmethod def to_payloads( self, values: Sequence[Any] @@ -617,10 +663,22 @@ def to_payloads( ) -> list[temporalio.api.common.v1.Payload]: """See base class.""" transfer_type_values: list[Any] = [] - for value in values: - converter = _get_transfer_type_converter(type(value)) + type_hints = ( + values.type_hints + if isinstance(values, _TypeHintedValues) + else (None,) * len(values) + ) + for index, value in enumerate(values): + type_hint = type_hints[index] if index < len(type_hints) else None + converter = ( + None + if isinstance(value, temporalio.common.RawValue) + else _get_transfer_type_converter( + type_hint if type_hint is not None else type(value) + ) + ) if converter is not None: - value = converter.to_transfer_type(value) + value = converter.to_transfer_type_with_type_hint(value, type_hint) transfer_type_values.append(value) return self._inner_payload_converter.to_payloads(transfer_type_values) diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index ded3047fc..07ee2f4c4 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -351,7 +351,12 @@ async def _handle_start_activity_task( result = await self._execute_activity( start, running_activity, task_token, data_converter ) - [payload] = await data_converter.encode([result]) + activity_def = self._activities.get( + start.activity_type, self._dynamic_activity + ) + [payload] = await data_converter.encode_with_type_hints( + [result], [activity_def.ret_type if activity_def else None] + ) completion.result.completed.result.CopyFrom(payload) except BaseException as err: try: diff --git a/temporalio/worker/_interceptor.py b/temporalio/worker/_interceptor.py index f59b534c2..1a4e125c0 100644 --- a/temporalio/worker/_interceptor.py +++ b/temporalio/worker/_interceptor.py @@ -231,6 +231,7 @@ class SignalChildWorkflowInput: args: Sequence[Any] child_workflow_id: str headers: Mapping[str, temporalio.api.common.v1.Payload] + arg_types: list[type] | None = None @dataclass @@ -243,6 +244,7 @@ class SignalExternalWorkflowInput: workflow_id: str workflow_run_id: str | None headers: Mapping[str, temporalio.api.common.v1.Payload] + arg_types: list[type] | None = None @dataclass @@ -314,15 +316,18 @@ class StartNexusOperationInput(Generic[InputT, OutputT]): headers: Mapping[str, str] | None summary: str | None output_type: type[OutputT] | None = None + input_type: type[InputT] | None = None def __post_init__(self) -> None: """Initialize operation-specific attributes after dataclass creation.""" if isinstance(self.operation, nexusrpc.Operation): self.output_type = self.operation.output_type + self.input_type = self.operation.input_type elif callable(self.operation): _, op = temporalio.nexus._util.get_operation_factory(self.operation) if isinstance(op, nexusrpc.Operation): self.output_type = op.output_type + self.input_type = op.input_type else: raise ValueError( f"Operation callable is not a Nexus operation: {self.operation}" diff --git a/temporalio/worker/_nexus.py b/temporalio/worker/_nexus.py index a2f4b8ca7..1455ef051 100644 --- a/temporalio/worker/_nexus.py +++ b/temporalio/worker/_nexus.py @@ -471,7 +471,14 @@ async def _start_operation( ) ) elif isinstance(result, nexusrpc.handler.StartOperationResultSync): - [payload] = data_converter.payload_converter.to_payloads([result.value]) + operation = self._handler.service_handlers[ + start_request.service + ].service.operation_definitions[start_request.operation] + [payload] = ( + data_converter.payload_converter.to_payloads_with_type_hints( + [result.value], [operation.output_type] + ) + ) return temporalio.api.nexus.v1.StartOperationResponse( sync_success=temporalio.api.nexus.v1.StartOperationResponse.Sync( payload=payload, diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 89625d64c..3c3e38b3c 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -723,8 +723,8 @@ async def run_update() -> None: # Run the handler success = await self._inbound.handle_update_handler(handler_input) - result_payloads = self._workflow_context_payload_converter.to_payloads( - [success] + result_payloads = self._workflow_context_payload_converter.to_payloads_with_type_hints( + [success], [defn.ret_type] ) if len(result_payloads) != 1: raise ValueError( @@ -829,8 +829,8 @@ async def run_query() -> None: headers=job.headers, ) success = await self._inbound.handle_query(input) - result_payloads = ( - self._workflow_context_payload_converter.to_payloads([success]) + result_payloads = self._workflow_context_payload_converter.to_payloads_with_type_hints( + [success], [defn.ret_type] ) if len(result_payloads) != 1: raise ValueError( @@ -1151,8 +1151,8 @@ def _apply_initialize_workflow( async def run_workflow(input: ExecuteWorkflowInput) -> None: try: result = await self._inbound.execute_workflow(input) - result_payloads = self._workflow_context_payload_converter.to_payloads( - [result] + result_payloads = self._workflow_context_payload_converter.to_payloads_with_type_hints( + [result], [self._defn.ret_type] ) if len(result_payloads) != 1: raise ValueError( @@ -1252,7 +1252,9 @@ def workflow_continue_as_new( defn = temporalio.workflow._Definition.must_from_run_fn(workflow) name = defn.name arg_types = defn.arg_types - elif workflow is not None: + elif workflow is None: + arg_types = self._defn.arg_types + else: raise TypeError("Workflow must be None, a string, or callable") # type:ignore[reportUnreachable] self._outbound.continue_as_new( @@ -2064,7 +2066,11 @@ async def _outbound_signal_child_workflow( workflow_id=input.child_workflow_id, ) ) - payloads = payload_converter.to_payloads(input.args) if input.args else None + payloads = ( + payload_converter.to_payloads_with_type_hints(input.args, input.arg_types) + if input.args + else None + ) command = self._add_command() v = command.signal_external_workflow_execution v.child_workflow_id = input.child_workflow_id @@ -2084,7 +2090,11 @@ async def _outbound_signal_external_workflow( workflow_id=input.workflow_id, ) ) - payloads = payload_converter.to_payloads(input.args) if input.args else None + payloads = ( + payload_converter.to_payloads_with_type_hints(input.args, input.arg_types) + if input.args + else None + ) command = self._add_command() v = command.signal_external_workflow_execution v.workflow_execution.namespace = input.namespace @@ -3372,7 +3382,9 @@ def _apply_schedule_command( ) -> None: # Convert arguments before creating command in case it raises error payloads = ( - self._payload_converter.to_payloads(self._input.args) + self._payload_converter.to_payloads_with_type_hints( + self._input.args, self._input.arg_types + ) if self._input.args else None ) @@ -3504,6 +3516,12 @@ async def signal( signal=temporalio.workflow._SignalDefinition.must_name_from_fn_or_str( signal ), + arg_types=( + defn.arg_types + if callable(signal) + and (defn := temporalio.workflow._SignalDefinition.from_fn(signal)) + else None + ), args=temporalio.common._arg_or_args(arg, args), child_workflow_id=self._input.id, headers={}, @@ -3532,7 +3550,9 @@ def _resolve_failure(self, err: BaseException) -> None: def _apply_start_command(self) -> None: # Convert arguments before creating command in case it raises error payloads = ( - self._payload_converter.to_payloads(self._input.args) + self._payload_converter.to_payloads_with_type_hints( + self._input.args, self._input.arg_types + ) if self._input.args else None ) @@ -3634,6 +3654,12 @@ async def signal( signal=temporalio.workflow._SignalDefinition.must_name_from_fn_or_str( signal ), + arg_types=( + defn.arg_types + if callable(signal) + and (defn := temporalio.workflow._SignalDefinition.from_fn(signal)) + else None + ), args=temporalio.common._arg_or_args(arg, args), namespace=self._instance._info.namespace, workflow_id=self._id, @@ -3704,7 +3730,9 @@ def _resolve_failure(self, err: BaseException) -> None: self._result_fut.set_result(None) def _apply_schedule_command(self) -> None: - payload = self._payload_converter.to_payload(self._input.input) + payload = self._payload_converter.to_payloads_with_type_hints( + [self._input.input], [self._input.input_type] + )[0] command = self._instance._add_command() v = command.schedule_nexus_operation v.seq = self._seq @@ -3754,8 +3782,8 @@ def __init__( def _apply_command(self) -> None: # Convert arguments before creating command in case it raises error payloads = ( - self._instance._workflow_context_payload_converter.to_payloads( - self._input.args + self._instance._workflow_context_payload_converter.to_payloads_with_type_hints( + self._input.args, self._input.arg_types ) if self._input.args else None diff --git a/tests/test_serialization_type_hints.py b/tests/test_serialization_type_hints.py new file mode 100644 index 000000000..807e077f2 --- /dev/null +++ b/tests/test_serialization_type_hints.py @@ -0,0 +1,389 @@ +from __future__ import annotations + +import asyncio +from collections.abc import Sequence +from datetime import timedelta +from typing import Any, Generic, TypeVar, cast +from uuid import uuid4 + +import nexusrpc +import nexusrpc.handler +import pytest + +import temporalio.activity +import temporalio.api.common.v1 +import temporalio.client +import temporalio.common +import temporalio.converter +import temporalio.testing +import temporalio.worker +import temporalio.workflow + + +class DeclaredValue(str): + pass + + +class RuntimeValue(str): + pass + + +class DeclaredConverter(temporalio.converter.TransferTypeConverter[DeclaredValue, str]): + transfer_type = str + + def to_transfer_type(self, value: DeclaredValue) -> str: + return f"declared:{value}" + + def from_transfer_type( + self, value: str, type_hint: type[DeclaredValue] + ) -> DeclaredValue: + assert value.startswith("declared:") + return DeclaredValue(value.removeprefix("declared:")) + + +class RuntimeConverter(temporalio.converter.TransferTypeConverter[RuntimeValue, str]): + transfer_type = str + + def to_transfer_type(self, value: RuntimeValue) -> str: + return f"runtime:{value}" + + def from_transfer_type( + self, value: str, type_hint: type[RuntimeValue] + ) -> RuntimeValue: + return RuntimeValue(value.removeprefix("runtime:")) + + +temporalio.converter.transfer_type_convertible(DeclaredConverter)(DeclaredValue) +temporalio.converter.transfer_type_convertible(RuntimeConverter)(RuntimeValue) + + +@pytest.mark.parametrize( + ("hints", "expected"), + [ + ([DeclaredValue], "declared:value"), + ([str], "value"), + ([None], "runtime:value"), + (None, "runtime:value"), + ([], "runtime:value"), + ([DeclaredValue, int], "declared:value"), + ], +) +async def test_transfer_serialization_type_selection( + hints: list[type | None] | None, expected: str +): + converter = temporalio.converter.DataConverter.default + payloads = await converter.encode_with_type_hints([RuntimeValue("value")], hints) + assert await converter.decode(payloads) == [expected] + + +T = TypeVar("T") + + +class GenericValue(Generic[T]): + pass + + +class GenericConverter( + temporalio.converter.TransferTypeConverter[GenericValue[Any], str] +): + transfer_type = str + + def to_transfer_type(self, value: GenericValue[Any]) -> str: + return "no hint" + + def to_transfer_type_with_type_hint( + self, value: GenericValue[Any], type_hint: type[GenericValue[Any]] | None + ) -> str: + assert type_hint == GenericValue[int] + return "generic hint" + + def from_transfer_type( + self, value: str, type_hint: type[GenericValue[Any]] + ) -> GenericValue[Any]: + return GenericValue() + + +temporalio.converter.transfer_type_convertible(GenericConverter)(GenericValue) + + +async def test_transfer_serialization_generic_hint(): + converter = temporalio.converter.DataConverter.default + payloads = await converter.encode_with_type_hints( + [GenericValue()], [GenericValue[int]] + ) + assert await converter.decode(payloads) == ["generic hint"] + + +async def test_transfer_serialization_legacy_overrides_and_nested_conversion(): + calls: list[str] = [] + + class LegacyPayloadConverter(temporalio.converter.DefaultPayloadConverter): + def to_payloads( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + calls.append("payload") + nested = ( + temporalio.converter.DataConverter.default.payload_converter.to_payload( + RuntimeValue("nested") + ) + ) + assert ( + temporalio.converter.DataConverter.default.payload_converter.from_payload( + nested + ) + == "runtime:nested" + ) + return super().to_payloads(values) + + class LegacyDataConverter(temporalio.converter.DataConverter): + async def encode( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + calls.append("data") + await asyncio.sleep(0) + return await super().encode(values) + + converter = LegacyDataConverter(payload_converter_class=LegacyPayloadConverter) + typed, untyped = await asyncio.gather( + converter.encode_with_type_hints([RuntimeValue("typed")], [DeclaredValue]), + converter.encode([RuntimeValue("untyped")]), + ) + assert await converter.decode(typed) == ["declared:typed"] + assert await converter.decode(untyped) == ["runtime:untyped"] + assert calls.count("data") == calls.count("payload") == 2 + + +async def test_transfer_serialization_raw_value(): + converter = temporalio.converter.DataConverter.default + [raw] = await converter.encode(["already encoded"]) + assert await converter.encode_with_type_hints( + [temporalio.common.RawValue(raw)], [DeclaredValue] + ) == [raw] + + +@temporalio.activity.defn +async def typed_activity(value: DeclaredValue) -> DeclaredValue: + assert type(value) is DeclaredValue + return cast(DeclaredValue, str(value)) + + +@temporalio.workflow.defn +class TypedChildWorkflow: + def __init__(self) -> None: + self.signal_count = 0 + + @temporalio.workflow.run + async def run(self, value: DeclaredValue) -> DeclaredValue: + assert type(value) is DeclaredValue + await temporalio.workflow.wait_condition(lambda: self.signal_count == 2) + return cast(DeclaredValue, str(value)) + + @temporalio.workflow.signal + def signal(self, value: DeclaredValue) -> None: + assert type(value) is DeclaredValue + self.signal_count += 1 + + +@temporalio.workflow.defn +class TypedWorkflow: + def __init__(self) -> None: + self.signalled = False + + @temporalio.workflow.run + async def run(self, value: DeclaredValue, continued: bool = False) -> DeclaredValue: + assert type(value) is DeclaredValue + if not continued: + temporalio.workflow.continue_as_new(args=[str(value), True]) + await temporalio.workflow.wait_condition(lambda: self.signalled) + value = await temporalio.workflow.execute_activity( + typed_activity, + cast(DeclaredValue, str(value)), + start_to_close_timeout=timedelta(seconds=10), + ) + value = await temporalio.workflow.execute_local_activity( + typed_activity, + cast(DeclaredValue, str(value)), + start_to_close_timeout=timedelta(seconds=10), + ) + child = await temporalio.workflow.start_child_workflow( + TypedChildWorkflow.run, cast(DeclaredValue, str(value)) + ) + await child.signal(TypedChildWorkflow.signal, cast(DeclaredValue, str(value))) + external: temporalio.workflow.ExternalWorkflowHandle[TypedChildWorkflow] = ( + temporalio.workflow.get_external_workflow_handle_for( + TypedChildWorkflow.run, child.id + ) + ) + await external.signal( + TypedChildWorkflow.signal, cast(DeclaredValue, str(value)) + ) + value = await child + return cast(DeclaredValue, str(value)) + + @temporalio.workflow.signal + def finish(self, value: DeclaredValue) -> None: + assert type(value) is DeclaredValue + self.signalled = True + + @temporalio.workflow.query + def echo_query(self, value: DeclaredValue) -> DeclaredValue: + assert type(value) is DeclaredValue + return cast(DeclaredValue, str(value)) + + @temporalio.workflow.update + async def echo_update(self, value: DeclaredValue) -> DeclaredValue: + assert type(value) is DeclaredValue + return cast(DeclaredValue, str(value)) + + @temporalio.workflow.query + def continued(self) -> bool: + return bool(temporalio.workflow.info().continued_run_id) + + +async def test_transfer_serialization_workflow_and_activity( + client: temporalio.client.Client, +): + task_queue = str(uuid4()) + value = cast(DeclaredValue, str("value")) + async with temporalio.worker.Worker( + client, + task_queue=task_queue, + workflows=[TypedWorkflow, TypedChildWorkflow], + activities=[typed_activity], + ): + handle = await client.start_workflow( + TypedWorkflow.run, + args=[value, False], + id=str(uuid4()), + task_queue=task_queue, + ) + while not await handle.query(TypedWorkflow.continued): + await asyncio.sleep(0.01) + assert await handle.query(TypedWorkflow.echo_query, value) == "value" + assert await handle.execute_update(TypedWorkflow.echo_update, value) == "value" + await handle.signal(TypedWorkflow.finish, value) + assert await handle.result() == "value" + activity = await client.start_activity( + typed_activity, + value, + id=str(uuid4()), + task_queue=task_queue, + start_to_close_timeout=timedelta(seconds=10), + ) + assert await activity.result() == "value" + + +async def test_transfer_serialization_schedule(client: temporalio.client.Client): + action = temporalio.client.ScheduleActionStartWorkflow( + TypedChildWorkflow.run, + cast(DeclaredValue, str("scheduled")), + id=str(uuid4()), + task_queue="unused", + ) + proto = await action._to_proto(client) + assert await client.data_converter.decode(proto.start_workflow.input.payloads) == [ + "declared:scheduled" + ] + + +@nexusrpc.service +class TypedService: + echo: nexusrpc.Operation[DeclaredValue, DeclaredValue] + + +@nexusrpc.handler.service_handler(service=TypedService) +class TypedServiceHandler: + @nexusrpc.handler.sync_operation + async def echo( + self, _ctx: nexusrpc.handler.StartOperationContext, value: DeclaredValue + ) -> DeclaredValue: + assert type(value) is DeclaredValue + return cast(DeclaredValue, str(value)) + + +@temporalio.workflow.defn +class TypedNexusWorkflow: + @temporalio.workflow.run + async def run(self, endpoint: str) -> DeclaredValue: + client = temporalio.workflow.create_nexus_client( + service=TypedService, endpoint=endpoint + ) + operations: Sequence[Any] = [TypedService.echo, TypedServiceHandler.echo] + for operation in operations: + result = await client.execute_operation( + operation, + cast(DeclaredValue, str("nexus")), + schedule_to_close_timeout=timedelta(seconds=10), + ) + assert type(result) is DeclaredValue + return cast(DeclaredValue, str("nexus")) + + +@pytest.mark.requires_local_server +async def test_transfer_serialization_nexus( + env: temporalio.testing.WorkflowEnvironment, +): + task_queue = str(uuid4()) + endpoint = await env.create_nexus_endpoint(f"typed-{task_queue}", task_queue) + try: + async with temporalio.worker.Worker( + env.client, + task_queue=task_queue, + workflows=[TypedNexusWorkflow], + nexus_service_handlers=[TypedServiceHandler()], + ): + nexus_client = env.client.create_nexus_client( + service=TypedService, endpoint=endpoint.spec.name + ) + operations: Sequence[Any] = [TypedService.echo, TypedServiceHandler.echo] + for operation in operations: + result = await nexus_client.execute_operation( + operation, + cast(DeclaredValue, str("nexus")), + id=str(uuid4()), + schedule_to_close_timeout=timedelta(seconds=10), + ) + assert type(result) is DeclaredValue + assert result == "nexus" + assert ( + await env.client.execute_workflow( + TypedNexusWorkflow.run, + endpoint.spec.name, + id=str(uuid4()), + task_queue=task_queue, + ) + == "nexus" + ) + finally: + await env.delete_nexus_endpoint(endpoint) + + +async def test_transfer_serialization_update_with_start( + client: temporalio.client.Client, +): + task_queue = str(uuid4()) + value = cast(DeclaredValue, str("value")) + async with temporalio.worker.Worker( + client, + task_queue=task_queue, + workflows=[TypedWorkflow, TypedChildWorkflow], + activities=[typed_activity], + ): + start = temporalio.client.WithStartWorkflowOperation( + TypedWorkflow.run, + args=[value, True], + id=str(uuid4()), + task_queue=task_queue, + id_conflict_policy=temporalio.common.WorkflowIDConflictPolicy.FAIL, + ) + assert ( + await client.execute_update_with_start_workflow( + TypedWorkflow.echo_update, + value, + start_workflow_operation=start, + ) + == "value" + ) + handle = await start.workflow_handle() + await handle.signal(TypedWorkflow.finish, value) + assert await handle.result() == "value" From 3c859e95a3dfe997c40f87172e21427422126d39 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 12:48:49 -0700 Subject: [PATCH 2/4] Scope transfer converter selection to declared hints --- CHANGELOG.md | 6 +- temporalio/converter/_data_converter.py | 11 +- temporalio/converter/_payload_converter.py | 100 ++++++++------- temporalio/nexus/system/__init__.py | 10 ++ temporalio/worker/_workflow_instance.py | 4 + tests/nexus/test_temporal_system_nexus.py | 17 ++- tests/test_converter.py | 20 ++- tests/test_serialization_type_hints.py | 140 ++++++++++++++++++--- 8 files changed, 228 insertions(+), 80 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index cfa2cf757..de70255d4 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -29,9 +29,9 @@ to include examples, links to docs, or any other relevant information. ### Fixed -- Use declared argument and result types, when available, to select transfer type - converters during serialization, while preserving existing data and payload - converter method signatures. +- Use declared argument and result types to select transfer type converters during + serialization, and skip transfer conversion when no hint is available. Preserve + existing data and payload converter method signatures. ### Security diff --git a/temporalio/converter/_data_converter.py b/temporalio/converter/_data_converter.py index 4c02fd2b4..8c584ce47 100644 --- a/temporalio/converter/_data_converter.py +++ b/temporalio/converter/_data_converter.py @@ -29,7 +29,7 @@ from temporalio.converter._payload_converter import ( PayloadConverter, _TemporalTransferTypePayloadConverter, - _TypeHintedValues, + _with_serialization_type_hints, ) from temporalio.converter._serialization_context import ( SerializationContext, @@ -106,15 +106,14 @@ async def encode_with_type_hints( ) -> list[temporalio.api.common.v1.Payload]: """Encode values using declared types for transfer converter selection. - Hints correspond to values by position. A missing or None hint uses the - value's runtime type; hints for omitted arguments are ignored. + Hints correspond to values by position. A missing or None hint disables + transfer conversion for that value; hints for omitted arguments are ignored. Existing :py:meth:`encode` overrides are invoked unchanged; overrides should forward the original sequence to preserve hints when delegating to payload conversion. """ - return await self.encode( - _TypeHintedValues(values, type_hints) if type_hints is not None else values - ) + with _with_serialization_type_hints(values, type_hints): + return await self.encode(values) async def encode( self, values: Sequence[Any] diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index a15cf4532..e28aa75b0 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -4,6 +4,7 @@ import collections import collections.abc +import contextvars import dataclasses import functools import inspect @@ -13,7 +14,8 @@ import uuid import warnings from abc import ABC, abstractmethod -from collections.abc import Callable, Mapping, Sequence +from collections.abc import Callable, Iterator, Mapping, Sequence +from contextlib import contextmanager from datetime import datetime from enum import IntEnum from itertools import zip_longest @@ -57,21 +59,42 @@ _TRANSFER_TYPE_CONVERTER_ATTR = "__temporal_transfer_type_converter" -class _TypeHintedValues(list[Any]): - """Carry hints through legacy overrides without changing their signatures.""" +@dataclasses.dataclass +class _SerializationTypeHints: + values: Sequence[Any] | None + type_hints: tuple[type | None, ...] - def __init__( - self, values: Sequence[Any], type_hints: Sequence[type | None] - ) -> None: - super().__init__(values) - self.type_hints = tuple(type_hints) + +_serialization_type_hints: contextvars.ContextVar[_SerializationTypeHints | None] = ( + contextvars.ContextVar("temporal_serialization_type_hints", default=None) +) + + +@contextmanager +def _with_serialization_type_hints( + values: Sequence[Any] | None, type_hints: Sequence[type | None] | None +) -> Iterator[None]: + context = ( + _SerializationTypeHints(values, tuple(type_hints)) + if values is not None and type_hints is not None + else None + ) + token = _serialization_type_hints.set(context) + try: + yield + finally: + if context is not None: + # A task that inherited this context must not use hints after the + # originating serialization call has finished. + context.values = None + _serialization_type_hints.reset(token) class TransferTypeConverter(Generic[ValueT, TransferTypeT], ABC): """Converter between a user-facing value and a transfer type value. - When available, the declared type determines which converter is used for - serialization. Otherwise, the value's runtime type is used. + The declared type determines which converter is used for serialization. + Without a declared type, no transfer type converter is used. .. warning:: This API is experimental and subject to change. @@ -93,19 +116,6 @@ def to_transfer_type(self, value: ValueT) -> TransferTypeT: """ raise NotImplementedError - def to_transfer_type_with_type_hint( - self, - value: ValueT, - type_hint: type[ValueT] | None, # type: ignore[reportUnusedParameter] - ) -> TransferTypeT: - """Convert a value with its declared type, including generic arguments. - - The default implementation delegates to :py:meth:`to_transfer_type`. - Override this method when conversion needs the declared type. The hint - is None when the caller has no declared type available. - """ - return self.to_transfer_type(value) - @abstractmethod def from_transfer_type( self, value: TransferTypeT, type_hint: type[ValueT] @@ -170,15 +180,14 @@ def to_payloads_with_type_hints( ) -> list[temporalio.api.common.v1.Payload]: """Convert values using declared types for transfer converter selection. - Hints correspond to values by position. A missing or None hint uses the - value's runtime type; hints for omitted arguments are ignored. + Hints correspond to values by position. A missing or None hint disables + transfer conversion for that value; hints for omitted arguments are ignored. Existing :py:meth:`to_payloads` overrides are invoked unchanged; overrides should forward the original sequence to preserve hints when delegating to another payload converter. """ - return self.to_payloads( - _TypeHintedValues(values, type_hints) if type_hints is not None else values - ) + with _with_serialization_type_hints(values, type_hints): + return self.to_payloads(values) @abstractmethod def to_payloads( @@ -662,25 +671,28 @@ def to_payloads( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: """See base class.""" - transfer_type_values: list[Any] = [] + context = _serialization_type_hints.get() + # Legacy overrides can perform unrelated conversions before delegating. + # Only the original value sequence should receive the declared hints. type_hints = ( - values.type_hints - if isinstance(values, _TypeHintedValues) - else (None,) * len(values) + context.type_hints + if context is not None and context.values is values + else () ) - for index, value in enumerate(values): - type_hint = type_hints[index] if index < len(type_hints) else None - converter = ( - None - if isinstance(value, temporalio.common.RawValue) - else _get_transfer_type_converter( - type_hint if type_hint is not None else type(value) + with _with_serialization_type_hints(None, None): + transfer_type_values: list[Any] = [] + for index, value in enumerate(values): + type_hint = type_hints[index] if index < len(type_hints) else None + converter = ( + None + if type_hint is None + or isinstance(value, temporalio.common.RawValue) + else _get_transfer_type_converter(type_hint) ) - ) - if converter is not None: - value = converter.to_transfer_type_with_type_hint(value, type_hint) - transfer_type_values.append(value) - return self._inner_payload_converter.to_payloads(transfer_type_values) + if converter is not None: + value = converter.to_transfer_type(value) + transfer_type_values.append(value) + return self._inner_payload_converter.to_payloads(transfer_type_values) def from_payloads( self, diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 1f8f7c6d2..4d21594d4 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -176,6 +176,16 @@ def _get_payload_converter( # pyright: ignore[reportUnusedFunction] return _SystemNexusPayloadConverter(user_payload_converter, user_failure_converter) +def _get_input_type( # pyright: ignore[reportUnusedFunction] + service: str, operation: str +) -> type | None: + """Return the declared input type of a registered system Nexus operation.""" + from .workflow_service import __nexus_operation_registry__ + + operation_info = __nexus_operation_registry__.get((service, operation)) + return operation_info.operation.input_type if operation_info is not None else None + + def _get_serialization_context( # pyright: ignore[reportUnusedFunction] service: str, operation: str, diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 3c3e38b3c..e6fcf6baa 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2243,6 +2243,10 @@ async def _intercept_system_nexus_operation( async def _schedule_system_nexus_operation( self, input: StartNexusOperationInput[Any, OutputT] ) -> _NexusOperationHandle[OutputT]: + if input.input_type is None: + input.input_type = temporalio.nexus.system._get_input_type( + input.service, input.operation_name + ) return await self._outbound_start_nexus_operation(input) #### Miscellaneous helpers #### diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index 49200aeab..2fb580138 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -462,7 +462,9 @@ async def test_nexus_payload_serializer_decodes_system_input() -> None: payload = nexus_system._get_payload_converter( data_converter.payload_converter, data_converter.failure_converter, - ).to_payload(request) + ).to_payloads_with_type_hints( + [request], [workflow_service_models.SignalWithStartWorkflowRequest] + )[0] assert payload is not None assert payload.metadata[SYSTEM_NEXUS_PAYLOAD_METADATA_KEY] == b"true" assert payload.metadata["encoding"] == b"binary/protobuf" @@ -496,7 +498,9 @@ async def test_nexus_payload_serializer_codec_skips_outer_envelope() -> None: payload = nexus_system._get_payload_converter( data_converter.payload_converter, data_converter.failure_converter, - ).to_payload(request) + ).to_payloads_with_type_hints( + [request], [workflow_service_models.SignalWithStartWorkflowRequest] + )[0] assert payload is not None decoded = await _NexusPayloadSerializer( @@ -713,8 +717,8 @@ def test_system_nexus_uses_user_failure_converter() -> None: payload_converter, failure_converter ) - payload = system_converter.to_payload( - _FailureTransferValue(RuntimeError("test failure")) + [payload] = system_converter.to_payloads_with_type_hints( + [_FailureTransferValue(RuntimeError("test failure"))], [_FailureTransferValue] ) converted = system_converter.from_payload(payload, _FailureTransferValue) @@ -756,8 +760,9 @@ def to_failure( with nexus_system._user_converter_context(outer_converters): assert nexus_system._current_user_converters() is outer_converters with pytest.raises(ValueError, match="conversion failed"): - inner_system_converter.to_payload( - _FailureTransferValue(RuntimeError("test failure")) + inner_system_converter.to_payloads_with_type_hints( + [_FailureTransferValue(RuntimeError("test failure"))], + [_FailureTransferValue], ) assert nexus_system._current_user_converters() is outer_converters diff --git a/tests/test_converter.py b/tests/test_converter.py index 499152096..6abea7d39 100644 --- a/tests/test_converter.py +++ b/tests/test_converter.py @@ -396,7 +396,9 @@ def test_temporal_transfer_type_payload_converter_wraps_user_converter(): assert isinstance(converter, _TemporalTransferTypePayloadConverter) value = TemporalTransferTypeValue("workflow-id") - payload = converter.to_payload(value) + payload = converter.to_payloads_with_type_hints( + [value], [TemporalTransferTypeValue] + )[0] assert payload.metadata["encoding"] == b"json/protobuf" assert ( @@ -412,11 +414,23 @@ def test_temporal_transfer_type_payload_converter_wraps_user_converter(): assert plain_proto_payload.metadata["encoding"] == b"json/protobuf" +def test_temporal_transfer_type_payload_converter_without_declared_type_hint(): + converter = DataConverter.default.payload_converter + value = TemporalTransferTypeValue("workflow-id") + + payload = converter.to_payload(value) + + assert payload.metadata["encoding"] == b"json/plain" + assert converter.from_payload(payload) == {"value": "workflow-id"} + + def test_temporal_transfer_type_payload_converter_without_transfer_type_hint(): converter = DataConverter.default.payload_converter value = TemporalTransferTypeValueWithoutHint("workflow-id") - payload = converter.to_payload(value) + payload = converter.to_payloads_with_type_hints( + [value], [TemporalTransferTypeValueWithoutHint] + )[0] assert payload.metadata["encoding"] == b"json/protobuf" assert ( @@ -446,7 +460,7 @@ def test_temporal_transfer_type_payload_converter_with_generic_value( ): converter = DataConverter.default.payload_converter - payload = converter.to_payload(value) + payload = converter.to_payloads_with_type_hints([value], [type_hint])[0] assert converter.from_payload(payload, type_hint) == value diff --git a/tests/test_serialization_type_hints.py b/tests/test_serialization_type_hints.py index 807e077f2..0115248e2 100644 --- a/tests/test_serialization_type_hints.py +++ b/tests/test_serialization_type_hints.py @@ -61,10 +61,11 @@ def from_transfer_type( ("hints", "expected"), [ ([DeclaredValue], "declared:value"), + ([RuntimeValue], "runtime:value"), ([str], "value"), - ([None], "runtime:value"), - (None, "runtime:value"), - ([], "runtime:value"), + ([None], "value"), + (None, "value"), + ([], "value"), ([DeclaredValue, int], "declared:value"), ], ) @@ -89,13 +90,7 @@ class GenericConverter( transfer_type = str def to_transfer_type(self, value: GenericValue[Any]) -> str: - return "no hint" - - def to_transfer_type_with_type_hint( - self, value: GenericValue[Any], type_hint: type[GenericValue[Any]] | None - ) -> str: - assert type_hint == GenericValue[int] - return "generic hint" + return "generic converter" def from_transfer_type( self, value: str, type_hint: type[GenericValue[Any]] @@ -108,14 +103,14 @@ def from_transfer_type( async def test_transfer_serialization_generic_hint(): converter = temporalio.converter.DataConverter.default - payloads = await converter.encode_with_type_hints( - [GenericValue()], [GenericValue[int]] - ) - assert await converter.decode(payloads) == ["generic hint"] + payloads = await converter.encode_with_type_hints([object()], [GenericValue[int]]) + assert await converter.decode(payloads) == ["generic converter"] async def test_transfer_serialization_legacy_overrides_and_nested_conversion(): calls: list[str] = [] + typed_values = (RuntimeValue("typed"),) + untyped_values = [RuntimeValue("untyped")] class LegacyPayloadConverter(temporalio.converter.DefaultPayloadConverter): def to_payloads( @@ -131,7 +126,7 @@ def to_payloads( temporalio.converter.DataConverter.default.payload_converter.from_payload( nested ) - == "runtime:nested" + == "nested" ) return super().to_payloads(values) @@ -140,19 +135,128 @@ async def encode( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: calls.append("data") + assert values is typed_values or values is untyped_values + nested = await temporalio.converter.DataConverter.default.encode( + [RuntimeValue("before")] + ) + assert await temporalio.converter.DataConverter.default.decode(nested) == [ + "before" + ] await asyncio.sleep(0) return await super().encode(values) converter = LegacyDataConverter(payload_converter_class=LegacyPayloadConverter) typed, untyped = await asyncio.gather( - converter.encode_with_type_hints([RuntimeValue("typed")], [DeclaredValue]), - converter.encode([RuntimeValue("untyped")]), + converter.encode_with_type_hints(typed_values, [DeclaredValue]), + converter.encode(untyped_values), ) assert await converter.decode(typed) == ["declared:typed"] - assert await converter.decode(untyped) == ["runtime:untyped"] + assert await converter.decode(untyped) == ["untyped"] assert calls.count("data") == calls.count("payload") == 2 +def test_transfer_serialization_payload_override_preserves_sequence(): + expected_values = (RuntimeValue("value"),) + inner = temporalio.converter.DataConverter.default.payload_converter + + class LegacyPayloadConverter(temporalio.converter.DefaultPayloadConverter): + def to_payloads( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + assert values is expected_values + return inner.to_payloads(values) + + payloads = LegacyPayloadConverter().to_payloads_with_type_hints( + expected_values, [DeclaredValue] + ) + assert inner.from_payloads(payloads) == ["declared:value"] + assert inner.from_payloads(inner.to_payloads(expected_values)) == ["value"] + + +@pytest.mark.parametrize("error", [RuntimeError, asyncio.CancelledError]) +async def test_transfer_serialization_context_reset_on_error( + error: type[BaseException], +): + expected_values = [RuntimeValue("value")] + + class FailingDataConverter(temporalio.converter.DataConverter): + async def encode( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + assert values is expected_values + raise error() + + with pytest.raises(error): + await FailingDataConverter().encode_with_type_hints( + expected_values, [DeclaredValue] + ) + converter = temporalio.converter.DataConverter.default + assert await converter.decode(await converter.encode(expected_values)) == ["value"] + + +async def test_transfer_serialization_restores_outer_hints(): + expected_values = [RuntimeValue("value")] + inner = temporalio.converter.DataConverter.default + + class NestedDataConverter(temporalio.converter.DataConverter): + async def encode( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + payloads = await inner.encode_with_type_hints(values, [str]) + assert await inner.decode(payloads) == ["value"] + return await super().encode(values) + + payloads = await NestedDataConverter().encode_with_type_hints( + expected_values, [DeclaredValue] + ) + assert await inner.decode(payloads) == ["declared:value"] + + +async def test_transfer_serialization_clears_hints_before_conversion( + monkeypatch: pytest.MonkeyPatch, +): + values = [RuntimeValue("value")] + converter = temporalio.converter.DataConverter.default + original = DeclaredConverter.to_transfer_type + + def convert(self: DeclaredConverter, value: DeclaredValue) -> str: + inner = converter.payload_converter + assert inner.from_payloads(inner.to_payloads(values)) == ["value"] + return original(self, value) + + monkeypatch.setattr(DeclaredConverter, "to_transfer_type", convert) + payloads = await converter.encode_with_type_hints(values, [DeclaredValue]) + assert await converter.decode(payloads) == ["declared:value"] + + +async def test_transfer_serialization_expires_inherited_hints(): + expected_values = [RuntimeValue("value")] + ready = asyncio.Event() + converter = temporalio.converter.DataConverter.default + tasks: list[asyncio.Task[list[temporalio.api.common.v1.Payload]]] = [] + + async def later() -> list[temporalio.api.common.v1.Payload]: + await ready.wait() + return await converter.encode(expected_values) + + class SpawningDataConverter(temporalio.converter.DataConverter): + async def encode( + self, values: Sequence[Any] + ) -> list[temporalio.api.common.v1.Payload]: + tasks.append(asyncio.create_task(later())) + return await super().encode(values) + + try: + payloads = await SpawningDataConverter().encode_with_type_hints( + expected_values, [DeclaredValue] + ) + assert await converter.decode(payloads) == ["declared:value"] + finally: + ready.set() + [payloads] = await asyncio.gather(*tasks) + assert await converter.decode(payloads) == ["value"] + + async def test_transfer_serialization_raw_value(): converter = temporalio.converter.DataConverter.default [raw] = await converter.encode(["already encoded"]) From 50c38b3872a9e22e1e5fa671e04d8e55e8cc61ff Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 12:54:15 -0700 Subject: [PATCH 3/4] Clarify the serialization hint identity check --- temporalio/converter/_payload_converter.py | 5 +++-- 1 file changed, 3 insertions(+), 2 deletions(-) diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index e28aa75b0..c39a2b3c9 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -672,8 +672,9 @@ def to_payloads( ) -> list[temporalio.api.common.v1.Payload]: """See base class.""" context = _serialization_type_hints.get() - # Legacy overrides can perform unrelated conversions before delegating. - # Only the original value sequence should receive the declared hints. + # Custom DataConverter.encode implementations can serialize other values + # before calling super().encode(values). Those nested calls inherit the + # context, so only the original sequence should receive these hints. type_hints = ( context.type_hints if context is not None and context.values is values From 7e6a6ff36467776f5fe2c2ebd692e21a5ecfbb08 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Wed, 16 Sep 2026 16:33:57 -0700 Subject: [PATCH 4/4] Fix transfer serialization tests for time-skipping --- tests/test_serialization_type_hints.py | 25 ++++++++++++++++++++----- 1 file changed, 20 insertions(+), 5 deletions(-) diff --git a/tests/test_serialization_type_hints.py b/tests/test_serialization_type_hints.py index 0115248e2..d4536af1d 100644 --- a/tests/test_serialization_type_hints.py +++ b/tests/test_serialization_type_hints.py @@ -339,10 +339,6 @@ async def echo_update(self, value: DeclaredValue) -> DeclaredValue: assert type(value) is DeclaredValue return cast(DeclaredValue, str(value)) - @temporalio.workflow.query - def continued(self) -> bool: - return bool(temporalio.workflow.info().continued_run_id) - async def test_transfer_serialization_workflow_and_activity( client: temporalio.client.Client, @@ -361,12 +357,29 @@ async def test_transfer_serialization_workflow_and_activity( id=str(uuid4()), task_queue=task_queue, ) - while not await handle.query(TypedWorkflow.continued): + # A query sent during continue-as-new can remain attached to the closing + # run on the time-skipping server. + while (await handle.describe()).run_id == handle.result_run_id: await asyncio.sleep(0.01) assert await handle.query(TypedWorkflow.echo_query, value) == "value" assert await handle.execute_update(TypedWorkflow.echo_update, value) == "value" await handle.signal(TypedWorkflow.finish, value) assert await handle.result() == "value" + + +async def test_transfer_serialization_standalone_activity( + env: temporalio.testing.WorkflowEnvironment, +): + if env.supports_time_skipping: + pytest.skip( + "Standalone activities are not supported by the time-skipping server" + ) + client = env.client + task_queue = str(uuid4()) + value = cast(DeclaredValue, str("value")) + async with temporalio.worker.Worker( + client, task_queue=task_queue, activities=[typed_activity] + ): activity = await client.start_activity( typed_activity, value, @@ -427,6 +440,8 @@ async def run(self, endpoint: str) -> DeclaredValue: async def test_transfer_serialization_nexus( env: temporalio.testing.WorkflowEnvironment, ): + if env.supports_time_skipping: + pytest.skip("Nexus operations are not supported by the time-skipping server") task_queue = str(uuid4()) endpoint = await env.create_nexus_endpoint(f"typed-{task_queue}", task_queue) try: