From 9b5db6f3d44a0b4b12935146d347ac1e1699a8fe Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Thu, 29 Oct 2026 12:32:23 -0500 Subject: [PATCH 1/5] Propagate Nexus serialization context to workflows and standalone activities --- changelog/added/wandering-wombat.md | 2 + temporalio/activity.py | 14 + temporalio/client/_activity.py | 50 ++- temporalio/client/_impl.py | 19 +- temporalio/client/_workflow.py | 42 +- .../converter/_serialization_context.py | 37 +- temporalio/nexus/_operation_context.py | 49 ++- temporalio/worker/_activity.py | 59 ++- temporalio/worker/_command_aware_visitor.py | 7 + temporalio/worker/_workflow.py | 4 + temporalio/worker/_workflow_instance.py | 35 +- tests/__init__.py | 2 +- tests/test_serialization_context.py | 369 ++++++++++++++++-- 13 files changed, 594 insertions(+), 95 deletions(-) create mode 100644 changelog/added/wandering-wombat.md diff --git a/changelog/added/wandering-wombat.md b/changelog/added/wandering-wombat.md new file mode 100644 index 0000000000..e1281c2b27 --- /dev/null +++ b/changelog/added/wandering-wombat.md @@ -0,0 +1,2 @@ +Nexus operation handlers now propagate their `NexusSerializationContext` to the workflows and standalone activities they start, which encode their results and failures (and a standalone activity's heartbeat details) with it because these are returned to the Nexus caller. Clients decode them with the context the server returns, including handles obtained by ID and standalone Nexus operation handles. Requires a server that propagates the context. +Added `activity.Info.nexus_serialization_context`. To complete a Nexus-started standalone activity asynchronously, pass it to `AsyncActivityHandle.with_context`. diff --git a/temporalio/activity.py b/temporalio/activity.py index 3f69bc17f8..bf1b4cb731 100644 --- a/temporalio/activity.py +++ b/temporalio/activity.py @@ -141,6 +141,20 @@ class Info: activity_run_id: str | None = None """Run ID of this activity. None for workflow activities.""" + nexus_serialization_context: ( + temporalio.converter.NexusSerializationContext | None + ) = None + """Serialization context of the Nexus operation handler that started this activity, if any. + + Only set for standalone activities. The activity's result, failure and heartbeat details are + encoded with it because they are returned to the Nexus caller. To complete the activity + asynchronously, store this context with the task token and pass it to + :py:meth:`temporalio.client.AsyncActivityHandle.with_context`. + + .. warning:: + This API is experimental and unstable. + """ + @property def in_workflow(self) -> bool: """Was this activity started by a workflow?""" diff --git a/temporalio/client/_activity.py b/temporalio/client/_activity.py index ec127d6313..92f0d75de4 100644 --- a/temporalio/client/_activity.py +++ b/temporalio/client/_activity.py @@ -28,6 +28,7 @@ import temporalio.common import temporalio.converter import temporalio.converter._search_attributes +import temporalio.converter._serialization_context from temporalio.converter import ( ActivitySerializationContext, DataConverter, @@ -437,6 +438,18 @@ def _from_resp( **kwargs, ) + @functools.cached_property + def _caller_facing_data_converter(self) -> DataConverter: + # An activity started by a Nexus operation handler encodes its result, failures and + # heartbeat details with the handler's Nexus context, which the server returns with the + # activity info. Its input keeps the activity context. + context = temporalio.converter._serialization_context._propagated_nexus_serialization_context( + self.raw_info + ) + if context is None: + return self.data_converter + return self.data_converter.with_context(context) + def has_heartbeat_details(self) -> bool: """True if heartbeat details are available. Use `heartbeat_details` to retrieve them. @@ -453,7 +466,7 @@ async def heartbeat_details( Type hints can be provided to aid data conversion. """ return ( - await self.data_converter.decode_wrapper( + await self._caller_facing_data_converter.decode_wrapper( self.raw_info.heartbeat_details, type_hints ) if self.has_heartbeat_details() @@ -473,7 +486,9 @@ async def last_failure(self) -> BaseException | None: Always `None` if ``include_last_failure`` was false in the `ActivityHandle.describe` call. """ return ( - await self.data_converter.decode_failure(self.raw_info.last_failure) + await self._caller_facing_data_converter.decode_failure( + self.raw_info.last_failure + ) if self.has_last_failure() else None ) @@ -515,7 +530,7 @@ async def result(self, type_hint: type | None = None) -> Any | None: if self.raw_outcome is None or not self.raw_outcome.HasField("result"): return None type_hints = [type_hint] if type_hint is not None else None - results = await self.data_converter.decode_wrapper( + results = await self._caller_facing_data_converter.decode_wrapper( self.raw_outcome.result, type_hints ) if not results: @@ -543,7 +558,9 @@ async def outcome_failure(self) -> BaseException | None: Always false if ``include_outcome`` was false in the `ActivityHandle.describe` call. """ return ( - await self.data_converter.decode_failure(self.raw_outcome.failure) + await self._caller_facing_data_converter.decode_failure( + self.raw_outcome.failure + ) if self.raw_outcome is not None and self.raw_outcome.HasField("failure") else None ) @@ -942,6 +959,11 @@ def with_context(self, context: SerializationContext) -> Self: with :py:class:`ActivitySerializationContext` set as context. If you are using a custom data converter that makes use of this context then you can use this method to supply matching context data to the data converter used to serialize and encode the outbound payloads. + + A standalone activity started by a Nexus operation handler encodes its result, failure and + heartbeat details with the handler's + :py:class:`temporalio.converter.NexusSerializationContext`. To complete or heartbeat such an + activity, pass :py:attr:`temporalio.activity.Info.nexus_serialization_context`. """ data_converter = self._client.data_converter.with_context(context) if data_converter is self._client.data_converter: @@ -978,6 +1000,9 @@ def __init__( self._known_outcome: ( temporalio.api.activity.v1.ActivityExecutionOutcome | None ) = None + self._known_outcome_nexus_context: ( + temporalio.converter.NexusSerializationContext | None + ) = None @functools.cached_property def _data_converter(self) -> temporalio.converter.DataConverter: @@ -1034,18 +1059,22 @@ async def result( rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout ) - # Convert outcome to failure or value + # Convert outcome to failure or value. An activity started by a Nexus operation handler + # encodes them with the handler's Nexus context, which the server returns with the outcome. assert self._known_outcome + data_converter = ( + self._client.data_converter.with_context(self._known_outcome_nexus_context) + if self._known_outcome_nexus_context + else self._data_converter + ) if self._known_outcome.HasField("failure"): raise ActivityFailureError( - cause=await self._data_converter.decode_failure( - self._known_outcome.failure - ), + cause=await data_converter.decode_failure(self._known_outcome.failure), ) if not self._known_outcome.result.payloads: return None # type: ignore type_hints = [self._result_type] if self._result_type else None - results = await self._data_converter.decode( + results = await data_converter.decode( self._known_outcome.result.payloads, type_hints ) if not results: @@ -1080,6 +1109,9 @@ async def _poll_until_outcome( ) if res.HasField("outcome"): self._known_outcome = res.outcome + self._known_outcome_nexus_context = temporalio.converter._serialization_context._propagated_nexus_serialization_context( + res + ) return except RPCError as err: if err.status == RPCStatusCode.DEADLINE_EXCEEDED: diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index 02036ec0a8..d2585ef789 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -29,6 +29,7 @@ import temporalio.api.workflowservice.v1 import temporalio.common import temporalio.converter +import temporalio.converter._serialization_context import temporalio.exceptions import temporalio.nexus import temporalio.nexus._operation_context @@ -264,6 +265,9 @@ async def _build_update_with_start_start_workflow_execution_request( ) -> temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest: req = temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest() await self._populate_start_workflow_execution_request(req, input) + temporalio.nexus._operation_context._apply_nexus_serialization_context_to_start_workflow_request( + req + ) return req async def _populate_start_workflow_execution_request( @@ -1676,10 +1680,8 @@ async def get_nexus_operation_result( ) -> Any: """Poll for nexus operation result until it's available.""" data_converter = self._client.data_converter - # These three are set together or not at all: a handle that started the operation has all - # of them, and a handle obtained by operation ID alone has none and defaults them to "". - # An empty endpoint therefore means "no operation to build a context from", not "an - # operation named the empty string". + # Fallback for older servers that don't return the context when polling. Handles obtained + # by operation ID alone have no start request context. if input.endpoint and input.service and input.operation: data_converter = data_converter.with_context( temporalio.converter.NexusSerializationContext( @@ -1706,6 +1708,15 @@ async def get_nexus_operation_result( timeout=input.rpc_timeout, ) ) + # The context the server returns also identifies an operation whose handle was + # obtained by operation ID + server_context = temporalio.converter._serialization_context._propagated_nexus_serialization_context( + res + ) + if server_context: + data_converter = self._client.data_converter.with_context( + server_context + ) match res.WhichOneof("outcome"): case "result": type_hints = [input.result_type] if input.result_type else None diff --git a/temporalio/client/_workflow.py b/temporalio/client/_workflow.py index 94bc2bb23d..53c05c8223 100644 --- a/temporalio/client/_workflow.py +++ b/temporalio/client/_workflow.py @@ -37,6 +37,7 @@ import temporalio.common import temporalio.converter import temporalio.converter._search_attributes +import temporalio.converter._serialization_context import temporalio.exceptions import temporalio.workflow from temporalio.converter import ( @@ -136,6 +137,17 @@ def _data_converter(self) -> temporalio.converter.DataConverter: ) ) + def _close_event_data_converter( + self, events: WorkflowHistoryEventAsyncIterator + ) -> temporalio.converter.DataConverter: + # A workflow started by a Nexus operation handler encodes its result and closing failure + # with the handler's Nexus context, which the server returns with the history. An + # interceptor may return its own iterator, which then falls back to the workflow context. + context = getattr(events, "_propagated_nexus_serialization_context", None) + if context is None: + return self._data_converter + return self._client.data_converter.with_context(context) + @property def id(self) -> str: """ID of the workflow.""" @@ -225,14 +237,16 @@ async def result( # executions hist_run_id = self._result_run_id while True: - async for event in self._fetch_history_events_for_run( + events = self._fetch_history_events_for_run( hist_run_id, wait_new_event=True, event_filter_type=WorkflowHistoryEventFilterType.CLOSE_EVENT, skip_archival=True, rpc_metadata=rpc_metadata, rpc_timeout=rpc_timeout, - ): + ) + async for event in events: + data_converter = self._close_event_data_converter(events) if event.HasField("workflow_execution_completed_event_attributes"): complete_attr = event.workflow_execution_completed_event_attributes # Follow execution @@ -241,7 +255,7 @@ async def result( break # Ignoring anything after the first response like TypeScript type_hints = [self._result_type] if self._result_type else None - results = await self._data_converter.decode_wrapper( + results = await data_converter.decode_wrapper( complete_attr.result, type_hints, ) @@ -257,20 +271,14 @@ async def result( hist_run_id = fail_attr.new_execution_run_id break raise WorkflowFailureError( - cause=await self._data_converter.decode_failure( - fail_attr.failure - ), + cause=await data_converter.decode_failure(fail_attr.failure), ) elif event.HasField("workflow_execution_canceled_event_attributes"): cancel_attr = event.workflow_execution_canceled_event_attributes raise WorkflowFailureError( cause=temporalio.exceptions.CancelledError( "Workflow cancelled", - *( - await self._data_converter.decode_wrapper( - cancel_attr.details - ) - ), + *(await data_converter.decode_wrapper(cancel_attr.details)), ) ) elif event.HasField("workflow_execution_terminated_event_attributes"): @@ -278,11 +286,7 @@ async def result( raise WorkflowFailureError( cause=temporalio.exceptions.TerminatedError( term_attr.reason or "Workflow terminated", - *( - await self._data_converter.decode_wrapper( - term_attr.details - ) - ), + *(await data_converter.decode_wrapper(term_attr.details)), ), ) elif event.HasField("workflow_execution_timed_out_event_attributes"): @@ -1760,6 +1764,9 @@ def __init__( None | (Sequence[temporalio.api.history.v1.HistoryEvent]) ) = None self._current_page_index = 0 + self._propagated_nexus_serialization_context: ( + temporalio.converter.NexusSerializationContext | None + ) = None @property def current_page_index(self) -> int: @@ -1811,6 +1818,9 @@ async def fetch_next_page(self, *, page_size: int | None = None) -> None: # typ self._current_page = list(resp.history.events) self._current_page_index = 0 self._next_page_token = resp.next_page_token or None + self._propagated_nexus_serialization_context = temporalio.converter._serialization_context._propagated_nexus_serialization_context( + resp + ) def __aiter__(self) -> WorkflowHistoryEventAsyncIterator: """Return self as the iterator.""" diff --git a/temporalio/converter/_serialization_context.py b/temporalio/converter/_serialization_context.py index 8046a814c8..73d19a5b43 100644 --- a/temporalio/converter/_serialization_context.py +++ b/temporalio/converter/_serialization_context.py @@ -5,6 +5,7 @@ from abc import ABC from dataclasses import dataclass +import google.protobuf.message from typing_extensions import Self @@ -102,15 +103,17 @@ class ActivitySerializationContext(SerializationContext): class NexusSerializationContext(SerializationContext): """Serialization context for Nexus operation payloads. - Callers receive this context when encoding inputs and decoding results or failures. The context - is not propagated to a handler that completes an asynchronous operation. Handlers receive it - when decoding inputs, encoding synchronous results, and encoding failures produced while - handling a Nexus task. + Callers receive this context when encoding inputs and decoding results or failures. Handlers + receive it when decoding inputs, encoding synchronous results, and encoding failures produced + while handling a Nexus task. - A standalone operation handle retains the context used to start the operation and uses it to - decode the result, including when the start request returns an existing operation. A handle - created with :py:meth:`temporalio.client.Client.get_nexus_operation_handle` has no endpoint, - service, or operation information and therefore decodes without Nexus context. + Workflows and standalone activities started by a handler receive this context through the + server and use it for payloads returned to the caller: results, failures, and standalone + activity heartbeat details. Clients decode these with the context the server returns. See + :py:attr:`temporalio.activity.Info.nexus_serialization_context` for asynchronous completion. + + A standalone operation handle decodes with the context the server returns, falling back to the + context used to start the operation on older servers. A failure encoded by a handler is later decoded by a caller. Because some operation paths may lack this context, contextual encodings must be self-describing and decoders must continue to @@ -130,6 +133,24 @@ class NexusSerializationContext(SerializationContext): """Nexus operation name.""" +def _propagated_nexus_serialization_context( # pyright: ignore[reportUnusedFunction] + message: google.protobuf.message.Message, +) -> NexusSerializationContext | None: + """Return the Nexus serialization context the server propagated on a message, if any. + + The server sets ``propagated_nexus_serialization_context`` on executions started by a Nexus + operation handler. + """ + if not message.HasField("propagated_nexus_serialization_context"): + return None + context = getattr(message, "propagated_nexus_serialization_context") + return NexusSerializationContext( + endpoint=context.endpoint, + service=context.service, + operation=context.operation, + ) + + class WithSerializationContext(ABC): """Interface for classes that can use serialization context. diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py index 76e2b2fd68..bd287e4e18 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -32,6 +32,7 @@ from typing_extensions import Self import temporalio.api.common.v1 +import temporalio.api.nexus.v1 import temporalio.api.workflowservice.v1 import temporalio.common from temporalio.types import ( @@ -173,6 +174,25 @@ def _try_start_operation_context() -> _TemporalStartOperationContext | None: # return _temporal_start_operation_context.get(None) +def _current_propagated_nexus_serialization_context() -> ( # pyright: ignore[reportUnusedFunction] + temporalio.api.nexus.v1.PropagatedSerializationContext | None +): + """The serialization context of the Nexus operation being handled, in the form sent with start + requests, or None outside a Nexus start operation handler. + + A workflow or standalone activity started by the handler returns its result and failure to the + Nexus caller, so the server propagates this context to the worker and to clients reading them. + """ + nexus_ctx = _try_start_operation_context() + if nexus_ctx is None: + return None + return temporalio.api.nexus.v1.PropagatedSerializationContext( + endpoint=nexus_ctx.info().endpoint, + service=nexus_ctx.nexus_context.service, + operation=nexus_ctx.nexus_context.operation, + ) + + @contextmanager def _nexus_backing_start_context() -> Generator[None]: token = _temporal_nexus_backing_start_context.set(True) @@ -821,14 +841,15 @@ def _apply_nexus_context_to_start_workflow_request( # pyright: ignore[reportUnu ) -> None: """Apply the current Nexus operation context to a workflow start request. - This is a no-op outside a Nexus operation context. Within one, it attaches - inbound links and configures conflict handling to preserve Nexus metadata. - The Nexus request ID and completion callbacks are added only when the + This is a no-op outside a Nexus operation context. Within one, it sends the + operation's serialization context, attaches inbound links and configures + conflict handling to preserve Nexus metadata. The Nexus request ID and completion callbacks are added only when the workflow is backing the Nexus operation. on_conflict_options is populated only when there are links or callbacks to attach. """ nexus_ctx = _try_start_operation_context() if nexus_ctx is not None: + _apply_nexus_serialization_context_to_start_workflow_request(req) request_links = nexus_ctx._get_request_links() # Links are duplicated on request for compatibility with older server versions. @@ -861,6 +882,18 @@ def _apply_nexus_context_to_start_workflow_request( # pyright: ignore[reportUnu req.on_conflict_options.attach_links = True +def _apply_nexus_serialization_context_to_start_workflow_request( # pyright: ignore[reportUnusedFunction] + req: temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest, +) -> None: + """Send the current Nexus operation's serialization context with a workflow start request. + + This is a no-op outside a Nexus operation context. + """ + context = _current_propagated_nexus_serialization_context() + if context is not None: + req.propagated_nexus_serialization_context.CopyFrom(context) + + def _apply_start_workflow_response_to_nexus_context( # pyright: ignore[reportUnusedFunction] workflow_handle: temporalio.client.WorkflowHandle[Any, Any], ) -> None: @@ -875,15 +908,19 @@ def _apply_nexus_context_to_start_activity_request( # pyright: ignore[reportUnu """Apply the current Nexus operation context to an activity start request. This is a no-op outside a Nexus operation context. Within one, it attaches - the Nexus request ID and configures conflict handling to preserve the Nexus - metadata. Inbound links are attached to the completion callback when the - activity backs the operation and has a callback, and to the request otherwise. + the Nexus request ID and the operation's serialization context and + configures conflict handling to preserve the Nexus metadata. Inbound links + are attached to the completion callback when the activity backs the + operation and has a callback, and to the request otherwise. on_conflict_options is populated only when there are links or callbacks to attach. """ nexus_ctx = _try_start_operation_context() if nexus_ctx is not None: req.request_id = nexus_ctx.nexus_context.request_id + context = _current_propagated_nexus_serialization_context() + if context is not None: + req.propagated_nexus_serialization_context.CopyFrom(context) request_links = nexus_ctx._get_request_links() callbacks: list[NexusCallback] = [] diff --git a/temporalio/worker/_activity.py b/temporalio/worker/_activity.py index 2730534ebe..2af18dc5fa 100644 --- a/temporalio/worker/_activity.py +++ b/temporalio/worker/_activity.py @@ -32,6 +32,7 @@ import temporalio.client import temporalio.common import temporalio.converter +import temporalio.converter._serialization_context import temporalio.exceptions from temporalio.converter import ( StorageDriverActivityInfo, @@ -277,14 +278,19 @@ async def _send_latest_heartbeat( data_converter = self._data_converter if activity.info: - context = temporalio.converter.ActivitySerializationContext( - namespace=activity.info.namespace, - workflow_id=activity.info.workflow_id, - workflow_type=activity.info.workflow_type, - activity_type=activity.info.activity_type, - activity_id=activity.info.activity_id, - activity_task_queue=self._task_queue, - is_local=activity.info.is_local, + # The server puts a standalone activity's last heartbeat details into its timeout + # failure, which goes back to the Nexus caller if a Nexus handler started it + context: temporalio.converter.SerializationContext = ( + activity.info.nexus_serialization_context + or temporalio.converter.ActivitySerializationContext( + namespace=activity.info.namespace, + workflow_id=activity.info.workflow_id, + workflow_type=activity.info.workflow_type, + activity_type=activity.info.activity_type, + activity_id=activity.info.activity_id, + activity_task_queue=self._task_queue, + is_local=activity.info.is_local, + ) ) data_converter = data_converter._with_contexts( context, @@ -374,17 +380,35 @@ async def _handle_start_activity_task( data_converter = self._data_converter._with_contexts( context, StorageDriverStoreContext(target=store_target) ) + # A standalone activity started by a Nexus operation handler returns its result and + # failure to the Nexus caller, which decodes them with the handler's Nexus context. The + # input keeps the activity context. + nexus_context = temporalio.converter._serialization_context._propagated_nexus_serialization_context( + start + ) + caller_facing_data_converter = ( + self._data_converter._with_contexts( + nexus_context, StorageDriverStoreContext(target=store_target) + ) + if nexus_context + else data_converter + ) try: try: result = await self._execute_activity( - start, running_activity, task_token, data_converter + start, + running_activity, + task_token, + data_converter, + caller_facing_data_converter, + nexus_context, ) finally: # The activity code has finished, so a cancel from here on must # not interrupt reporting its outcome; done stops cancel() from # cancelling this task running_activity.done = True - [payload] = await data_converter.encode([result]) + [payload] = await caller_facing_data_converter.encode([result]) completion.result.completed.result.CopyFrom(payload) except BaseException as err: try: @@ -405,7 +429,7 @@ async def _handle_start_activity_task( temporalio.activity.logger.warning( f"Completing as failure during heartbeat with error of type {type(err)}: {err}", ) - await data_converter.encode_failure( + await caller_facing_data_converter.encode_failure( err, completion.result.failed.failure ) elif ( @@ -422,7 +446,7 @@ async def _handle_start_activity_task( temporalio.activity.logger.warning( "Completing as failure due to unhandled cancel error produced by activity pause", ) - await data_converter.encode_failure( + await caller_facing_data_converter.encode_failure( temporalio.exceptions.ApplicationError( type="ActivityPause", message="Unhandled activity cancel error produced by activity pause", @@ -443,7 +467,7 @@ async def _handle_start_activity_task( temporalio.activity.logger.warning( "Completing as failure due to unhandled cancel error produced by activity reset", ) - await data_converter.encode_failure( + await caller_facing_data_converter.encode_failure( temporalio.exceptions.ApplicationError( type="ActivityReset", message="Unhandled activity cancel error produced by activity reset", @@ -461,7 +485,7 @@ async def _handle_start_activity_task( and running_activity.cancelled_by_request ): temporalio.activity.logger.debug("Completing as cancelled") - await data_converter.encode_failure( + await caller_facing_data_converter.encode_failure( # TODO(cretz): Should use some other message? temporalio.exceptions.CancelledError("Cancelled"), completion.result.cancelled.failure, @@ -487,7 +511,7 @@ async def _handle_start_activity_task( exc_info=True, extra={"__temporal_error_identifier": "ActivityFailure"}, ) - await data_converter.encode_failure( + await caller_facing_data_converter.encode_failure( err, completion.result.failed.failure ) # For broken executors, we have to fail the entire worker @@ -528,6 +552,8 @@ async def _execute_activity( running_activity: _RunningActivity, task_token: bytes, data_converter: temporalio.converter.DataConverter, + caller_facing_data_converter: temporalio.converter.DataConverter, + nexus_context: temporalio.converter.NexusSerializationContext | None, ) -> Any: """Invoke the user's activity function. @@ -617,7 +643,7 @@ async def _execute_activity( heartbeat_details = ( [] if not start.heartbeat_details - else await data_converter.decode(start.heartbeat_details) + else await caller_facing_data_converter.decode(start.heartbeat_details) ) except Exception as err: raise temporalio.exceptions.ApplicationError( @@ -664,6 +690,7 @@ async def _execute_activity( activity_run_id=getattr(start, "run_id", None) if not started_by_workflow else None, + nexus_serialization_context=nexus_context, ) if self._encode_headers: diff --git a/temporalio/worker/_command_aware_visitor.py b/temporalio/worker/_command_aware_visitor.py index 7c03c2cd43..69667079c3 100644 --- a/temporalio/worker/_command_aware_visitor.py +++ b/temporalio/worker/_command_aware_visitor.py @@ -19,6 +19,7 @@ ) from temporalio.bridge.proto.workflow_commands.workflow_commands_pb2 import ( CompleteWorkflowExecution, + FailWorkflowExecution, ScheduleActivity, ScheduleLocalActivity, ScheduleNexusOperation, @@ -78,6 +79,12 @@ async def _visit_coresdk_workflow_commands_CompleteWorkflowExecution( fs, o ) + async def _visit_coresdk_workflow_commands_FailWorkflowExecution( + self, fs: VisitorFunctions, o: FailWorkflowExecution + ) -> None: + with current_command(CommandType.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION, 0): + await super()._visit_coresdk_workflow_commands_FailWorkflowExecution(fs, o) + async def _visit_coresdk_workflow_commands_ScheduleActivity( self, fs: VisitorFunctions, o: ScheduleActivity ) -> None: diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py index 3e6ab1e6a8..5798a3b0ac 100644 --- a/temporalio/worker/_workflow.py +++ b/temporalio/worker/_workflow.py @@ -23,6 +23,7 @@ import temporalio.common import temporalio.converter import temporalio.converter._extstore +import temporalio.converter._serialization_context import temporalio.exceptions import temporalio.workflow from temporalio.bridge.worker import PollShutdownError @@ -741,6 +742,9 @@ def _create_workflow_instance( last_completion_result=init.last_completion_result, last_failure=last_failure, default_workflow_logic_flags=frozenset(self._default_workflow_logic_flags), + nexus_serialization_context=temporalio.converter._serialization_context._propagated_nexus_serialization_context( + init + ), ) if defn.sandboxed: return self._workflow_runner.create_instance(det) diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 5bcda83f9d..43a56f5af9 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -181,6 +181,10 @@ class WorkflowInstanceDetails: default_workflow_logic_flags: frozenset[_WorkflowLogicFlag] = field( default_factory=lambda: _DEFAULT_ENABLED_WORKFLOW_LOGIC_FLAGS ) + nexus_serialization_context: ( + temporalio.converter.NexusSerializationContext | None + ) = None + """Serialization context of the Nexus operation handler that started the workflow, if any.""" class WorkflowInstance(ABC): @@ -293,6 +297,19 @@ def __init__(self, det: WorkflowInstanceDetails) -> None: self._workflow_context_failure_converter = self._failure_converter_with_context( workflow_context ) + # A workflow started by a Nexus operation handler returns its result and closing failure to + # the Nexus caller, which decodes them with the handler's Nexus context. + self._nexus_serialization_context = det.nexus_serialization_context + if self._nexus_serialization_context: + self._result_payload_converter = self._payload_converter_with_context( + self._nexus_serialization_context + ) + self._result_failure_converter = self._failure_converter_with_context( + self._nexus_serialization_context + ) + else: + self._result_payload_converter = self._workflow_context_payload_converter + self._result_failure_converter = self._workflow_context_failure_converter self._extern_functions = det.extern_functions self._disable_eager_activity_execution = det.disable_eager_activity_execution @@ -1173,9 +1190,7 @@ 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._result_payload_converter.to_payloads([result]) if len(result_payloads) != 1: raise ValueError( f"Expected 1 result payload, got {len(result_payloads)}" @@ -2479,6 +2494,13 @@ def get_serialization_context( workflow_id=self._info.workflow_id, ) + if self._nexus_serialization_context and command_info.command_type in ( + temporalio.api.enums.v1.command_type_pb2.CommandType.COMMAND_TYPE_COMPLETE_WORKFLOW_EXECUTION, + temporalio.api.enums.v1.command_type_pb2.CommandType.COMMAND_TYPE_FAIL_WORKFLOW_EXECUTION, + ): + # The result and closing failure go back to the Nexus caller + return self._nexus_serialization_context + if ( command_info.command_type == temporalio.api.enums.v1.command_type_pb2.CommandType.COMMAND_TYPE_SCHEDULE_ACTIVITY_TASK @@ -2896,8 +2918,8 @@ def _set_workflow_failure(self, err: BaseException) -> None: failure = self._add_command().fail_workflow_execution.failure failure.SetInParent() try: - self._workflow_context_failure_converter.to_failure( - err, self._workflow_context_payload_converter, failure + self._result_failure_converter.to_failure( + err, self._result_payload_converter, failure ) except Exception as inner_err: raise ValueError("Failed converting workflow exception") from inner_err @@ -3137,7 +3159,10 @@ def create_task( *, name: str | None = None, context: contextvars.Context | None = None, + eager_start: bool | None = None, ) -> asyncio.Task[_T]: + # eager_start (Python 3.14+) is ignored because starting a task eagerly would change the + # deterministic order in which workflow tasks run. # Context only supported on newer Python versions if sys.version_info >= (3, 11): task = asyncio.Task(coro, loop=self, context=context) # type: ignore diff --git a/tests/__init__.py b/tests/__init__.py index 84f4d78821..4d76781a29 100644 --- a/tests/__init__.py +++ b/tests/__init__.py @@ -1 +1 @@ -DEV_SERVER_DOWNLOAD_VERSION = "v1.9.2-nexus-handler-callbacks" +DEV_SERVER_DOWNLOAD_VERSION = "v1.9.3-nexus-serialization-context" diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py index 5f304b2e7e..9ace8ff14c 100644 --- a/tests/test_serialization_context.py +++ b/tests/test_serialization_context.py @@ -24,8 +24,10 @@ import temporalio.api.common.v1 import temporalio.api.failure.v1 -from temporalio import activity, workflow +import temporalio.api.nexus.v1 +from temporalio import activity, nexus, workflow from temporalio.client import ( + ActivityFailureError, AsyncActivityHandle, Client, GetNexusOperationResultInput, @@ -56,6 +58,8 @@ from temporalio.testing import WorkflowEnvironment from temporalio.worker import Replayer, Worker from temporalio.worker._workflow_instance import UnsandboxedWorkflowRunner +from tests.helpers import assert_eq_eventually +from tests.helpers.nexus import make_nexus_endpoint_name @dataclass @@ -1723,25 +1727,14 @@ def __init__( self, markers: dict[NexusSerializationContext, bytes], context: SerializationContext | None = None, - allow_contextless_decode_of_marked_payload: bool = False, ): self.markers = markers self.context = context - # A handle obtained by operation ID decodes a payload that was encoded under - # a Nexus context without one. Every other caller must decode under the same context it - # encoded with, so that direction is an error unless a test opts out here. - self.allow_contextless_decode_of_marked_payload = ( - allow_contextless_decode_of_marked_payload - ) def with_context( self, context: SerializationContext ) -> NexusContextMarkerPayloadCodec: - return NexusContextMarkerPayloadCodec( - self.markers, - context, - self.allow_contextless_decode_of_marked_payload, - ) + return NexusContextMarkerPayloadCodec(self.markers, context) def _marker(self) -> bytes | None: if not isinstance(self.context, NexusSerializationContext): @@ -1772,12 +1765,12 @@ async def decode( ) -> list[temporalio.api.common.v1.Payload]: marker = self._marker() if marker is None: - if not self.allow_contextless_decode_of_marked_payload: - for payload in payloads: - assert self.MARKER_KEY not in payload.metadata, ( - f"payload encoded under a Nexus context was decoded under " - f"{self.context!r}" - ) + # Every reader must decode under the same Nexus context the payload was encoded with + for payload in payloads: + assert self.MARKER_KEY not in payload.metadata, ( + f"payload encoded under a Nexus context was decoded under " + f"{self.context!r}" + ) return list(payloads) decoded = [] for payload in payloads: @@ -2000,14 +1993,11 @@ async def test_workflow_nexus_payload_codec_receives_context( @pytest.mark.requires_local_server -async def test_standalone_nexus_handle_obtained_by_id_has_no_context( +async def test_standalone_nexus_handle_obtained_by_id_uses_server_context( env: WorkflowEnvironment, ): - """A handle obtained by operation ID alone decodes its result without a Nexus context. - - Such a handle never issued a start request, so it has no endpoint, service or operation to - build a context from. The result was still encoded by the handler under the operation's - context, so this is the one asymmetry the strict codec has to tolerate. + """A handle obtained by operation ID alone decodes its result with the context the server + returns, since it never issued a start request to build a context from. """ if env.supports_time_skipping_v1: pytest.skip("Nexus tests don't work with the Java test server") @@ -2022,10 +2012,7 @@ async def test_standalone_nexus_handle_obtained_by_id_has_no_context( config = env.client.config() config["data_converter"] = dataclasses.replace( DataConverter.default, - payload_codec=NexusContextMarkerPayloadCodec( - {context: b"detached"}, - allow_contextless_decode_of_marked_payload=True, - ), + payload_codec=NexusContextMarkerPayloadCodec({context: b"detached"}), ) client = Client(**config) @@ -2047,7 +2034,7 @@ async def test_standalone_nexus_handle_obtained_by_id_has_no_context( ) assert await started.result() == "detached" - # The fresh handle knows only the operation ID, so it decodes without a context. + # The strict codec rejects a marked result decoded without the operation's context detached = client.get_nexus_operation_handle(operation_id, result_type=str) assert await detached.result() == "detached" @@ -2561,3 +2548,325 @@ async def test_nexus_worker_task_without_endpoint_is_scoped_by_an_empty_endpoint endpoint="endpoint", service="Service", operation="operation" ) ] + + +@dataclass +class NexusStartedInput: + id: str + value: str + outcome: Literal["result", "failure", "continue_as_new", "async"] = "result" + + +@workflow.defn +class NexusStartedWorkflow: + @workflow.run + async def run(self, input: NexusStartedInput) -> str: + if input.outcome == "continue_as_new": + workflow.continue_as_new(dataclasses.replace(input, outcome="result")) + if input.outcome == "failure": + raise ApplicationError( + "nexus-started-failure", "failure-detail", non_retryable=True + ) + return input.value + + +nexus_started_activity_contexts: list[NexusSerializationContext | None] = [] +nexus_started_async_completions: list[ + tuple[bytes, NexusSerializationContext | None] +] = [] + + +@activity.defn +async def nexus_started_activity(input: NexusStartedInput) -> str: + info = activity.info() + nexus_started_activity_contexts.append(info.nexus_serialization_context) + if input.outcome == "failure": + raise ApplicationError( + "nexus-started-failure", "failure-detail", non_retryable=True + ) + if input.outcome == "async": + nexus_started_async_completions.append( + (info.task_token, info.nexus_serialization_context) + ) + activity.raise_complete_async() + if info.attempt == 1: + # The next attempt decodes these details with the same context + activity.heartbeat("heartbeat-detail") + raise ApplicationError("retry") + assert list(info.heartbeat_details) == ["heartbeat-detail"] + return input.value + + +@nexusrpc.handler.service_handler +class NexusStartedExecutionHandler: + @nexus.temporal_operation + async def start_workflow( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: NexusStartedInput, + ) -> nexus.TemporalOperationResult[str]: + return await client.start_workflow(NexusStartedWorkflow.run, input, id=input.id) + + @nexus.temporal_operation + async def start_activity( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: NexusStartedInput, + ) -> nexus.TemporalOperationResult[str]: + return await client.start_activity( + nexus_started_activity, + input, + id=input.id, + start_to_close_timeout=timedelta(seconds=10), + retry_policy=RetryPolicy( + initial_interval=timedelta(milliseconds=100), maximum_attempts=2 + ), + ) + + +def _nexus_started_context( + endpoint_name: str, operation: str +) -> NexusSerializationContext: + return NexusSerializationContext( + endpoint=endpoint_name, + service="NexusStartedExecutionHandler", + operation=operation, + ) + + +def _nexus_started_client( + env: WorkflowEnvironment, context: NexusSerializationContext +) -> Client: + config = env.client.config() + config["data_converter"] = dataclasses.replace( + DataConverter.default, + payload_codec=NexusContextMarkerPayloadCodec({context: b"nexus"}), + ) + return Client(**config) + + +def _marker(payload: temporalio.api.common.v1.Payload) -> bytes | None: + return payload.metadata.get(NexusContextMarkerPayloadCodec.MARKER_KEY) + + +@pytest.mark.requires_local_server +@pytest.mark.parametrize("outcome", ["result", "failure", "continue_as_new"]) +async def test_nexus_started_workflow_uses_nexus_context( + env: WorkflowEnvironment, + outcome: Literal["result", "failure", "continue_as_new"], +): + """A workflow started by a Nexus handler encodes its result and closing failure with the + handler's context, and clients decode them with the context the server returns.""" + if env.supports_time_skipping_v1: + pytest.skip("Nexus tests don't work with the Java test server") + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + context = _nexus_started_context(endpoint_name, "start_workflow") + client = _nexus_started_client(env, context) + workflow_id = f"nexus-started-workflow-{uuid.uuid4()}" + + async with Worker( + client, + task_queue=task_queue, + workflows=[NexusStartedWorkflow], + nexus_service_handlers=[NexusStartedExecutionHandler()], + ): + await env.create_nexus_endpoint(endpoint_name, task_queue) + nexus_client = client.create_nexus_client( + NexusStartedExecutionHandler, endpoint_name + ) + input = NexusStartedInput(id=workflow_id, value="nexus-value", outcome=outcome) + # This handle never started the workflow, so it relies on the context the server + # returns, including for the run after continue-as-new + handle = client.get_workflow_handle(workflow_id, result_type=str) + if outcome == "failure": + with pytest.raises(NexusOperationFailureError): + await nexus_client.execute_operation( + NexusStartedExecutionHandler.start_workflow, + input, + id=str(uuid.uuid4()), + ) + with pytest.raises(WorkflowFailureError) as err: + await handle.result() + assert isinstance(err.value.cause, ApplicationError) + assert err.value.cause.details == ("failure-detail",) + else: + assert ( + await nexus_client.execute_operation( + NexusStartedExecutionHandler.start_workflow, + input, + id=str(uuid.uuid4()), + ) + == "nexus-value" + ) + assert await handle.result() == "nexus-value" + + history = await handle.fetch_history() + started = history.events[0].workflow_execution_started_event_attributes + assert ( + started.propagated_nexus_serialization_context + == temporalio.api.nexus.v1.PropagatedSerializationContext( + endpoint=context.endpoint, + service=context.service, + operation=context.operation, + ) + ) + # The input keeps the workflow context + assert all(_marker(p) is None for p in started.input.payloads) + close_event = history.events[-1] + if outcome == "failure": + failure = close_event.workflow_execution_failed_event_attributes.failure + [detail] = failure.application_failure_info.details.payloads + assert _marker(detail) == b"nexus" + else: + [result] = ( + close_event.workflow_execution_completed_event_attributes.result.payloads + ) + assert _marker(result) == b"nexus" + + +@pytest.mark.requires_local_server +async def test_nexus_started_activity_uses_nexus_context(env: WorkflowEnvironment): + """A standalone activity started by a Nexus handler encodes its result, failures and + heartbeat details with the handler's context, and clients decode them with the context the + server returns.""" + if env.supports_time_skipping_v1: + pytest.skip("Nexus tests don't work with the Java test server") + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + context = _nexus_started_context(endpoint_name, "start_activity") + client = _nexus_started_client(env, context) + activity_id = f"nexus-started-activity-{uuid.uuid4()}" + nexus_started_activity_contexts.clear() + + async with Worker( + client, + task_queue=task_queue, + activities=[nexus_started_activity], + nexus_service_handlers=[NexusStartedExecutionHandler()], + ): + await env.create_nexus_endpoint(endpoint_name, task_queue) + result = await client.create_nexus_client( + NexusStartedExecutionHandler, endpoint_name + ).execute_operation( + NexusStartedExecutionHandler.start_activity, + NexusStartedInput(id=activity_id, value="nexus-value"), + id=str(uuid.uuid4()), + ) + assert result == "nexus-value" + # Both attempts saw the context, and the second decoded the first's heartbeat details + assert nexus_started_activity_contexts == [context, context] + + handle = client.get_activity_handle(activity_id, result_type=str) + assert await handle.result() == "nexus-value" + desc = await handle.describe( + include_outcome=True, + include_heartbeat_details=True, + include_last_failure=True, + ) + assert await desc.result(str) == "nexus-value" + assert await desc.heartbeat_details() == ["heartbeat-detail"] + assert isinstance(await desc.last_failure(), ApplicationError) + assert desc.raw_outcome is not None + assert _marker(desc.raw_outcome.result.payloads[0]) == b"nexus" + assert _marker(desc.raw_info.heartbeat_details.payloads[0]) == b"nexus" + + +@pytest.mark.requires_local_server +async def test_nexus_started_activity_failure_uses_nexus_context( + env: WorkflowEnvironment, +): + if env.supports_time_skipping_v1: + pytest.skip("Nexus tests don't work with the Java test server") + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + client = _nexus_started_client( + env, _nexus_started_context(endpoint_name, "start_activity") + ) + activity_id = f"nexus-started-activity-{uuid.uuid4()}" + + async with Worker( + client, + task_queue=task_queue, + activities=[nexus_started_activity], + nexus_service_handlers=[NexusStartedExecutionHandler()], + ): + await env.create_nexus_endpoint(endpoint_name, task_queue) + with pytest.raises(NexusOperationFailureError): + await client.create_nexus_client( + NexusStartedExecutionHandler, endpoint_name + ).execute_operation( + NexusStartedExecutionHandler.start_activity, + NexusStartedInput(id=activity_id, value="", outcome="failure"), + id=str(uuid.uuid4()), + ) + + handle = client.get_activity_handle(activity_id, result_type=str) + with pytest.raises(ActivityFailureError) as err: + await handle.result() + assert isinstance(err.value.cause, ApplicationError) + assert err.value.cause.details == ("failure-detail",) + desc = await handle.describe(include_outcome=True) + outcome_failure = await desc.outcome_failure() + assert isinstance(outcome_failure, ApplicationError) + assert outcome_failure.details == ("failure-detail",) + + +@pytest.mark.requires_local_server +async def test_nexus_started_activity_async_completion_with_context( + env: WorkflowEnvironment, +): + """An activity started by a Nexus handler and completed asynchronously uses the context the + completing client passes in from the activity info.""" + if env.supports_time_skipping_v1: + pytest.skip("Nexus tests don't work with the Java test server") + + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + context = _nexus_started_context(endpoint_name, "start_activity") + client = _nexus_started_client(env, context) + activity_id = f"nexus-started-activity-{uuid.uuid4()}" + nexus_started_async_completions.clear() + + async with Worker( + client, + task_queue=task_queue, + activities=[nexus_started_activity], + nexus_service_handlers=[NexusStartedExecutionHandler()], + ): + await env.create_nexus_endpoint(endpoint_name, task_queue) + operation = await client.create_nexus_client( + NexusStartedExecutionHandler, endpoint_name + ).start_operation( + NexusStartedExecutionHandler.start_activity, + NexusStartedInput(id=activity_id, value="", outcome="async"), + id=str(uuid.uuid4()), + ) + + async def activity_started() -> bool: + return bool(nexus_started_async_completions) + + await assert_eq_eventually(True, activity_started) + [(task_token, info_context)] = nexus_started_async_completions + assert info_context is not None + assert info_context == context + async_handle = client.get_async_activity_handle( + task_token=task_token + ).with_context(info_context) + await async_handle.heartbeat("client-heartbeat") + desc = await client.get_activity_handle(activity_id).describe( + include_heartbeat_details=True + ) + assert await desc.heartbeat_details() == ["client-heartbeat"] + assert _marker(desc.raw_info.heartbeat_details.payloads[0]) == b"nexus" + await async_handle.complete("async-value") + assert await operation.result() == "async-value" + assert ( + await client.get_activity_handle(activity_id, result_type=str).result() + == "async-value" + ) From bd0d5df239f331cab01b293ae283165efe82924c Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Thu, 29 Oct 2026 12:35:33 -0500 Subject: [PATCH 2/5] fixing eager_start field comment --- temporalio/worker/_workflow_instance.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 43a56f5af9..3e6aca853a 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -3159,10 +3159,9 @@ def create_task( *, name: str | None = None, context: contextvars.Context | None = None, + # Accepted for Python 3.14 signature compatibility; ignored eager_start: bool | None = None, ) -> asyncio.Task[_T]: - # eager_start (Python 3.14+) is ignored because starting a task eagerly would change the - # deterministic order in which workflow tasks run. # Context only supported on newer Python versions if sys.version_info >= (3, 11): task = asyncio.Task(coro, loop=self, context=context) # type: ignore From 374dfbfee47a8b9eaebde22207dd4ad419c949af Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Thu, 29 Oct 2026 14:25:12 -0500 Subject: [PATCH 3/5] Keep the Nexus serialization context across workflow history pages --- temporalio/client/_workflow.py | 6 +++- tests/test_serialization_context.py | 54 +++++++++++++++++++++++++++++ 2 files changed, 59 insertions(+), 1 deletion(-) diff --git a/temporalio/client/_workflow.py b/temporalio/client/_workflow.py index 53c05c8223..fc20210987 100644 --- a/temporalio/client/_workflow.py +++ b/temporalio/client/_workflow.py @@ -1818,9 +1818,13 @@ async def fetch_next_page(self, *, page_size: int | None = None) -> None: # typ self._current_page = list(resp.history.events) self._current_page_index = 0 self._next_page_token = resp.next_page_token or None - self._propagated_nexus_serialization_context = temporalio.converter._serialization_context._propagated_nexus_serialization_context( + # Pages after the first may omit the context, such as the page holding the close event + # after a long poll times out + context = temporalio.converter._serialization_context._propagated_nexus_serialization_context( resp ) + if context is not None: + self._propagated_nexus_serialization_context = context def __aiter__(self) -> WorkflowHistoryEventAsyncIterator: """Return self as the iterator.""" diff --git a/tests/test_serialization_context.py b/tests/test_serialization_context.py index 9ace8ff14c..819ee483c6 100644 --- a/tests/test_serialization_context.py +++ b/tests/test_serialization_context.py @@ -24,7 +24,9 @@ import temporalio.api.common.v1 import temporalio.api.failure.v1 +import temporalio.api.history.v1 import temporalio.api.nexus.v1 +import temporalio.api.workflowservice.v1 from temporalio import activity, nexus, workflow from temporalio.client import ( ActivityFailureError, @@ -2651,6 +2653,58 @@ def _marker(payload: temporalio.api.common.v1.Payload) -> bytes | None: return payload.metadata.get(NexusContextMarkerPayloadCodec.MARKER_KEY) +async def test_workflow_result_keeps_nexus_context_across_history_pages(): + """The server may omit the context on later history pages, such as the page holding the + close event after a long poll times out with no events.""" + from unittest import mock + + context = NexusSerializationContext( + endpoint="endpoint", service="service", operation="operation" + ) + result = temporalio.api.common.v1.Payload( + metadata={ + "encoding": b"json/plain", + NexusContextMarkerPayloadCodec.MARKER_KEY: b"nexus", + }, + data=b'"nexus-value"', + ) + first_page = temporalio.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse( + next_page_token=b"next-page", + propagated_nexus_serialization_context=temporalio.api.nexus.v1.PropagatedSerializationContext( + endpoint=context.endpoint, + service=context.service, + operation=context.operation, + ), + ) + close_page = temporalio.api.workflowservice.v1.GetWorkflowExecutionHistoryResponse( + history=temporalio.api.history.v1.History( + events=[ + temporalio.api.history.v1.HistoryEvent( + workflow_execution_completed_event_attributes=temporalio.api.history.v1.WorkflowExecutionCompletedEventAttributes( + result=temporalio.api.common.v1.Payloads(payloads=[result]) + ) + ) + ] + ) + ) + service_client = mock.MagicMock() + service_client.workflow_service.get_workflow_execution_history = mock.AsyncMock( + side_effect=[first_page, close_page] + ) + client = Client( + service_client, + data_converter=dataclasses.replace( + DataConverter.default, + payload_codec=NexusContextMarkerPayloadCodec({context: b"nexus"}), + ), + ) + + assert ( + await client.get_workflow_handle("workflow-id", result_type=str).result() + == "nexus-value" + ) + + @pytest.mark.requires_local_server @pytest.mark.parametrize("outcome", ["result", "failure", "continue_as_new"]) async def test_nexus_started_workflow_uses_nexus_context( From 191557c5a99a186a443489c3d7553d780d99c5c6 Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Thu, 29 Oct 2026 14:28:24 -0500 Subject: [PATCH 4/5] Use a custom search attribute in the standalone activity search attribute test --- tests/test_activity.py | 15 ++++++++++----- 1 file changed, 10 insertions(+), 5 deletions(-) diff --git a/tests/test_activity.py b/tests/test_activity.py index 23d322e33c..2a6ea8c243 100644 --- a/tests/test_activity.py +++ b/tests/test_activity.py @@ -31,7 +31,11 @@ from temporalio.service import RPCError, RPCStatusCode from temporalio.testing import WorkflowEnvironment from temporalio.worker import Worker -from tests.helpers import assert_eq_eventually, assert_eventually +from tests.helpers import ( + assert_eq_eventually, + assert_eventually, + ensure_search_attributes_present, +) @activity.defn @@ -1158,9 +1162,10 @@ async def test_search_attributes(client: Client, env: WorkflowEnvironment): activity_id = str(uuid.uuid4()) task_queue = str(uuid.uuid4()) - temporal_change_version_key = SearchAttributeKey.for_keyword_list( - "TemporalChangeVersion" + keyword_list_key = SearchAttributeKey.for_keyword_list( + "python-test-activity-keyword-list" ) + await ensure_search_attributes_present(client, keyword_list_key) handle = await client.start_activity( increment, @@ -1169,12 +1174,12 @@ async def test_search_attributes(client: Client, env: WorkflowEnvironment): task_queue=task_queue, schedule_to_close_timeout=timedelta(seconds=60), search_attributes=TypedSearchAttributes( - [SearchAttributePair(temporal_change_version_key, ["test-1", "test-2"])] + [SearchAttributePair(keyword_list_key, ["test-1", "test-2"])] ), ) desc = await handle.describe() - assert desc.typed_search_attributes[temporal_change_version_key] == [ + assert desc.typed_search_attributes[keyword_list_key] == [ "test-1", "test-2", ] From a72fa128b2a72f8fe0218d25a6fd7fc285f9fdb6 Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Thu, 29 Oct 2026 15:25:02 -0500 Subject: [PATCH 5/5] Run the standalone activity search attribute test only on a local server --- tests/test_activity.py | 2 ++ 1 file changed, 2 insertions(+) diff --git a/tests/test_activity.py b/tests/test_activity.py index 2a6ea8c243..3f10578958 100644 --- a/tests/test_activity.py +++ b/tests/test_activity.py @@ -1148,6 +1148,8 @@ async def test_id_reuse_policy_allow_duplicate( assert handle1.run_id != handle2.run_id +# Registering a custom search attribute needs the operator service, which cloud doesn't allow +@pytest.mark.requires_local_server async def test_search_attributes(client: Client, env: WorkflowEnvironment): if env.supports_time_skipping_v1: pytest.skip(