NexusSerializationContext for data/failure converters

CHANGELOG.md

@@ -27,6 +27,10 @@ to include examples, links to docs, or any other relevant information.
 - Added GCP Cloud Run serverless-worker OpenTelemetry plugin in `temporalio.contrib.opentelemetry`.
 - Added new options to ActivityHandle.describe() to retrieve associated payloads, such as activity input and outcome.
 - New properties and methods in ActivityExecution and ActivityExecutionDescription.
+- Added experimental `temporalio.converter.NexusSerializationContext` support for Nexus callers
+  and handlers. Callers use it for inputs, results, and failures; handlers use it for inputs,
+  synchronous results, and failures. Asynchronous handler results and detached standalone handles
+  are not yet supported. Standalone `USE_EXISTING` handles use their start request's context.
 
 ### Changed
 
@@ -36,6 +40,8 @@ to include examples, links to docs, or any other relevant information.
 - System Nexus Signal-with-Start Workflow operations now invoke
   `WorkflowOutboundInterceptor.start_system_nexus_operation` after their typed interception
   point. They continue not to invoke `WorkflowOutboundInterceptor.start_nexus_operation`.
+- The experimental `GetNexusOperationResultInput` now includes the Nexus endpoint, service, and
+  operation.
 
 ### Deprecated
 

temporalio/client/_impl.py

@@ -1549,6 +1549,12 @@ async def start_nexus_operation(
         self, input: StartNexusOperationInput
     ) -> NexusOperationHandle[Any]:
         """Start a nexus operation and return a handle to it."""
+        nexus_context = temporalio.converter.NexusSerializationContext(
+            endpoint=input.endpoint,
+            service=input.service,
+            operation=input.operation,
+        )
+        data_converter = self._client.data_converter.with_context(nexus_context)
         req = temporalio.api.workflowservice.v1.StartNexusOperationExecutionRequest(
             namespace=self._client.namespace,
             identity=self._client.identity,
@@ -1575,7 +1581,7 @@ async def start_nexus_operation(
             req.start_to_close_timeout.FromTimedelta(input.start_to_close_timeout)
 
         # Set input payload
-        encoded = await self._client.data_converter.encode([input.arg])
+        encoded = await data_converter.encode([input.arg])
         if encoded:
             req.input.CopyFrom(encoded[0])
 
@@ -1620,6 +1626,7 @@ async def start_nexus_operation(
             result_type=input.result_type,
             endpoint=input.endpoint,
             service=input.service,
+            operation=input.operation,
         )
 
     async def describe_nexus_operation(
@@ -1637,15 +1644,31 @@ async def describe_nexus_operation(
             metadata=input.rpc_metadata,
             timeout=input.rpc_timeout,
         )
+        data_converter = self._client.data_converter.with_context(
+            temporalio.converter.NexusSerializationContext(
+                endpoint=resp.info.endpoint,
+                service=resp.info.service,
+                operation=resp.info.operation,
+            )
+        )
         return await NexusOperationExecutionDescription._from_execution_info(
             info=resp.info,
-            data_converter=self._client.data_converter,
+            data_converter=data_converter,
         )
 
     async def get_nexus_operation_result(
         self, input: GetNexusOperationResultInput
     ) -> Any:
         """Poll for nexus operation result until it's available."""
+        data_converter = self._client.data_converter
+        if input.endpoint and input.service and input.operation:
+            data_converter = data_converter.with_context(
+                temporalio.converter.NexusSerializationContext(
+                    endpoint=input.endpoint,
+                    service=input.service,
+                    operation=input.operation,
+                )
+            )
         req = temporalio.api.workflowservice.v1.PollNexusOperationExecutionRequest(
             namespace=self._client.namespace,
             operation_id=input.operation_id,
@@ -1667,21 +1690,14 @@ async def get_nexus_operation_result(
                 match res.WhichOneof("outcome"):
                     case "result":
                         type_hints = [input.result_type] if input.result_type else None
-                        [result] = await self._client.data_converter.decode(
-                            [res.result], type_hints
-                        )
+                        [result] = await data_converter.decode([res.result], type_hints)
                         return result
-
                     case "failure":
                         raise NexusOperationFailureError(
-                            cause=await self._client.data_converter.decode_failure(
-                                res.failure
-                            )
+                            cause=await data_converter.decode_failure(res.failure)
                         )
-
                     case None:
-                        # poll again
-                        pass
+                        continue
             except RPCError as err:
                 match err.status:
                     case RPCStatusCode.DEADLINE_EXCEEDED:

temporalio/client/_interceptor.py

@@ -18,9 +18,7 @@
 import temporalio.api.common.v1
 import temporalio.api.workflowservice.v1
 import temporalio.common
-from temporalio.converter import (
-    DataConverter,
-)
+from temporalio.converter import DataConverter
 
 if TYPE_CHECKING:
     from ._activity import (
@@ -657,6 +655,9 @@ class GetNexusOperationResultInput:
 
     operation_id: str
     run_id: str | None
+    endpoint: str
+    service: str
+    operation: str
     rpc_metadata: Mapping[str, str | bytes]
     rpc_timeout: timedelta | None
     result_type: type[Any] | None

temporalio/client/_nexus.py

@@ -1066,6 +1066,7 @@ def __init__(
         result_type: type | None = None,
         endpoint: str = "",
         service: str = "",
+        operation: str = "",
     ) -> None:
         """Create nexus operation handle."""
         self._client = client
@@ -1074,6 +1075,7 @@ def __init__(
         self._result_type = result_type
         self._endpoint = endpoint
         self._service = service
+        self._operation = operation
         # the default value is `_arg_unset` because ReturnType could be None
         self._known_outcome: ReturnType | NexusOperationFailureError | object = (
             temporalio.common._arg_unset
@@ -1136,9 +1138,12 @@ async def result(
                         GetNexusOperationResultInput(
                             operation_id=self._operation_id,
                             run_id=self._run_id,
-                            result_type=self._result_type,
+                            endpoint=self._endpoint,
+                            service=self._service,
+                            operation=self._operation,
                             rpc_metadata=rpc_metadata,
                             rpc_timeout=rpc_timeout,
+                            result_type=self._result_type,
                         )
                     )
                 )

temporalio/converter/__init__.py

@@ -50,6 +50,7 @@
 )
 from temporalio.converter._serialization_context import (
     ActivitySerializationContext,
+    NexusSerializationContext,
     SerializationContext,
     WithSerializationContext,
     WorkflowSerializationContext,
@@ -82,6 +83,7 @@
     "JSONProtoPayloadConverter",
     "JSONTypeConverter",
     "JSONTypeConverterUnhandled",
+    "NexusSerializationContext",
     "PayloadCodec",
     "PayloadConverter",
     "SerializationContext",

temporalio/converter/_serialization_context.py

@@ -28,6 +28,10 @@ class SerializationContext(ABC):
         context type is :py:class:`ActivitySerializationContext` and the workflow ID is that of the
         currently-executing workflow. ActivitySerializationContext is also set on data converter
         operations in the activity context.
+
+        When operating on a Nexus operation payload, the context type is
+        :py:class:`NexusSerializationContext` and identifies the Nexus endpoint, service, and
+        resolved operation name.
     """
 
     pass
@@ -94,6 +98,38 @@ class ActivitySerializationContext(SerializationContext):
     """Whether the activity is a local activity started from a workflow."""
 
 
+@dataclass(frozen=True)
+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.
+
+    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.
+
+    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
+    accept payloads encoded without context.
+
+    .. warning::
+        This API is experimental and unstable.
+    """
+
+    endpoint: str
+    """Nexus endpoint name."""
+
+    service: str
+    """Nexus service name."""
+
+    operation: str
+    """Nexus operation name."""
+
+
 class WithSerializationContext(ABC):
     """Interface for classes that can use serialization context.
 

temporalio/worker/_command_aware_visitor.py

@@ -24,6 +24,7 @@
     ScheduleNexusOperation,
     SignalExternalWorkflowExecution,
     StartChildWorkflowExecution,
+    WorkflowCommand,
 )
 
 
@@ -115,6 +116,18 @@ async def _visit_coresdk_workflow_commands_ScheduleNexusOperation(
         with current_command(CommandType.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION, o.seq):
             await super()._visit_coresdk_workflow_commands_ScheduleNexusOperation(fs, o)
 
+    async def _visit_coresdk_workflow_commands_WorkflowCommand(
+        self, fs: VisitorFunctions, o: WorkflowCommand
+    ) -> None:
+        if o.HasField("schedule_nexus_operation"):
+            with current_command(
+                CommandType.COMMAND_TYPE_SCHEDULE_NEXUS_OPERATION,
+                o.schedule_nexus_operation.seq,
+            ):
+                await super()._visit_coresdk_workflow_commands_WorkflowCommand(fs, o)
+        else:
+            await super()._visit_coresdk_workflow_commands_WorkflowCommand(fs, o)
+
     # Workflow activation jobs with payloads
     async def _visit_coresdk_workflow_activation_ResolveActivity(
         self, fs: VisitorFunctions, o: ResolveActivity