Nexus prototype

Author: dandavison

README.md

@@ -96,6 +96,8 @@ informal introduction to the features and their implementation.
         - [Heartbeating and Cancellation](#heartbeating-and-cancellation)
         - [Worker Shutdown](#worker-shutdown)
       - [Testing](#testing-1)
+    - [Nexus](#nexus)
+      - [hello](#hello)
     - [Workflow Replay](#workflow-replay)
     - [Observability](#observability)
       - [Metrics](#metrics)
@@ -1314,6 +1316,71 @@ affect calls activity code might make to functions on the `temporalio.activity`
 * `cancel()` can be invoked to simulate a cancellation of the activity
 * `worker_shutdown()` can be invoked to simulate a worker shutdown during execution of the activity
 
+
+### Nexus
+
+See [docs.temporal.io/nexus](https://docs.temporal.io/nexus).
+
+#### Service Interface Definition
+
+A Nexus Service interface definition is a set of named operations, where each operation is an
+`(input_type,output_type)` pair:
+
+```python
+@nexus_service
+class MyNexusService:
+    my_operation: NexusOperation[MyOpInput, MyOpOutput]
+```
+
+### Operation implementation
+
+```python
+@nexus.service(interface.MyNexusService)
+class MyNexusService:
+
+    @nexus.sync_operation
+    def echo(self, input: EchoInput) -> EchoOutput:
+        return EchoOutput(message=input.message)
+```
+
+```python
+@nexus.service(interface.MyNexusService)
+class MyNexusService:
+
+    @temporalio.nexus.workflow_operation
+    async def hello(
+        self, input: HelloInput
+    ) -> AsyncWorkflowOperationResult[HelloOutput]:
+        return await temporalio.nexus.handler.start_workflow(HelloWorkflow.run, input)
+```
+
+
+### Request options
+
+```python
+@dataclass
+class OperationOptions:
+    """Options passed by the Nexus caller when starting an operation."""
+
+    # A callback URL is required to deliver the completion of an async operation. This URL should be
+    # called by a handler upon completion if the started operation is async.
+    callback_url: Optional[str] = None
+
+    # Optional header fields set by the caller to be attached to the callback request when an
+    # asynchronous operation completes.
+    callback_header: dict[str, str] = field(default_factory=dict)
+
+    # Request ID that may be used by the server handler to dedupe a start request.
+    # By default a v4 UUID will be generated by the client.
+    request_id: Optional[str] = None
+
+    # Links contain arbitrary caller information. Handlers may use these links as
+    # metadata on resources associated with an operation.
+    links: list[Link] = field(default_factory=list)
+```
+
+
+
 ### Workflow Replay
 
 Given a workflow's history, it can be replayed locally to check for things like non-determinism errors. For example,

pyproject.toml

@@ -11,8 +11,11 @@ keywords = [
     "workflow",
 ]
 dependencies = [
+    "nexus-rpc",
     "protobuf>=3.20",
+    "pyright==1.1.394",
     "python-dateutil>=2.8.2,<3 ; python_version < '3.11'",
+    "temporalio-xray",
     "types-protobuf>=3.20",
     "typing-extensions>=4.2.0,<5",
 ]
@@ -40,7 +43,7 @@ dev = [
     "psutil>=5.9.3,<6",
     "pydocstyle>=6.3.0,<7",
     "pydoctor>=24.11.1,<25",
-    "pyright==1.1.377",
+    "pyright==1.1.394",
     "pytest~=7.4",
     "pytest-asyncio>=0.21,<0.22",
     "pytest-timeout~=2.2",
@@ -185,6 +188,7 @@ exclude = [
 
 [tool.ruff]
 target-version = "py39"
+extend-ignore = ["E741"]  # Allow single-letter variable names like I, O
 
 [build-system]
 requires = ["maturin>=1.0,<2.0"]
@@ -202,3 +206,7 @@ exclude = [
 [tool.uv]
 # Prevent uv commands from building the package by default
 package = false
+
+[tool.uv.sources]
+nexus-rpc = { path = "../nexus-sdk-python", editable = true }
+temporalio-xray = { path = "../xray/sdks/python", editable = true }

temporalio/bridge/src/worker.rs

@@ -20,7 +20,7 @@ use temporal_sdk_core_api::worker::{
 };
 use temporal_sdk_core_api::Worker;
 use temporal_sdk_core_protos::coresdk::workflow_completion::WorkflowActivationCompletion;
-use temporal_sdk_core_protos::coresdk::{ActivityHeartbeat, ActivityTaskCompletion};
+use temporal_sdk_core_protos::coresdk::{ActivityHeartbeat, ActivityTaskCompletion, nexus::NexusTaskCompletion};
 use temporal_sdk_core_protos::temporal::api::history::v1::History;
 use tokio::sync::mpsc::{channel, Sender};
 use tokio_stream::wrappers::ReceiverStream;
@@ -475,6 +475,19 @@ impl WorkerRef {
         })
     }
 
+    fn poll_nexus_task<'p>(&self, py: Python<'p>) -> PyResult<&'p PyAny> {
+        let worker = self.worker.as_ref().unwrap().clone();
+        self.runtime.future_into_py(py, async move {
+            let bytes = match worker.poll_nexus_task().await {
+                Ok(task) => task.encode_to_vec(),
+                Err(PollError::ShutDown) => return Err(PollShutdownError::new_err(())),
+                Err(err) => return Err(PyRuntimeError::new_err(format!("Poll failure: {}", err))),
+            };
+            let bytes: &[u8] = &bytes;
+            Ok(Python::with_gil(|py| bytes.into_py(py)))
+        })
+    }
+
     fn complete_workflow_activation<'p>(
         &self,
         py: Python<'p>,
@@ -505,6 +518,19 @@ impl WorkerRef {
         })
     }
 
+    fn complete_nexus_task<'p>(&self, py: Python<'p>, proto: &PyBytes) -> PyResult<&'p PyAny> {
+        let worker = self.worker.as_ref().unwrap().clone();
+        let completion = NexusTaskCompletion::decode(proto.as_bytes())
+            .map_err(|err| PyValueError::new_err(format!("Invalid proto: {}", err)))?;
+        self.runtime.future_into_py(py, async move {
+            worker
+                .complete_nexus_task(completion)
+                .await
+                .context("Completion failure")
+                .map_err(Into::into)
+        })
+    }
+
     fn record_activity_heartbeat(&self, proto: &PyBytes) -> PyResult<()> {
         enter_sync!(self.runtime);
         let heartbeat = ActivityHeartbeat::decode(proto.as_bytes())

temporalio/bridge/worker.py

@@ -26,6 +26,7 @@
 import temporalio.bridge.client
 import temporalio.bridge.proto
 import temporalio.bridge.proto.activity_task
+import temporalio.bridge.proto.nexus
 import temporalio.bridge.proto.workflow_activation
 import temporalio.bridge.proto.workflow_completion
 import temporalio.bridge.runtime
@@ -35,7 +36,7 @@
 from temporalio.bridge.temporal_sdk_bridge import (
     CustomSlotSupplier as BridgeCustomSlotSupplier,
 )
-from temporalio.bridge.temporal_sdk_bridge import PollShutdownError
+from temporalio.bridge.temporal_sdk_bridge import PollShutdownError  # type: ignore
 
 
 @dataclass
@@ -156,6 +157,14 @@ async def poll_activity_task(
             await self._ref.poll_activity_task()
         )
 
+    async def poll_nexus_task(
+        self,
+    ) -> temporalio.bridge.proto.nexus.NexusTask:
+        """Poll for a nexus task."""
+        return temporalio.bridge.proto.nexus.NexusTask.FromString(
+            await self._ref.poll_nexus_task()
+        )
+
     async def complete_workflow_activation(
         self,
         comp: temporalio.bridge.proto.workflow_completion.WorkflowActivationCompletion,
@@ -169,6 +178,12 @@ async def complete_activity_task(
         """Complete an activity task."""
         await self._ref.complete_activity_task(comp.SerializeToString())
 
+    async def complete_nexus_task(
+        self, comp: temporalio.bridge.proto.nexus.NexusTaskCompletion
+    ) -> None:
+        """Complete a nexus task."""
+        await self._ref.complete_nexus_task(comp.SerializeToString())
+
     def record_activity_heartbeat(
         self, comp: temporalio.bridge.proto.ActivityHeartbeat
     ) -> None:

temporalio/client.py

@@ -316,6 +316,7 @@ async def start_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -350,6 +351,7 @@ async def start_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -386,6 +388,7 @@ async def start_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -422,6 +425,7 @@ async def start_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -456,6 +460,7 @@ async def start_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -500,6 +505,8 @@ async def start_workflow(
                 instead of traditional workflow start.
             start_signal_args: Arguments for start_signal if start_signal
                 present.
+            completion_callbacks: Callbacks to be called by the server when the workflow reaches a
+                terminal state.
             rpc_metadata: Headers used on the RPC call. Keys here override
                 client-level RPC metadata keys.
             rpc_timeout: Optional RPC deadline to set for the RPC call.
@@ -544,6 +551,7 @@ async def start_workflow(
                 static_details=static_details,
                 start_signal=start_signal,
                 start_signal_args=start_signal_args,
+                completion_callbacks=completion_callbacks,
                 ret_type=result_type or result_type_from_type_hint,
                 rpc_metadata=rpc_metadata,
                 rpc_timeout=rpc_timeout,
@@ -579,6 +587,7 @@ async def execute_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -613,6 +622,7 @@ async def execute_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -649,6 +659,7 @@ async def execute_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -685,6 +696,7 @@ async def execute_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -719,6 +731,7 @@ async def execute_workflow(
         start_delay: Optional[timedelta] = None,
         start_signal: Optional[str] = None,
         start_signal_args: Sequence[Any] = [],
+        completion_callbacks: Sequence[temporalio.common.CompletionCallback] = [],
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
@@ -753,6 +766,7 @@ async def execute_workflow(
                 start_delay=start_delay,
                 start_signal=start_signal,
                 start_signal_args=start_signal_args,
+                completion_callbacks=completion_callbacks,
                 rpc_metadata=rpc_metadata,
                 rpc_timeout=rpc_timeout,
                 request_eager_start=request_eager_start,
@@ -5148,6 +5162,7 @@ class StartWorkflowInput:
     headers: Mapping[str, temporalio.api.common.v1.Payload]
     start_signal: Optional[str]
     start_signal_args: Sequence[Any]
+    completion_callbacks: Sequence[temporalio.common.CompletionCallback]
     static_summary: Optional[str]
     static_details: Optional[str]
     # Type may be absent
@@ -5770,6 +5785,11 @@ async def _build_start_workflow_execution_request(
         req = temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest()
         req.request_eager_execution = input.request_eager_start
         await self._populate_start_workflow_execution_request(req, input)
+        for callback in input.completion_callbacks:
+            c = temporalio.api.common.v1.Callback()
+            c.nexus.url = callback.url
+            c.nexus.header.update(callback.header)
+            req.completion_callbacks.append(c)
         return req
 
     async def _build_signal_with_start_workflow_execution_request(

temporalio/common.py

@@ -195,6 +195,17 @@ def __setstate__(self, state: object) -> None:
         )
 
 
+@dataclass(frozen=True)
+class CompletionCallback:
+    """Callback to attach to various events in the system, e.g. workflow run completion."""
+
+    url: str
+    """Callback URL."""
+
+    header: Mapping[str, str]
+    """Header to attach to callback request."""
+
+
 # We choose to make this a list instead of an sequence so we can catch if people
 # are not sending lists each time but maybe accidentally sending a string (which
 # is a sequence)