Merge branch 'main' into openai/mcp

Author: tconley1428

.github/workflows/ci.yml

@@ -75,7 +75,7 @@ jobs:
       - run: poe build-develop
       - run: mkdir junit-xml
       - run: poe test ${{matrix.pytestExtraArgs}} -s --junit-xml=junit-xml/${{ matrix.python }}--${{ matrix.os }}.xml
-        timeout-minutes: 10
+        timeout-minutes: 15
       # 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

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "temporalio"
-version = "1.15.0"
+version = "1.16.0"
 description = "Temporal.io Python SDK"
 authors = [{ name = "Temporal Technologies Inc", email = "[email protected]" }]
 requires-python = ">=3.9"

temporalio/bridge/sdk-core

@@ -57,6 +57,7 @@
 import temporalio.converter
 import temporalio.exceptions
 import temporalio.nexus
+import temporalio.nexus._operation_context
 import temporalio.runtime
 import temporalio.service
 import temporalio.workflow
@@ -5877,6 +5878,12 @@ async def _build_start_workflow_execution_request(
         )
         # Links are duplicated on request for compatibility with older server versions.
         req.links.extend(links)
+
+        if temporalio.nexus._operation_context._in_nexus_backing_workflow_start_context():
+            req.on_conflict_options.attach_request_id = True
+            req.on_conflict_options.attach_completion_callbacks = True
+            req.on_conflict_options.attach_links = True
+
         return req
 
     async def _build_signal_with_start_workflow_execution_request(
@@ -5932,6 +5939,7 @@ async def _populate_start_workflow_execution_request(
             "temporalio.api.enums.v1.WorkflowIdConflictPolicy.ValueType",
             int(input.id_conflict_policy),
         )
+
         if input.retry_policy is not None:
             input.retry_policy.apply_to_proto(req.retry_policy)
         req.cron_schedule = input.cron_schedule

temporalio/client.py

@@ -982,7 +982,7 @@ class Priority:
     2. Then consider "fairness_key" and "fairness_weight" for fairness balancing.
     """
 
-    priority_key: Optional[int]
+    priority_key: Optional[int] = None
     """Priority key is a positive integer from 1 to n, where smaller integers correspond to higher
     priorities (tasks run sooner). In general, tasks in a queue should be processed in close to
     priority order, although small deviations are possible.
@@ -997,22 +997,22 @@ class Priority:
     fairness_key: Optional[str] = None
     """A short string (max 64 bytes) that is used as a key for a fairness balancing mechanism.
     This can correspond to a tenant id or even fixed strings like "high", "low", etc.
-    
+
     The fairness mechanism attempts to dispatch tasks for a given key in proportion to its weight.
-    For example, using a thousand distinct tenant ids, each with a weight of 1.0 will result in 
+    For example, using a thousand distinct tenant ids, each with a weight of 1.0 will result in
     each tenant getting a roughly equal share of task dispatch throughput.
-    
+
     Default is an empty string.
     """
 
     fairness_weight: Optional[float] = None
     """A float that represents the weight for task dispatch for the associated fairness key.
     Tasks for a fairness key are dispatched in proportion to their weight.
-    
+
     Server will clamp the values between 0.001 and 1000. Default weight is 1.0.
-    
+
     The effective weight of a task is determined by the following precedence (highest to lowest):
-    1. Weights overridden in task queue configuration 
+    1. Weights overridden in task queue configuration
     2. Weights attached to workflow/activity
     3. Default weight of 1.0
     """

temporalio/common.py

@@ -3,13 +3,15 @@
 import dataclasses
 import logging
 from collections.abc import Awaitable, Mapping, MutableMapping, Sequence
+from contextlib import contextmanager
 from contextvars import ContextVar
 from dataclasses import dataclass
 from datetime import timedelta
 from typing import (
     TYPE_CHECKING,
     Any,
     Callable,
+    Generator,
     Optional,
     Union,
     overload,
@@ -47,6 +49,14 @@
     ContextVar("temporal-cancel-operation-context")
 )
 
+# A Nexus start handler might start zero or more workflows as usual using a Temporal client. In
+# addition, it may start one "nexus-backing" workflow, using
+# WorkflowRunOperationContext.start_workflow. This context is active while the latter is being done.
+# It is thus a narrower context than _temporal_start_operation_context.
+_temporal_nexus_backing_workflow_start_context: ContextVar[bool] = ContextVar(
+    "temporal-nexus-backing-workflow-start-context"
+)
+
 
 @dataclass(frozen=True)
 class Info:
@@ -96,6 +106,19 @@ def _try_temporal_context() -> (
     return start_ctx or cancel_ctx
 
 
+@contextmanager
+def _nexus_backing_workflow_start_context() -> Generator[None, None, None]:
+    token = _temporal_nexus_backing_workflow_start_context.set(True)
+    try:
+        yield
+    finally:
+        _temporal_nexus_backing_workflow_start_context.reset(token)
+
+
+def _in_nexus_backing_workflow_start_context() -> bool:
+    return _temporal_nexus_backing_workflow_start_context.get(False)
+
+
 @dataclass
 class _TemporalStartOperationContext:
     """Context for a Nexus start operation being handled by a Temporal Nexus Worker."""
@@ -396,56 +419,46 @@ async def start_workflow(
             Nexus caller is itself a workflow, this means that the workflow in the caller
             namespace web UI will contain links to the started workflow, and vice versa.
         """
-        # TODO(nexus-preview): When sdk-python supports on_conflict_options, Typescript does this:
-        # if (workflowOptions.workflowIdConflictPolicy === 'USE_EXISTING') {
-        #     internalOptions.onConflictOptions = {
-        #     attachLinks: true,
-        #     attachCompletionCallbacks: true,
-        #     attachRequestId: true,
-        #     };
-        # }
-        if (
-            id_conflict_policy
-            == temporalio.common.WorkflowIDConflictPolicy.USE_EXISTING
-        ):
-            raise RuntimeError(
-                "WorkflowIDConflictPolicy.USE_EXISTING is not yet supported when starting a workflow "
-                "that backs a Nexus operation (Python SDK Nexus support is at Pre-release stage)."
-            )
-
         # We must pass nexus_completion_callbacks, workflow_event_links, and request_id,
         # but these are deliberately not exposed in overloads, hence the type-check
         # violation.
-        wf_handle = await self._temporal_context.client.start_workflow(  # type: ignore
-            workflow=workflow,
-            arg=arg,
-            args=args,
-            id=id,
-            task_queue=task_queue or self._temporal_context.info().task_queue,
-            result_type=result_type,
-            execution_timeout=execution_timeout,
-            run_timeout=run_timeout,
-            task_timeout=task_timeout,
-            id_reuse_policy=id_reuse_policy,
-            id_conflict_policy=id_conflict_policy,
-            retry_policy=retry_policy,
-            cron_schedule=cron_schedule,
-            memo=memo,
-            search_attributes=search_attributes,
-            static_summary=static_summary,
-            static_details=static_details,
-            start_delay=start_delay,
-            start_signal=start_signal,
-            start_signal_args=start_signal_args,
-            rpc_metadata=rpc_metadata,
-            rpc_timeout=rpc_timeout,
-            request_eager_start=request_eager_start,
-            priority=priority,
-            versioning_override=versioning_override,
-            callbacks=self._temporal_context._get_callbacks(),
-            workflow_event_links=self._temporal_context._get_workflow_event_links(),
-            request_id=self._temporal_context.nexus_context.request_id,
-        )
+
+        # Here we are starting a "nexus-backing" workflow. That means that the StartWorkflow request
+        # contains nexus-specific data such as a completion callback (used by the handler server
+        # namespace to deliver the result to the caller namespace when the workflow reaches a
+        # terminal state) and inbound links to the caller workflow (attached to history events of
+        # the workflow started in the handler namespace, and displayed in the UI).
+        with _nexus_backing_workflow_start_context():
+            wf_handle = await self._temporal_context.client.start_workflow(  # type: ignore
+                workflow=workflow,
+                arg=arg,
+                args=args,
+                id=id,
+                task_queue=task_queue or self._temporal_context.info().task_queue,
+                result_type=result_type,
+                execution_timeout=execution_timeout,
+                run_timeout=run_timeout,
+                task_timeout=task_timeout,
+                id_reuse_policy=id_reuse_policy,
+                id_conflict_policy=id_conflict_policy,
+                retry_policy=retry_policy,
+                cron_schedule=cron_schedule,
+                memo=memo,
+                search_attributes=search_attributes,
+                static_summary=static_summary,
+                static_details=static_details,
+                start_delay=start_delay,
+                start_signal=start_signal,
+                start_signal_args=start_signal_args,
+                rpc_metadata=rpc_metadata,
+                rpc_timeout=rpc_timeout,
+                request_eager_start=request_eager_start,
+                priority=priority,
+                versioning_override=versioning_override,
+                callbacks=self._temporal_context._get_callbacks(),
+                workflow_event_links=self._temporal_context._get_workflow_event_links(),
+                request_id=self._temporal_context.nexus_context.request_id,
+            )
 
         self._temporal_context._add_outbound_links(wf_handle)
 

temporalio/nexus/_operation_context.py

@@ -26,7 +26,7 @@
 import temporalio.exceptions
 import temporalio.runtime
 
-__version__ = "1.15.0"
+__version__ = "1.16.0"
 
 ServiceRequest = TypeVar("ServiceRequest", bound=google.protobuf.message.Message)
 ServiceResponse = TypeVar("ServiceResponse", bound=google.protobuf.message.Message)

temporalio/service.py

@@ -298,6 +298,7 @@ class StartNexusOperationInput(Generic[InputT, OutputT]):
     operation: Union[nexusrpc.Operation[InputT, OutputT], str, Callable[..., Any]]
     input: InputT
     schedule_to_close_timeout: Optional[timedelta]
+    cancellation_type: temporalio.workflow.NexusOperationCancellationType
     headers: Optional[Mapping[str, str]]
     output_type: Optional[Type[OutputT]] = None