Implement feature

Author: dandavison

temporalio/client.py

@@ -5877,6 +5877,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 +5938,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/nexus/_operation_context.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,10 @@
     ContextVar("temporal-cancel-operation-context")
 )
 
+_temporal_nexus_backing_workflow_start_context: ContextVar[bool] = ContextVar(
+    "temporal-nexus-backing-workflow-start-context"
+)
+
 
 @dataclass(frozen=True)
 class Info:
@@ -96,6 +102,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,48 +415,40 @@ 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,
-        #     };
-        # }
-
         # 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,
-        )
+        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/worker/_workflow_instance.py

@@ -3029,13 +3029,6 @@ def operation_token(self) -> Optional[str]:
     def __await__(self) -> Generator[Any, Any, OutputT]:
         return self._task.__await__()
 
-    def __repr__(self) -> str:
-        return (
-            f"{self._start_fut} "
-            f"{self._result_fut} "
-            f"Task[{self._task._state}] fut_waiter = {self._task._fut_waiter}) ({self._task._must_cancel})"  # type: ignore
-        )
-
     def cancel(self) -> bool:
         return self._task.cancel()