Merge branch 'main' into otel-refactor

Author: Sushisource

.github/workflows/build-binaries.yml

@@ -65,7 +65,7 @@ jobs:
           if [ "$RUNNER_OS" = "Windows" ]; then
             bindir=Scripts
           fi
-          ./.venv/$bindir/pip install 'protobuf>=3.20,<6' 'types-protobuf>=3.20,<6' 'typing-extensions>=4.2.0,<5' pytest pytest_asyncio grpcio pydantic opentelemetry-api opentelemetry-sdk python-dateutil openai-agents
+          ./.venv/$bindir/pip install 'protobuf>=3.20,<6' 'types-protobuf>=3.20,<6' 'typing-extensions>=4.2.0,<5' pytest pytest_asyncio grpcio 'nexus-rpc>=1.1.0' pydantic opentelemetry-api opentelemetry-sdk python-dateutil openai-agents
           ./.venv/$bindir/pip install --no-index --find-links=../dist temporalio
           ./.venv/$bindir/python -m pytest -s -k test_workflow_hello
 

.github/workflows/ci.yml

@@ -28,6 +28,7 @@ jobs:
             python: "3.13"
             docsTarget: true
             cloudTestTarget: true
+            openaiTestTarget: true
             clippyLinter: true
           - os: ubuntu-latest
             python: "3.9"
@@ -75,8 +76,6 @@ jobs:
       - run: mkdir junit-xml
       - run: poe test ${{matrix.pytestExtraArgs}} -s --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}.xml
         timeout-minutes: 10
-        env:
-          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
       # Time skipping doesn't yet support ARM
       - if: ${{ !endsWith(matrix.os, '-arm') }}
         run: poe test ${{matrix.pytestExtraArgs}} -s --workflow-environment time-skipping --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}--time-skipping.xml
@@ -89,6 +88,11 @@ jobs:
           TEMPORAL_CLIENT_CLOUD_API_KEY: ${{ secrets.TEMPORAL_CLIENT_CLOUD_API_KEY }}
           TEMPORAL_CLIENT_CLOUD_API_VERSION: 2024-05-13-00
           TEMPORAL_CLIENT_CLOUD_NAMESPACE: sdk-ci.a2dd6
+      - if: ${{ matrix.openaiTestTarget && (github.event.pull_request.head.repo.full_name == '' || github.event.pull_request.head.repo.full_name == 'temporalio/sdk-python') }}
+        run: poe test tests/contrib/openai_agents/test_openai.py ${{matrix.pytestExtraArgs}} -s  --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}--openai.xml
+        timeout-minutes: 10
+        env:
+          OPENAI_API_KEY: ${{ secrets.OPENAI_API_KEY }}
       - name: "Upload junit-xml artifacts"
         uses: actions/upload-artifact@v4
         if: always()

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,113 @@ 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
+
+⚠️  **Nexus support is currently at an experimental release stage. Backwards-incompatible changes are anticipated until a stable release is announced.** ⚠️
+
+[Nexus](https://github.com/nexus-rpc/) is a synchronous RPC protocol. Arbitrary duration operations that can respond
+asynchronously are modeled on top of a set of pre-defined synchronous RPCs.
+
+Temporal supports calling Nexus operations **from a workflow**. See https://docs.temporal.io/nexus. There is no support
+currently for calling a Nexus operation from non-workflow code.
+
+To get started quickly using Nexus with Temporal, see the Python Nexus sample:
+https://github.com/temporalio/samples-python/tree/nexus/hello_nexus.
+
+
+Two types of Nexus operation are supported, each using a decorator:
+
+- `@temporalio.nexus.workflow_run_operation`: a Nexus operation that is backed by a Temporal workflow. The operation
+  handler you write will start the handler workflow and then respond with a token indicating that the handler workflow
+  is in progress. When the handler workflow completes, Temporal server will automatically deliver the result (success or
+  failure) to the caller workflow.
+- `@nexusrpc.handler.sync_operation`: an operation that responds synchronously. It may be `def` or `async def` and it
+may do network I/O, but it must respond within 10 seconds.
+
+The following steps are an overview of the [Python Nexus sample](
+https://github.com/temporalio/samples-python/tree/nexus/hello_nexus).
+
+1. Create the caller and handler namespaces, and the Nexus endpoint. For example,
+    ```
+    temporal operator namespace create --namespace my-handler-namespace
+    temporal operator namespace create --namespace my-caller-namespace
+
+    temporal operator nexus endpoint create \
+      --name my-nexus-endpoint \
+      --target-namespace my-handler-namespace \
+      --target-task-queue my-handler-task-queue
+    ```
+
+2. Define your service contract. This specifies the names and input/output types of your operations. You will use this
+   to refer to the operations when calling them from a workflow.
+    ```python
+    @nexusrpc.service
+    class MyNexusService:
+        my_sync_operation: nexusrpc.Operation[MyInput, MyOutput]
+        my_workflow_run_operation: nexusrpc.Operation[MyInput, MyOutput]
+    ```
+
+3. Implement your operation handlers in a service handler:
+    ```python
+    @service_handler(service=MyNexusService)
+    class MyNexusServiceHandler:
+        @sync_operation
+        async def my_sync_operation(
+            self, ctx: StartOperationContext, input: MyInput
+        ) -> MyOutput:
+            return MyOutput(message=f"Hello {input.name} from sync operation!")
+
+        @workflow_run_operation
+        async def my_workflow_run_operation(
+            self, ctx: WorkflowRunOperationContext, input: MyInput
+        ) -> nexus.WorkflowHandle[MyOutput]:
+            return await ctx.start_workflow(
+                WorkflowStartedByNexusOperation.run,
+                input,
+                id=str(uuid.uuid4()),
+            )
+    ```
+
+4. Register your service handler with a Temporal worker.
+    ```python
+    client = await Client.connect("localhost:7233", namespace="my-handler-namespace")
+    worker = Worker(
+        client,
+        task_queue="my-handler-task-queue",
+        workflows=[WorkflowStartedByNexusOperation],
+        nexus_service_handlers=[MyNexusServiceHandler()],
+    )
+    await worker.run()
+    ```
+
+5. Call your Nexus operations from your caller workflow.
+    ```python
+    @workflow.defn
+    class CallerWorkflow:
+        def __init__(self):
+            self.nexus_client = workflow.create_nexus_client(
+                service=MyNexusService, endpoint="my-nexus-endpoint"
+            )
+
+        @workflow.run
+        async def run(self, name: str) -> tuple[MyOutput, MyOutput]:
+            # Start the Nexus operation and wait for the result in one go, using execute_operation.
+            wf_result = await self.nexus_client.execute_operation(
+                MyNexusService.my_workflow_run_operation,
+                MyInput(name),
+            )
+            # Or alternatively, obtain the operation handle using start_operation,
+            # and then use it to get the result:
+            sync_operation_handle = await self.nexus_client.start_operation(
+                MyNexusService.my_sync_operation,
+                MyInput(name),
+            )
+            sync_result = await sync_operation_handle
+            return sync_result, wf_result
+    ```
+
+
 ### Workflow Replay
 
 Given a workflow's history, it can be replayed locally to check for things like non-determinism errors. For example,

pyproject.toml

@@ -1,16 +1,17 @@
 [project]
 name = "temporalio"
-version = "1.13.0"
+version = "1.14.1"
 description = "Temporal.io Python SDK"
 authors = [{ name = "Temporal Technologies Inc", email = "[email protected]" }]
-requires-python = "~=3.9"
+requires-python = ">=3.9"
 readme = "README.md"
 license = { file = "LICENSE" }
 keywords = [
     "temporal",
     "workflow",
 ]
 dependencies = [
+    "nexus-rpc>=1.1.0",
     "protobuf>=3.20,<6",
     "python-dateutil>=2.8.2,<3 ; python_version < '3.11'",
     "types-protobuf>=3.20",
@@ -25,7 +26,7 @@ opentelemetry = [
 ]
 pydantic = ["pydantic>=2.0.0,<3"]
 openai-agents = [
-    "openai-agents >= 0.0.19,<0.1",
+    "openai-agents >= 0.1,<0.2",
     "eval-type-backport>=0.2.2; python_version < '3.10'"
 ]
 
@@ -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.402",
     "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",
 ]
 
@@ -155,13 +158,24 @@ project-name = "Temporal Python"
 sidebar-expand-depth = 2
 
 [tool.pyright]
+reportUnknownVariableType = "none"
+reportUnknownParameterType = "none"
+reportUnusedCallResult = "none"
+reportImplicitStringConcatenation = "none"
+reportPrivateUsage = "none"
+reportExplicitAny = "none"
+reportMissingTypeArgument = "none"
+reportAny = "none"
+enableTypeIgnoreComments = true
+
 include = ["temporalio", "tests"]
 exclude = [
   "temporalio/api",
   "temporalio/bridge/proto",
   "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 +187,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",

temporalio/bridge/src/worker.rs

@@ -19,7 +19,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)]
@@ -557,6 +558,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>,
@@ -591,6 +604,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())
@@ -688,6 +717,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}")))
 }