Adding links to Nexus signals (#1593)

Evanthx · VegetarianOrc · web-flow · commit 31a0a1a881be · 2026-06-24T22:07:44.000Z
* Signal work

* Reviewing

* Renaming to RequestLink and ResponseLink

* Addressing PR comments

* Updating method name

* Removed unused test methods

* Adding tests, fixed an issue

* Adding tests

* Fix a linter issue

* Added entry to Changelog

* Fixing a race condition

* centralize and simplify start workflow response link handling. Move private start context accessor to nexus package. Update tests to reflect that response links are created when using plain start workflow

* run formatter, address lint

* move changelog entry. Remove inaccurate comment

---------

Co-authored-by: Alex Mazzeo &lt;alex.mazzeo@temporal.io&gt;
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -20,6 +20,11 @@ to include examples, links to docs, or any other relevant information.
 
 ### Added
 
+- Nexus operation link propagation for signals. When a Nexus operation handler signals a workflow
+  (including signal-with-start), the inbound Nexus request links are now forwarded onto the signaled
+  workflow so its history events link back to the caller, and the link the server returns for the
+  signaled event is attached to the caller workflow's Nexus operation history event. This makes the
+  caller and callee mutually navigable in the UI for signal-based Nexus operations.
 - Exposed `backoff_start_interval` for continue-as-new, to allow the new workflow to start after a delay.
 
 ### Changed
diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py
@@ -200,6 +200,9 @@ async def start_workflow(
             start_workflow_response=resp,
         )
         setattr(handle, "__temporal_eagerly_started", eagerly_started)
+        nexus_ctx = temporalio.nexus._operation_context._try_start_operation_context()
+        if nexus_ctx is not None:
+            nexus_ctx._add_start_workflow_response_link(handle)
         return handle
 
     async def _build_start_workflow_execution_request(
@@ -237,10 +240,24 @@ 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():
+        nexus_ctx = temporalio.nexus._operation_context._try_start_operation_context()
+        if nexus_ctx is not None:
+            # This start was issued from inside a Nexus operation handler. If the workflow ID
+            # conflict policy is WORKFLOW_ID_CONFLICT_POLICY_USE_EXISTING and a conflict is
+            # detected, attach this request's request ID, completion callbacks, and links to
+            # the existing run. The TemporalNexusClient and WorkflowRunOperationContext are
+            # responsible for setting the callbacks correctly, so it is safe to enable all
+            # on-conflict options whenever we are invoked from an operation handler.
             req.on_conflict_options.attach_request_id = True
             req.on_conflict_options.attach_completion_callbacks = True
             req.on_conflict_options.attach_links = True
+            # The nexus-backing workflow already carries its inbound links via input.links
+            # (start_workflow forwards them as links=...). A plain start_workflow issued from
+            # inside a Nexus operation handler must forward the inbound Nexus task links
+            # explicitly so the started callee's WorkflowExecutionStarted event links back to
+            # the caller.
+            if not temporalio.nexus._operation_context._in_nexus_backing_workflow_start_context():
+                req.links.extend(nexus_ctx._get_request_links())
 
         return req
 
@@ -267,6 +284,15 @@ async def _build_signal_with_start_workflow_execution_request(
                 await data_converter.encode(input.start_signal_args)
             )
         await self._populate_start_workflow_execution_request(req, input)
+        # If this signal-with-start is issued from inside a Nexus operation handler (but not the
+        # nexus-backing workflow), forward the inbound Nexus task links so both the callee's
+        # WorkflowExecutionStarted and WorkflowExecutionSignaled events link back to the caller.
+        if not temporalio.nexus._operation_context._in_nexus_backing_workflow_start_context():
+            nexus_ctx = (
+                temporalio.nexus._operation_context._try_start_operation_context()
+            )
+            if nexus_ctx is not None:
+                req.links.extend(nexus_ctx._get_request_links())
         return req
 
     async def _build_update_with_start_start_workflow_execution_request(
@@ -500,9 +526,18 @@ async def signal_workflow(self, input: SignalWorkflowInput) -> None:
             req.input.payloads.extend(await data_converter.encode(input.args))
         if input.headers is not None:  # type:ignore[reportUnnecessaryComparison]
             await self._apply_headers(input.headers, req.header.fields)
-        await self._client.workflow_service.signal_workflow_execution(
+        # If this signal is issued from inside a Nexus operation handler, forward the inbound
+        # Nexus task links so the WorkflowExecutionSignaled event links back to the caller.
+        nexus_ctx = temporalio.nexus._operation_context._try_start_operation_context()
+        if nexus_ctx is not None:
+            req.links.extend(nexus_ctx._get_request_links())
+        resp = await self._client.workflow_service.signal_workflow_execution(
             req, retry=True, metadata=input.rpc_metadata, timeout=input.rpc_timeout
         )
+        # Server >= 1.31 with EnableCHASMSignalBacklinks returns a response link pointing at the
+        # signal event; older servers leave it unset. Propagate when present.
+        if nexus_ctx is not None and resp.HasField("link"):
+            nexus_ctx._add_response_link(resp.link)
 
     async def terminate_workflow(self, input: TerminateWorkflowInput) -> None:
         data_converter = self._client.data_converter._with_contexts(
diff --git a/temporalio/nexus/_link_conversion.py b/temporalio/nexus/_link_conversion.py
@@ -20,7 +20,7 @@
 logger = logging.getLogger(__name__)
 
 _NEXUS_OPERATION_LINK_URL_PATH_REGEX = re.compile(
-    r"^/namespaces/(?P<namespace>[^/]+)/nexus-operations/(?P<operation_id>[^/]+)$"
+    r"^/namespaces/(?P<namespace>[^/]+)/nexus-operations/(?P<operation_id>[^/]+)/(?P<run_id>[^/]*)/details$"
 )
 
 _WORKFLOW_LINK_URL_PATH_REGEX = re.compile(
@@ -38,7 +38,6 @@ class _LinkType(str, Enum):
 LINK_EVENT_TYPE_PARAM_NAME = "eventType"
 LINK_REQUEST_ID_PARAM_NAME = "requestID"
 LINK_REFERENCE_TYPE_PARAM_NAME = "referenceType"
-LINK_RUN_ID_PARAM_NAME = "runID"
 LINK_REASON_PARAM_NAME = "reason"
 
 EVENT_REFERENCE_TYPE = "EventReference"
@@ -182,18 +181,11 @@ def nexus_operation_to_nexus_link(
     """
     namespace = urllib.parse.quote(op_link.namespace, safe="")
     operation_id = urllib.parse.quote(op_link.operation_id, safe="")
-    path = f"/namespaces/{namespace}/nexus-operations/{operation_id}"
-
-    query_params = ""
-    if op_link.run_id:
-        query_params = urllib.parse.urlencode(
-            {
-                LINK_RUN_ID_PARAM_NAME: op_link.run_id,
-            },
-        )
+    run_id = urllib.parse.quote(op_link.run_id, safe="")
+    path = f"/namespaces/{namespace}/nexus-operations/{operation_id}/{run_id}/details"
 
     return nexusrpc.Link(
-        url=_temporal_nexus_url(path, query_params=query_params),
+        url=_temporal_nexus_url(path),
         type=_LinkType.NEXUS_OPERATION.value,
     )
 
@@ -333,19 +325,11 @@ def nexus_link_to_nexus_operation_link(
         )
         return None
 
-    query_params = urllib.parse.parse_qs(url.query)
-
-    try:
-        run_id = _optional_single_query_param(query_params, LINK_RUN_ID_PARAM_NAME)
-    except ValueError as err:
-        logger.warning(f"Invalid Nexus link: {nexus_link}. {err}")
-        return None
-
     groups = match.groupdict()
     nexus_op_link = temporalio.api.common.v1.Link.NexusOperation(
         namespace=urllib.parse.unquote(groups["namespace"]),
         operation_id=urllib.parse.unquote(groups["operation_id"]),
-        run_id=run_id,
+        run_id=urllib.parse.unquote(groups["run_id"]),
     )
     return temporalio.api.common.v1.Link(nexus_operation=nexus_op_link)
 
diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py
@@ -23,6 +23,7 @@
     overload,
 )
 
+import nexusrpc
 from nexusrpc.handler import (
     CancelOperationContext,
     OperationContext,
@@ -44,6 +45,7 @@
 
 from ._link_conversion import (
     nexus_link_to_temporal_link,
+    temporal_link_to_nexus_link,
     workflow_event_to_nexus_link,
     workflow_execution_started_event_link_from_workflow_handle,
 )
@@ -167,6 +169,11 @@ def _try_temporal_context() -> (
     return start_ctx or cancel_ctx
 
 
+def _try_start_operation_context() -> _TemporalStartOperationContext | None:  # pyright: ignore[reportUnusedFunction]
+    """The Nexus start-operation context if a handler is currently running, else None."""
+    return _temporal_start_operation_context.get(None)
+
+
 @contextmanager
 def _nexus_backing_workflow_start_context() -> Generator[None]:
     token = _temporal_nexus_backing_workflow_start_context.set(True)
@@ -239,44 +246,73 @@ def _get_callbacks(self, token: str) -> list[temporalio.client.Callback]:
             else []
         )
 
-    def _get_links(
-        self,
-    ) -> list[temporalio.api.common.v1.Link]:
+    def _get_request_links(self) -> list[temporalio.api.common.v1.Link]:
+        """Request links to attach to RPCs the operation handler issues.
+
+        These are the inbound Nexus task links. When the operation handler signals,
+        signal-with-starts, or starts a workflow, these links are added to the request's
+        ``links`` field so the callee's history event links back to whatever scheduled this
+        Nexus operation.
+        """
         event_links: list[temporalio.api.common.v1.Link] = []
         for inbound_link in self.nexus_context.inbound_links:
             if link := nexus_link_to_temporal_link(inbound_link):
                 event_links.append(link)
         return event_links
 
-    def _add_outbound_links(
+    def _add_start_workflow_response_link(
         self, workflow_handle: temporalio.client.WorkflowHandle[Any, Any]
     ):
-        # If links were not sent in StartWorkflowExecutionResponse then construct them.
-        wf_event_links: list[temporalio.api.common.v1.Link.WorkflowEvent] = []
-        try:
-            if isinstance(
-                workflow_handle._start_workflow_response,
-                temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse,
-            ):
-                if workflow_handle._start_workflow_response.HasField("link"):
-                    if link := workflow_handle._start_workflow_response.link:
-                        if link.HasField("workflow_event"):
-                            wf_event_links.append(link.workflow_event)
-            if not wf_event_links:
-                wf_event_links = [
-                    workflow_execution_started_event_link_from_workflow_handle(
+        response = workflow_handle._start_workflow_response
+
+        nexus_link: nexusrpc.Link | None = None
+        if isinstance(
+            response, temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse
+        ):
+            if response.HasField("link"):
+                nexus_link = temporal_link_to_nexus_link(response.link)
+            else:
+                # If a link was not sent in response then construct it.
+                link = temporalio.api.common.v1.Link(
+                    workflow_event=workflow_execution_started_event_link_from_workflow_handle(
                         workflow_handle,
                         self.nexus_context.request_id,
                     )
-                ]
-            self.nexus_context.outbound_links.extend(
-                workflow_event_to_nexus_link(link) for link in wf_event_links
-            )
+                )
+                nexus_link = temporal_link_to_nexus_link(link)
+
+        elif isinstance(
+            response,
+            temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse,
+        ):
+            # Server >= 1.31 with EnableCHASMSignalBacklinks returns signal_link pointing at
+            # the WorkflowExecutionSignaled event; older servers leave it unset.
+            if response.HasField("signal_link"):
+                nexus_link = temporal_link_to_nexus_link(response.signal_link)
+
+        try:
+            if nexus_link is not None:
+                self.nexus_context.outbound_links.append(nexus_link)
         except Exception as e:
             logger.warning(
-                f"Failed to create WorkflowExecutionStarted event links for workflow {workflow_handle}: {e}"
+                f"Failed to create event links for workflow {workflow_handle}: {e}"
             )
-        return workflow_handle
+
+    def _add_response_link(self, link: temporalio.api.common.v1.Link | None) -> None:
+        """Append a response link returned by an RPC the operation handler issued.
+
+        ``link`` is the ``common.v1.Link`` returned on a signal, signal-with-start, or start
+        response (or ``None`` against a server that did not return one). When present and of the
+        ``workflow_event`` variant, it is converted to a Nexus link and added to the operation's
+        outbound links so the caller workflow's Nexus history event links to the callee event.
+
+        This is only safe to call from the single thread/task that runs the operation handler.
+        """
+        if link is None or not link.HasField("workflow_event"):
+            return
+        self.nexus_context.outbound_links.append(
+            workflow_event_to_nexus_link(link.workflow_event)
+        )
 
 
 class WorkflowRunOperationContext(StartOperationContext):
@@ -674,10 +710,8 @@ async def _start_nexus_backing_workflow(
             priority=priority,
             versioning_override=versioning_override,
             callbacks=temporal_context._get_callbacks(token),
-            links=temporal_context._get_links(),
+            links=temporal_context._get_request_links(),
             request_id=temporal_context.nexus_context.request_id,
         )
 
-    temporal_context._add_outbound_links(wf_handle)
-
     return WorkflowHandle[ReturnType]._unsafe_from_client_workflow_handle(wf_handle)
diff --git a/temporalio/worker/_workflow.py b/temporalio/worker/_workflow.py
@@ -518,12 +518,13 @@ async def _handle_activation(
 
         # Log workflow task duration with external storage metrics
         self._log_workflow_task_duration(
-            act, task_start_time, download_metrics, upload_metrics
+            act, workflow, task_start_time, download_metrics, upload_metrics
         )
 
     def _log_workflow_task_duration(
         self,
         act: temporalio.bridge.proto.workflow_activation.WorkflowActivation,
+        workflow: _RunningWorkflow | None,
         task_start_time: float,
         download_metrics: temporalio.converter._extstore.StorageOperationMetrics,
         upload_metrics: temporalio.converter._extstore.StorageOperationMetrics,
@@ -537,8 +538,7 @@ def _fmt_duration(td: timedelta) -> str:
             return f"{secs * 1000:.3f}ms"
 
         completed_event_id = act.history_length + 1
-        _running = self._running_workflows.get(act.run_id)
-        _info = _running.get_info() if _running is not None else None
+        _info = workflow.get_info() if workflow is not None else None
         attempt = _info.attempt if _info is not None else "unknown"
         log_id = f"{act.run_id}:{completed_event_id}:{attempt}"
         msg_details, extra = temporalio.workflow._build_log_context(
diff --git a/tests/conftest.py b/tests/conftest.py
@@ -134,6 +134,8 @@ async def env(env_type: str) -> AsyncGenerator[WorkflowEnvironment, None]:
                 "--dynamic-config-value",
                 "history.enableChasmCallbacks=true",
                 "--dynamic-config-value",
+                "history.enableCHASMSignalBacklinks=true",
+                "--dynamic-config-value",
                 "nexusoperation.enableStandalone=true",
                 "--dynamic-config-value",
                 'system.system.refreshNexusEndpointsMinWait="0s"',
diff --git a/tests/nexus/test_link_conversion.py b/tests/nexus/test_link_conversion.py
@@ -277,7 +277,7 @@ def test_link_conversion_workflow_to_link_and_back(
             ),
             nexusrpc.Link(
                 type=temporalio.api.common.v1.Link.NexusOperation.DESCRIPTOR.full_name,
-                url="temporal:///namespaces/ns/nexus-operations/op-id?runID=run-id",
+                url="temporal:///namespaces/ns/nexus-operations/op-id/run-id/details",
             ),
         ),
         (
@@ -289,7 +289,7 @@ def test_link_conversion_workflow_to_link_and_back(
             ),
             nexusrpc.Link(
                 type=temporalio.api.common.v1.Link.NexusOperation.DESCRIPTOR.full_name,
-                url="temporal:///namespaces/ns/nexus-operations/op-id",
+                url="temporal:///namespaces/ns/nexus-operations/op-id//details",
             ),
         ),
         (
@@ -301,7 +301,7 @@ def test_link_conversion_workflow_to_link_and_back(
             ),
             nexusrpc.Link(
                 type=temporalio.api.common.v1.Link.NexusOperation.DESCRIPTOR.full_name,
-                url="temporal:///namespaces/ns/nexus-operations/op%2Fid",
+                url="temporal:///namespaces/ns/nexus-operations/op%2Fid//details",
             ),
         ),
     ],
@@ -332,10 +332,12 @@ def test_link_conversion_nexus_operation_to_link_and_back(
     )
 
 
-def test_nexus_operation_link_with_duplicate_run_id_is_ignored():
+def test_nexus_operation_link_with_unparseable_url_is_ignored():
+    # The canonical path is /nexus-operations/{op_id}/{run_id}/details; a URL missing the
+    # run-id/details suffix (e.g. the legacy ?runID= form) does not parse.
     link = nexusrpc.Link(
         type=temporalio.api.common.v1.Link.NexusOperation.DESCRIPTOR.full_name,
-        url="temporal:///namespaces/ns/nexus-operations/op-id?runID=one&runID=two",
+        url="temporal:///namespaces/ns/nexus-operations/op-id?runID=run-id",
     )
     assert temporalio.nexus._link_conversion.nexus_link_to_temporal_link(link) is None
 
diff --git a/tests/nexus/test_signal_link_propagation.py b/tests/nexus/test_signal_link_propagation.py
diff --git a/tests/nexus/test_signal_link_propagation_e2e.py b/tests/nexus/test_signal_link_propagation_e2e.py
