Add nexus handler callbacks for SANO

changelog/added/dancing-firefly.md

@@ -0,0 +1 @@
+Add experimental Nexus Handler Callbacks for standalone Nexus operations through `NexusClient.start_operation`. Use `temporalio.nexus.create_completion_callback` to send the result or failure and source context to a Nexus handler.

pyproject.toml

@@ -9,7 +9,7 @@ license = "MIT"
 license-files = ["LICENSE"]
 keywords = ["temporal", "workflow"]
 dependencies = [
-  "nexus-rpc==1.4.0",
+  "nexus-rpc==1.5.0",
   "protobuf>=3.20,<8.0.0",
   "python-dateutil>=2.8.2,<3 ; python_version < '3.11'",
   "types-protobuf>=3.20,<8.0.0",
@@ -280,5 +280,5 @@ exclude = ["temporalio/bridge/target/**/*", "temporalio/bridge/sdk-core/.git"]
 # Prevent uv commands from building the package by default
 package = false
 exclude-newer = "2 weeks"
-# The SDK compatibility extras need the newly published standalone integrations.
-exclude-newer-package = { temporalio-deepagents = false, temporalio-google-adk = false, temporalio-google-genai = false, temporalio-langgraph = false, temporalio-langsmith = false, temporalio-strands-agents = false }
+# Nexus and the SDK compatibility extras need newly published releases.
+exclude-newer-package = { nexus-rpc = false, temporalio-deepagents = false, temporalio-google-adk = false, temporalio-google-genai = false, temporalio-langgraph = false, temporalio-langsmith = false, temporalio-strands-agents = false }

scripts/gen_payload_visitor.py

@@ -12,6 +12,7 @@
 sys.path.insert(0, str(base_dir))
 
 from temporalio.api.common.v1.message_pb2 import Payload, Payloads, SearchAttributes
+from temporalio.api.notificationservice.v1 import OnCompleteRequest, OnCompleteResponse
 from temporalio.bridge.proto.nexus import NexusTaskCompletion
 from temporalio.bridge.proto.workflow_activation.workflow_activation_pb2 import (
     WorkflowActivation,
@@ -440,11 +441,14 @@ def write_bridge_visitors() -> None:
     out_path = base_dir / "temporalio" / "bridge" / "_visitor.py"
 
     # Build root descriptors: WorkflowActivation, WorkflowActivationCompletion,
-    # NexusTaskCompletion, and the system Nexus operation roots.
+    # NexusTaskCompletion, the system Nexus operation roots, and the notification
+    # service messages delivered to Nexus handler callbacks.
     roots: list[Descriptor] = [
         WorkflowActivation.DESCRIPTOR,
         WorkflowActivationCompletion.DESCRIPTOR,
         NexusTaskCompletion.DESCRIPTOR,
+        OnCompleteRequest.DESCRIPTOR,
+        OnCompleteResponse.DESCRIPTOR,
     ] + discover_system_nexus_roots()
 
     code = VisitorGenerator().generate(roots)

temporalio/bridge/_visitor.py

@@ -624,6 +624,21 @@ async def _visit_coresdk_nexus_NexusTaskCompletion(
         elif o.HasField("failure"):
             await self._visit_temporal_api_failure_v1_Failure(fs, o.failure)
 
+    async def _visit_temporal_api_notificationservice_v1_OnCompleteRequest(
+        self, fs: VisitorFunctions, o: Any
+    ):
+        if o.HasField("source_context"):
+            await self._visit_temporal_api_common_v1_Payload(fs, o.source_context)
+        if o.HasField("success"):
+            await self._visit_temporal_api_common_v1_Payload(fs, o.success)
+        elif o.HasField("failure"):
+            await self._visit_temporal_api_failure_v1_Failure(fs, o.failure)
+
+    async def _visit_temporal_api_notificationservice_v1_OnCompleteResponse(
+        self, fs: VisitorFunctions, o: Any
+    ) -> None:
+        pass
+
     async def _visit_temporal_api_common_v1_Header(self, fs: VisitorFunctions, o: Any):
         for v in o.fields.values():
             await self._visit_temporal_api_common_v1_Payload(fs, v)

temporalio/client/_impl.py

@@ -1583,6 +1583,11 @@ async def start_nexus_operation(
                 "temporalio.api.enums.v1.NexusOperationIdConflictPolicy.ValueType",
                 int(input.id_conflict_policy),
             ),
+            # Callback source contexts use the started operation's serialization
+            # context, like the rest of the request.
+            completion_callbacks=[
+                await cb._to_proto(data_converter) for cb in input.completion_callbacks
+            ],
         )
 
         if input.schedule_to_close_timeout is not None:

temporalio/client/_interceptor.py

@@ -18,6 +18,7 @@
 import temporalio.api.common.v1
 import temporalio.api.workflowservice.v1
 import temporalio.common
+import temporalio.nexus
 from temporalio.converter import DataConverter
 
 if TYPE_CHECKING:
@@ -607,6 +608,7 @@ class StartNexusOperationInput:
     headers: Mapping[str, str]
     rpc_metadata: Mapping[str, str | bytes]
     rpc_timeout: timedelta | None
+    completion_callbacks: Sequence[temporalio.nexus.CompletionCallback[Any]] = ()
 
 
 @dataclass

temporalio/client/_nexus.py

@@ -490,11 +490,15 @@ async def start_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
-    # Overload for string operation name
+    # Overload for string operation name with result_type. Callbacks must accept
+    # the result type.
     @overload
     @abstractmethod
     async def start_operation(
@@ -505,17 +509,46 @@ async def start_operation(
         id: str,
         id_reuse_policy: temporalio.common.NexusOperationIDReusePolicy = temporalio.common.NexusOperationIDReusePolicy.ALLOW_DUPLICATE,
         id_conflict_policy: temporalio.common.NexusOperationIDConflictPolicy = temporalio.common.NexusOperationIDConflictPolicy.FAIL,
-        result_type: type[OutputT] | None = None,
+        result_type: type[OutputT],
         schedule_to_close_timeout: timedelta | None = None,
         schedule_to_start_timeout: timedelta | None = None,
         start_to_close_timeout: timedelta | None = None,
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
 
+    # Overload for string operation name without result_type. Callbacks of any
+    # output type are accepted.
+    @overload
+    @abstractmethod
+    async def start_operation(
+        self,
+        operation: str,
+        arg: Any,
+        *,
+        id: str,
+        id_reuse_policy: temporalio.common.NexusOperationIDReusePolicy = temporalio.common.NexusOperationIDReusePolicy.ALLOW_DUPLICATE,
+        id_conflict_policy: temporalio.common.NexusOperationIDConflictPolicy = temporalio.common.NexusOperationIDConflictPolicy.FAIL,
+        result_type: None = None,
+        schedule_to_close_timeout: timedelta | None = None,
+        schedule_to_start_timeout: timedelta | None = None,
+        start_to_close_timeout: timedelta | None = None,
+        search_attributes: temporalio.common.TypedSearchAttributes | None = None,
+        summary: str | None = None,
+        headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
+        rpc_metadata: Mapping[str, str | bytes] = {},
+        rpc_timeout: timedelta | None = None,
+    ) -> NexusOperationHandle[Any]: ...
+
     # Overload for workflow_run_operation methods
     @overload
     @abstractmethod
@@ -536,6 +569,9 @@ async def start_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
@@ -560,6 +596,9 @@ async def start_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
@@ -584,6 +623,9 @@ async def start_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
@@ -607,6 +649,9 @@ async def start_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
@@ -636,6 +681,9 @@ async def start_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[OutputT]: ...
@@ -656,6 +704,7 @@ async def start_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[temporalio.nexus.CompletionCallback[Any]] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[Any]:
@@ -686,6 +735,12 @@ async def start_operation(
             search_attributes: Search attributes for the operation.
             summary: Summary for the operation.
             headers: Headers to attach to the Nexus request.
+            completion_callbacks: Callbacks that report the outcome of the
+                operation. Create them with
+                :py:func:`temporalio.nexus.create_completion_callback`. Each
+                callback must accept the output type of the operation. For a
+                string operation name, type checkers check callbacks against
+                ``result_type`` if it is set.
             rpc_metadata: Headers used on the RPC call.
             rpc_timeout: Optional RPC deadline to set for the RPC call.
 
@@ -711,11 +766,15 @@ async def execute_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> OutputT: ...
 
-    # Overload for string operation name
+    # Overload for string operation name with result_type. Callbacks must accept
+    # the result type.
     @overload
     @abstractmethod
     async def execute_operation(
@@ -726,17 +785,46 @@ async def execute_operation(
         id: str,
         id_reuse_policy: temporalio.common.NexusOperationIDReusePolicy = temporalio.common.NexusOperationIDReusePolicy.ALLOW_DUPLICATE,
         id_conflict_policy: temporalio.common.NexusOperationIDConflictPolicy = temporalio.common.NexusOperationIDConflictPolicy.FAIL,
-        result_type: type[OutputT] | None = None,
+        result_type: type[OutputT],
         schedule_to_close_timeout: timedelta | None = None,
         schedule_to_start_timeout: timedelta | None = None,
         start_to_close_timeout: timedelta | None = None,
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> OutputT: ...
 
+    # Overload for string operation name without result_type. Callbacks of any
+    # output type are accepted.
+    @overload
+    @abstractmethod
+    async def execute_operation(
+        self,
+        operation: str,
+        arg: Any,
+        *,
+        id: str,
+        id_reuse_policy: temporalio.common.NexusOperationIDReusePolicy = temporalio.common.NexusOperationIDReusePolicy.ALLOW_DUPLICATE,
+        id_conflict_policy: temporalio.common.NexusOperationIDConflictPolicy = temporalio.common.NexusOperationIDConflictPolicy.FAIL,
+        result_type: None = None,
+        schedule_to_close_timeout: timedelta | None = None,
+        schedule_to_start_timeout: timedelta | None = None,
+        start_to_close_timeout: timedelta | None = None,
+        search_attributes: temporalio.common.TypedSearchAttributes | None = None,
+        summary: str | None = None,
+        headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
+        rpc_metadata: Mapping[str, str | bytes] = {},
+        rpc_timeout: timedelta | None = None,
+    ) -> Any: ...
+
     # Overload for workflow_run_operation methods
     @overload
     @abstractmethod
@@ -757,6 +845,9 @@ async def execute_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> OutputT: ...
@@ -781,6 +872,9 @@ async def execute_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> OutputT: ...
@@ -805,6 +899,9 @@ async def execute_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> OutputT: ...
@@ -829,6 +926,9 @@ async def execute_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> OutputT: ...
@@ -858,6 +958,9 @@ async def execute_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[
+            temporalio.nexus.CompletionCallback[OutputT]
+        ] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> OutputT: ...
@@ -878,6 +981,7 @@ async def execute_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[temporalio.nexus.CompletionCallback[Any]] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> Any:
@@ -910,6 +1014,12 @@ async def execute_operation(
             search_attributes: Search attributes for the operation.
             summary: Summary for the operation.
             headers: Headers to attach to the Nexus request.
+            completion_callbacks: Callbacks that report the outcome of the
+                operation. Create them with
+                :py:func:`temporalio.nexus.create_completion_callback`. Each
+                callback must accept the output type of the operation. For a
+                string operation name, type checkers check callbacks against
+                ``result_type`` if it is set.
             rpc_metadata: Headers used on the RPC call.
             rpc_timeout: Optional RPC deadline to set for the RPC call.
 
@@ -974,6 +1084,7 @@ async def start_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[temporalio.nexus.CompletionCallback[Any]] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> NexusOperationHandle[Any]:
@@ -1005,6 +1116,7 @@ async def start_operation(
                 headers=dict(headers) if headers else {},
                 rpc_metadata=rpc_metadata,
                 rpc_timeout=rpc_timeout,
+                completion_callbacks=completion_callbacks,
             )
         )
 
@@ -1023,6 +1135,7 @@ async def execute_operation(
         search_attributes: temporalio.common.TypedSearchAttributes | None = None,
         summary: str | None = None,
         headers: Mapping[str, str] | None = None,
+        completion_callbacks: Sequence[temporalio.nexus.CompletionCallback[Any]] = (),
         rpc_metadata: Mapping[str, str | bytes] = {},
         rpc_timeout: timedelta | None = None,
     ) -> Any:
@@ -1046,6 +1159,7 @@ async def execute_operation(
             headers=headers,
             rpc_metadata=rpc_metadata,
             rpc_timeout=rpc_timeout,
+            completion_callbacks=completion_callbacks,
         )
         return await handle.result()