Merge branch 'main' into plugins_to_core_debug

Author: yuandrew

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "temporalio"
-version = "1.18.1"
+version = "1.19.0"
 description = "Temporal.io Python SDK"
 authors = [{ name = "Temporal Technologies Inc", email = "[email protected]" }]
 requires-python = ">=3.10"
@@ -9,7 +9,7 @@ license = "MIT"
 license-files = ["LICENSE"]
 keywords = ["temporal", "workflow"]
 dependencies = [
-  "nexus-rpc==1.1.0",
+  "nexus-rpc==1.2.0",
   "protobuf>=3.20,<7.0.0",
   "python-dateutil>=2.8.2,<3 ; python_version < '3.11'",
   "types-protobuf>=3.20",
@@ -28,10 +28,7 @@ classifiers = [
 grpc = ["grpcio>=1.48.2,<2"]
 opentelemetry = ["opentelemetry-api>=1.11.1,<2", "opentelemetry-sdk>=1.11.1,<2"]
 pydantic = ["pydantic>=2.0.0,<3"]
-openai-agents = [
-  "openai-agents>=0.3,<0.5",
-  "mcp>=1.9.4, <2",
-]
+openai-agents = ["openai-agents>=0.3,<0.5", "mcp>=1.9.4, <2"]
 
 [project.urls]
 Homepage = "https://github.com/temporalio/sdk-python"

scripts/gen_bridge_client.py

@@ -171,6 +171,7 @@ def generate_rust_service_call(service_descriptor: ServiceDescriptor) -> str:
     py: Python<'p>,
     call: RpcCall,
   ) -> PyResult<Bound<'p, PyAny>> {
+    self.runtime.assert_same_process("use client")?;
     use temporal_client::${descriptor_name};
     let mut retry_client = self.retry_client.clone();
     self.runtime.future_into_py(py, async move {

temporalio/bridge/src/client.rs

@@ -93,6 +93,7 @@ pub fn connect_client<'a>(
     config: ClientConfig,
 ) -> PyResult<Bound<'a, PyAny>> {
     let opts: ClientOptions = config.try_into()?;
+    runtime_ref.runtime.assert_same_process("create client")?;
     let runtime = runtime_ref.runtime.clone();
     runtime_ref.runtime.future_into_py(py, async move {
         Ok(ClientRef {

temporalio/bridge/src/client_rpc_generated.rs

@@ -15,6 +15,7 @@ impl ClientRef {
         py: Python<'p>,
         call: RpcCall,
     ) -> PyResult<Bound<'p, PyAny>> {
+        self.runtime.assert_same_process("use client")?;
         use temporalio_client::WorkflowService;
         let mut retry_client = self.retry_client.clone();
         self.runtime.future_into_py(py, async move {
@@ -566,6 +567,7 @@ impl ClientRef {
         py: Python<'p>,
         call: RpcCall,
     ) -> PyResult<Bound<'p, PyAny>> {
+        self.runtime.assert_same_process("use client")?;
         use temporalio_client::OperatorService;
         let mut retry_client = self.retry_client.clone();
         self.runtime.future_into_py(py, async move {
@@ -628,6 +630,7 @@ impl ClientRef {
     }
 
     fn call_cloud_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<Bound<'p, PyAny>> {
+        self.runtime.assert_same_process("use client")?;
         use temporalio_client::CloudService;
         let mut retry_client = self.retry_client.clone();
         self.runtime.future_into_py(py, async move {
@@ -842,6 +845,7 @@ impl ClientRef {
     }
 
     fn call_test_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<Bound<'p, PyAny>> {
+        self.runtime.assert_same_process("use client")?;
         use temporalio_client::TestService;
         let mut retry_client = self.retry_client.clone();
         self.runtime.future_into_py(py, async move {
@@ -881,6 +885,7 @@ impl ClientRef {
     }
 
     fn call_health_service<'p>(&self, py: Python<'p>, call: RpcCall) -> PyResult<Bound<'p, PyAny>> {
+        self.runtime.assert_same_process("use client")?;
         use temporalio_client::HealthService;
         let mut retry_client = self.retry_client.clone();
         self.runtime.future_into_py(py, async move {

temporalio/bridge/src/runtime.rs

@@ -1,5 +1,5 @@
 use futures::channel::mpsc::Receiver;
-use pyo3::exceptions::{PyRuntimeError, PyValueError};
+use pyo3::exceptions::{PyAssertionError, PyRuntimeError, PyValueError};
 use pyo3::prelude::*;
 use pythonize::pythonize;
 use std::collections::HashMap;
@@ -33,6 +33,7 @@ pub struct RuntimeRef {
 
 #[derive(Clone)]
 pub(crate) struct Runtime {
+    pub(crate) pid: u32,
     pub(crate) core: Arc<CoreRuntime>,
     metrics_call_buffer: Option<Arc<MetricsCallBuffer<BufferedMetricRef>>>,
     log_forwarder_handle: Option<Arc<JoinHandle<()>>>,
@@ -189,6 +190,7 @@ pub fn init_runtime(options: RuntimeOptions) -> PyResult<RuntimeRef> {
 
     Ok(RuntimeRef {
         runtime: Runtime {
+            pid: std::process::id(),
             core: Arc::new(core),
             metrics_call_buffer,
             log_forwarder_handle,
@@ -213,6 +215,18 @@ impl Runtime {
         let _guard = self.core.tokio_handle().enter();
         pyo3_async_runtimes::generic::future_into_py::<TokioRuntime, _, T>(py, fut)
     }
+
+    pub(crate) fn assert_same_process(&self, action: &'static str) -> PyResult<()> {
+        let current_pid = std::process::id();
+        if self.pid != current_pid {
+            Err(PyAssertionError::new_err(format!(
+                "Cannot {} across forks (original runtime PID is {}, current is {})",
+                action, self.pid, current_pid,
+            )))
+        } else {
+            Ok(())
+        }
+    }
 }
 
 impl Drop for Runtime {

temporalio/bridge/src/worker.rs

@@ -493,6 +493,7 @@ pub fn new_worker(
     config: WorkerConfig,
 ) -> PyResult<WorkerRef> {
     enter_sync!(runtime_ref.runtime);
+    runtime_ref.runtime.assert_same_process("create worker")?;
     let event_loop_task_locals = Arc::new(OnceLock::new());
     let config = convert_worker_config(config, event_loop_task_locals.clone())?;
     let worker = temporalio_sdk_core::init_worker(
@@ -514,6 +515,9 @@ pub fn new_replay_worker<'a>(
     config: WorkerConfig,
 ) -> PyResult<Bound<'a, PyTuple>> {
     enter_sync!(runtime_ref.runtime);
+    runtime_ref
+        .runtime
+        .assert_same_process("create replay worker")?;
     let event_loop_task_locals = Arc::new(OnceLock::new());
     let config = convert_worker_config(config, event_loop_task_locals.clone())?;
     let (history_pusher, stream) = HistoryPusher::new(runtime_ref.runtime.clone());
@@ -538,6 +542,7 @@ pub fn new_replay_worker<'a>(
 #[pymethods]
 impl WorkerRef {
     fn validate<'p>(&self, py: Python<'p>) -> PyResult<Bound<PyAny, 'p>> {
+        self.runtime.assert_same_process("use worker")?;
         let worker = self.worker.as_ref().unwrap().clone();
         // Set custom slot supplier task locals so they can run futures.
         // Event loop is assumed to be running at this point.
@@ -557,6 +562,7 @@ impl WorkerRef {
     }
 
     fn poll_workflow_activation<'p>(&self, py: Python<'p>) -> PyResult<Bound<'p, PyAny>> {
+        self.runtime.assert_same_process("use worker")?;
         let worker = self.worker.as_ref().unwrap().clone();
         self.runtime.future_into_py(py, async move {
             let bytes = match worker.poll_workflow_activation().await {
@@ -569,6 +575,7 @@ impl WorkerRef {
     }
 
     fn poll_activity_task<'p>(&self, py: Python<'p>) -> PyResult<Bound<'p, PyAny>> {
+        self.runtime.assert_same_process("use worker")?;
         let worker = self.worker.as_ref().unwrap().clone();
         self.runtime.future_into_py(py, async move {
             let bytes = match worker.poll_activity_task().await {
@@ -581,6 +588,7 @@ impl WorkerRef {
     }
 
     fn poll_nexus_task<'p>(&self, py: Python<'p>) -> PyResult<Bound<'p, PyAny>> {
+        self.runtime.assert_same_process("use worker")?;
         let worker = self.worker.as_ref().unwrap().clone();
         self.runtime.future_into_py(py, async move {
             let bytes = match worker.poll_nexus_task().await {

temporalio/nexus/_decorators.py

@@ -117,7 +117,7 @@ async def _start(
             return WorkflowRunOperationHandler(_start)
 
         method_name = get_callable_name(start)
-        nexusrpc.set_operation_definition(
+        nexusrpc.set_operation(
             operation_handler_factory,
             nexusrpc.Operation(
                 name=name or method_name,

temporalio/nexus/_operation_handlers.py

@@ -10,13 +10,10 @@
     HandlerError,
     HandlerErrorType,
     InputT,
-    OperationInfo,
     OutputT,
 )
 from nexusrpc.handler import (
     CancelOperationContext,
-    FetchOperationInfoContext,
-    FetchOperationResultContext,
     OperationHandler,
     StartOperationContext,
     StartOperationResultAsync,
@@ -81,22 +78,6 @@ async def cancel(self, ctx: CancelOperationContext, token: str) -> None:
         """Cancel the operation, by cancelling the workflow."""
         await _cancel_workflow(token)
 
-    async def fetch_info(
-        self, ctx: FetchOperationInfoContext, token: str
-    ) -> OperationInfo:
-        """Fetch operation info (not supported for Temporal Nexus operations)."""
-        raise NotImplementedError(
-            "Temporal Nexus operation handlers do not support fetching operation info."
-        )
-
-    async def fetch_result(
-        self, ctx: FetchOperationResultContext, token: str
-    ) -> OutputT:
-        """Fetch operation result (not supported for Temporal Nexus operations)."""
-        raise NotImplementedError(
-            "Temporal Nexus operation handlers do not support fetching the operation result."
-        )
-
 
 async def _cancel_workflow(
     token: str,

temporalio/nexus/_util.py

@@ -129,12 +129,12 @@ def get_operation_factory(
 
     ``obj`` should be a decorated operation start method.
     """
-    op_defn = nexusrpc.get_operation_definition(obj)
+    op_defn = nexusrpc.get_operation(obj)
     if op_defn:
         factory = obj
     else:
         if factory := getattr(obj, "__nexus_operation_factory__", None):
-            op_defn = nexusrpc.get_operation_definition(factory)
+            op_defn = nexusrpc.get_operation(factory)
     if not isinstance(op_defn, nexusrpc.Operation):
         return None, None
     return factory, op_defn

temporalio/runtime.py

@@ -24,33 +24,82 @@
 import temporalio.bridge.runtime
 import temporalio.common
 
-_default_runtime: Optional[Runtime] = None
+
+class _RuntimeRef:
+    def __init__(
+        self,
+    ) -> None:
+        self._default_runtime: Runtime | None = None
+        self._prevent_default = False
+
+    def default(self) -> Runtime:
+        if not self._default_runtime:
+            if self._prevent_default:
+                raise RuntimeError(
+                    "Cannot create default Runtime after Runtime.prevent_default has been called"
+                )
+            self._default_runtime = Runtime(telemetry=TelemetryConfig())
+            self._default_created = True
+        return self._default_runtime
+
+    def prevent_default(self):
+        if self._default_runtime:
+            raise RuntimeError(
+                "Runtime.prevent_default called after default runtime has been created or set"
+            )
+        self._prevent_default = True
+
+    def set_default(
+        self, runtime: Runtime, *, error_if_already_set: bool = True
+    ) -> None:
+        if self._default_runtime and error_if_already_set:
+            raise RuntimeError("Runtime default already set")
+
+        self._default_runtime = runtime
+
+
+_runtime_ref: _RuntimeRef = _RuntimeRef()
 
 
 class Runtime:
     """Runtime for Temporal Python SDK.
 
-    Users are encouraged to use :py:meth:`default`. It can be set with
+    Most users are encouraged to use :py:meth:`default`. It can be set with
     :py:meth:`set_default`. Every time a new runtime is created, a new internal
     thread pool is created.
 
-    Runtimes do not work across forks.
+    Runtimes do not work across forks. Advanced users should consider using
+    :py:meth:`prevent_default` and `:py:meth`set_default` to ensure each
+    fork creates it's own runtime.
+
     """
 
     @classmethod
     def default(cls) -> Runtime:
-        """Get the default runtime, creating if not already created.
+        """Get the default runtime, creating if not already created. If :py:meth:`prevent_default`
+        is called before this method it will raise a RuntimeError instead of creating a default
+        runtime.
 
         If the default runtime needs to be different, it should be done with
         :py:meth:`set_default` before this is called or ever used.
 
         Returns:
             The default runtime.
         """
-        global _default_runtime
-        if not _default_runtime:
-            _default_runtime = cls(telemetry=TelemetryConfig())
-        return _default_runtime
+        global _runtime_ref
+        return _runtime_ref.default()
+
+    @classmethod
+    def prevent_default(cls):
+        """Prevent :py:meth:`default` from lazily creating a :py:class:`Runtime`.
+
+        Raises a RuntimeError if a default :py:class:`Runtime` has already been created.
+
+        Explicitly setting a default runtime with :py:meth:`set_default` bypasses this setting and
+        future calls to :py:meth:`default` will return the provided runtime.
+        """
+        global _runtime_ref
+        _runtime_ref.prevent_default()
 
     @staticmethod
     def set_default(runtime: Runtime, *, error_if_already_set: bool = True) -> None:
@@ -65,10 +114,8 @@ def set_default(runtime: Runtime, *, error_if_already_set: bool = True) -> None:
             error_if_already_set: If True and default is already set, this will
                 raise a RuntimeError.
         """
-        global _default_runtime
-        if _default_runtime and error_if_already_set:
-            raise RuntimeError("Runtime default already set")
-        _default_runtime = runtime
+        global _runtime_ref
+        _runtime_ref.set_default(runtime, error_if_already_set=error_if_already_set)
 
     def __init__(
         self,