Nexus: worker, workflow-backed operations, and workflow caller

README.md

@@ -94,6 +94,7 @@ informal introduction to the features and their implementation.
         - [Heartbeating and Cancellation](#heartbeating-and-cancellation)
         - [Worker Shutdown](#worker-shutdown)
       - [Testing](#testing-1)
+    - [Nexus](#nexus)
     - [Workflow Replay](#workflow-replay)
     - [Observability](#observability)
       - [Metrics](#metrics)
@@ -1308,6 +1309,7 @@ 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
 
+
 ### 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,6 +11,7 @@ keywords = [
     "workflow",
 ]
 dependencies = [
+    "nexus-rpc",
     "protobuf>=3.20,<6",
     "python-dateutil>=2.8.2,<3 ; python_version < '3.11'",
     "types-protobuf>=3.20",
@@ -44,7 +45,7 @@ dev = [
     "psutil>=5.9.3,<6",
     "pydocstyle>=6.3.0,<7",
     "pydoctor>=24.11.1,<25",
-    "pyright==1.1.377",
+    "pyright==1.1.400",
     "pytest~=7.4",
     "pytest-asyncio>=0.21,<0.22",
     "pytest-timeout~=2.2",
@@ -53,6 +54,8 @@ dev = [
     "twine>=4.0.1,<5",
     "ruff>=0.5.0,<0.6",
     "maturin>=1.8.2",
+    "pytest-cov>=6.1.1",
+    "httpx>=0.28.1",
     "pytest-pretty>=1.3.0",
 ]
 
@@ -162,6 +165,7 @@ exclude = [
   "tests/worker/workflow_sandbox/testmodules/proto",
   "temporalio/bridge/worker.py",
   "temporalio/contrib/opentelemetry.py",
+  "temporalio/contrib/pydantic.py",
   "temporalio/converter.py",
   "temporalio/testing/_workflow.py",
   "temporalio/worker/_activity.py",
@@ -173,6 +177,10 @@ exclude = [
   "tests/api/test_grpc_stub.py",
   "tests/conftest.py",
   "tests/contrib/test_opentelemetry.py",
+  "tests/contrib/pydantic/models.py",
+  "tests/contrib/pydantic/models_2.py",
+  "tests/contrib/pydantic/test_pydantic.py",
+  "tests/contrib/pydantic/workflows.py",
   "tests/test_converter.py",
   "tests/test_service.py",
   "tests/test_workflow.py",
@@ -208,3 +216,6 @@ 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/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;
@@ -60,6 +60,7 @@ pub struct WorkerConfig {
     graceful_shutdown_period_millis: u64,
     nondeterminism_as_workflow_fail: bool,
     nondeterminism_as_workflow_fail_for_types: HashSet<String>,
+    nexus_task_poller_behavior: PollerBehavior,
 }
 
 #[derive(FromPyObject)]
@@ -565,6 +566,18 @@ impl WorkerRef {
         })
     }
 
+    fn poll_nexus_task<'p>(&self, py: Python<'p>) -> PyResult<Bound<'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))),
+            };
+            Ok(bytes)
+        })
+    }
+
     fn complete_workflow_activation<'p>(
         &self,
         py: Python<'p>,
@@ -599,6 +612,22 @@ impl WorkerRef {
         })
     }
 
+    fn complete_nexus_task<'p>(&self,
+        py: Python<'p>,
+        proto: &Bound<'_, PyBytes>,
+) -> PyResult<Bound<'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: &Bound<'_, PyBytes>) -> PyResult<()> {
         enter_sync!(self.runtime);
         let heartbeat = ActivityHeartbeat::decode(proto.as_bytes())
@@ -696,6 +725,7 @@ fn convert_worker_config(
                 })
                 .collect::<HashMap<String, HashSet<WorkflowErrorType>>>(),
         )
+        .nexus_task_poller_behavior(conf.nexus_task_poller_behavior)
         .build()
         .map_err(|err| PyValueError::new_err(format!("Invalid worker config: {err}")))
 }

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
@@ -60,6 +61,7 @@ class WorkerConfig:
     graceful_shutdown_period_millis: int
     nondeterminism_as_workflow_fail: bool
     nondeterminism_as_workflow_fail_for_types: Set[str]
+    nexus_task_poller_behavior: PollerBehavior
 
 
 @dataclass
@@ -216,6 +218,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,
@@ -229,6 +239,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

@@ -464,9 +464,17 @@ async def start_workflow(
         rpc_metadata: Mapping[str, str] = {},
         rpc_timeout: Optional[timedelta] = None,
         request_eager_start: bool = False,
-        stack_level: int = 2,
         priority: temporalio.common.Priority = temporalio.common.Priority.default,
         versioning_override: Optional[temporalio.common.VersioningOverride] = None,
+        # The following options should not be considered part of the public API. They
+        # are deliberately not exposed in overloads, and are not subject to any
+        # backwards compatibility guarantees.
+        nexus_completion_callbacks: Sequence[NexusCompletionCallback] = [],
+        workflow_event_links: Sequence[
+            temporalio.api.common.v1.Link.WorkflowEvent
+        ] = [],
+        request_id: Optional[str] = None,
+        stack_level: int = 2,
     ) -> WorkflowHandle[Any, Any]:
         """Start a workflow and return its handle.
 
@@ -529,7 +537,6 @@ async def start_workflow(
         name, result_type_from_type_hint = (
             temporalio.workflow._Definition.get_name_and_result_type(workflow)
         )
-
         return await self._impl.start_workflow(
             StartWorkflowInput(
                 workflow=name,
@@ -557,6 +564,9 @@ async def start_workflow(
                 rpc_timeout=rpc_timeout,
                 request_eager_start=request_eager_start,
                 priority=priority,
+                nexus_completion_callbacks=nexus_completion_callbacks,
+                workflow_event_links=workflow_event_links,
+                request_id=request_id,
             )
         )
 
@@ -5193,6 +5203,10 @@ class StartWorkflowInput:
     rpc_timeout: Optional[timedelta]
     request_eager_start: bool
     priority: temporalio.common.Priority
+    # The following options are experimental and unstable.
+    nexus_completion_callbacks: Sequence[NexusCompletionCallback]
+    workflow_event_links: Sequence[temporalio.api.common.v1.Link.WorkflowEvent]
+    request_id: Optional[str]
     versioning_override: Optional[temporalio.common.VersioningOverride] = None
 
 
@@ -5807,8 +5821,26 @@ async def _build_start_workflow_execution_request(
         self, input: StartWorkflowInput
     ) -> temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest:
         req = temporalio.api.workflowservice.v1.StartWorkflowExecutionRequest()
-        req.request_eager_execution = input.request_eager_start
         await self._populate_start_workflow_execution_request(req, input)
+        # _populate_start_workflow_execution_request is used for both StartWorkflowInput
+        # and UpdateWithStartStartWorkflowInput. UpdateWithStartStartWorkflowInput does
+        # not have the following two fields so they are handled here.
+        req.request_eager_execution = input.request_eager_start
+        if input.request_id:
+            req.request_id = input.request_id
+
+        req.completion_callbacks.extend(
+            temporalio.api.common.v1.Callback(
+                nexus=temporalio.api.common.v1.Callback.Nexus(
+                    url=callback.url, header=callback.header
+                )
+            )
+            for callback in input.nexus_completion_callbacks
+        )
+        req.links.extend(
+            temporalio.api.common.v1.Link(workflow_event=link)
+            for link in input.workflow_event_links
+        )
         return req
 
     async def _build_signal_with_start_workflow_execution_request(
@@ -7231,6 +7263,21 @@ def api_key(self, value: Optional[str]) -> None:
         self.service_client.update_api_key(value)
 
 
+@dataclass(frozen=True)
+class NexusCompletionCallback:
+    """Nexus callback to attach to events such as workflow completion.
+
+    .. warning::
+        This option is experimental and unstable.
+    """
+
+    url: str
+    """Callback URL."""
+
+    header: Mapping[str, str]
+    """Header to attach to callback request."""
+
+
 async def _encode_user_metadata(
     converter: temporalio.converter.DataConverter,
     summary: Optional[Union[str, temporalio.api.common.v1.Payload]],

temporalio/common.py

@@ -8,7 +8,7 @@
 from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from datetime import datetime, timedelta
-from enum import Enum, IntEnum
+from enum import IntEnum
 from typing import (
     Any,
     Callable,

temporalio/converter.py

@@ -16,6 +16,7 @@
 from datetime import datetime
 from enum import IntEnum
 from itertools import zip_longest
+from logging import getLogger
 from typing import (
     Any,
     Awaitable,
@@ -40,6 +41,7 @@
 import google.protobuf.json_format
 import google.protobuf.message
 import google.protobuf.symbol_database
+import nexusrpc
 import typing_extensions
 
 import temporalio.api.common.v1
@@ -60,6 +62,8 @@
 if sys.version_info >= (3, 10):
     from types import UnionType
 
+logger = getLogger(__name__)
+
 
 class PayloadConverter(ABC):
     """Base payload converter to/from multiple payloads/values."""
@@ -911,6 +915,12 @@ def _error_to_failure(
             failure.child_workflow_execution_failure_info.retry_state = (
                 temporalio.api.enums.v1.RetryState.ValueType(error.retry_state or 0)
             )
+        # TODO(nexus-prerelease): test coverage for this
+        elif isinstance(error, temporalio.exceptions.NexusOperationError):
+            failure.nexus_operation_execution_failure_info.SetInParent()
+            failure.nexus_operation_execution_failure_info.operation_token = (
+                error.operation_token
+            )
 
     def from_failure(
         self,
@@ -1006,6 +1016,33 @@ def from_failure(
                 if child_info.retry_state
                 else None,
             )
+        elif failure.HasField("nexus_handler_failure_info"):
+            nexus_handler_failure_info = failure.nexus_handler_failure_info
+            try:
+                _type = nexusrpc.HandlerErrorType[nexus_handler_failure_info.type]
+            except KeyError:
+                logger.warning(
+                    f"Unknown Nexus HandlerErrorType: {nexus_handler_failure_info.type}"
+                )
+                _type = nexusrpc.HandlerErrorType.INTERNAL
+            return nexusrpc.HandlerError(
+                failure.message or "Nexus handler error",
+                type=_type,
+                retryable={
+                    temporalio.api.enums.v1.NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_RETRYABLE: True,
+                    temporalio.api.enums.v1.NexusHandlerErrorRetryBehavior.NEXUS_HANDLER_ERROR_RETRY_BEHAVIOR_NON_RETRYABLE: False,
+                }.get(nexus_handler_failure_info.retry_behavior),
+            )
+        elif failure.HasField("nexus_operation_execution_failure_info"):
+            nexus_op_failure_info = failure.nexus_operation_execution_failure_info
+            err = temporalio.exceptions.NexusOperationError(
+                failure.message or "Nexus operation error",
+                scheduled_event_id=nexus_op_failure_info.scheduled_event_id,
+                endpoint=nexus_op_failure_info.endpoint,
+                service=nexus_op_failure_info.service,
+                operation=nexus_op_failure_info.operation,
+                operation_token=nexus_op_failure_info.operation_token,
+            )
         else:
             err = temporalio.exceptions.FailureError(failure.message or "Failure error")
         err._failure = failure