Reviewing

Author: Evanthx

temporalio/client/_impl.py

@@ -200,42 +200,25 @@ async def start_workflow(
             start_workflow_response=resp,
         )
         setattr(handle, "__temporal_eagerly_started", eagerly_started)
-        # If this start / signal-with-start is issued from inside a Nexus operation handler (but
-        # not as the nexus-backing workflow, whose links are handled separately by
-        # WorkflowRunOperationContext.start_workflow), capture the backlink the server returned so
-        # the caller workflow's Nexus history event links to the callee event.
+        # If this signal-with-start is issued from inside a Nexus operation handler (but not as the
+        # nexus-backing workflow, whose links are handled separately by
+        # WorkflowRunOperationContext.start_workflow), capture the signal backlink the server
+        # returned so the caller workflow's Nexus history event links to the signaled event. A
+        # plain start does not capture a backlink: it only forwards the inbound links onto the
+        # start request.
         nexus_ctx = self._try_nexus_start_operation_context()
         if (
             nexus_ctx is not None
             and not temporalio.nexus._operation_context._in_nexus_backing_workflow_start_context()
-        ):
-            if isinstance(
+            and isinstance(
                 resp,
                 temporalio.api.workflowservice.v1.SignalWithStartWorkflowExecutionResponse,
-            ):
-                # Server >= 1.31 with EnableCHASMSignalBacklinks returns signal_link pointing at
-                # the WorkflowExecutionSignaled event; older servers leave it unset.
-                if resp.HasField("signal_link"):
-                    nexus_ctx._add_backlink(resp.signal_link)
-            else:
-                if resp.HasField("link"):
-                    nexus_ctx._add_backlink(resp.link)
-                else:
-                    # Older servers (pre-1.31) don't return a link on the start response.
-                    # Fabricate one pointing at the started workflow's WorkflowExecutionStarted
-                    # event so the caller still gets a backlink.
-                    nexus_ctx._add_backlink(
-                        temporalio.api.common.v1.Link(
-                            workflow_event=temporalio.api.common.v1.Link.WorkflowEvent(
-                                namespace=self._client.namespace,
-                                workflow_id=req.workflow_id,
-                                run_id=resp.run_id,
-                                event_ref=temporalio.api.common.v1.Link.WorkflowEvent.EventReference(
-                                    event_type=temporalio.api.enums.v1.EventType.EVENT_TYPE_WORKFLOW_EXECUTION_STARTED,
-                                ),
-                            )
-                        )
-                    )
+            )
+        ):
+            # Server >= 1.31 with EnableCHASMSignalBacklinks returns signal_link pointing at
+            # the WorkflowExecutionSignaled event; older servers leave it unset.
+            if resp.HasField("signal_link"):
+                nexus_ctx._add_backlink(resp.signal_link)
         return handle
 
     async def _build_start_workflow_execution_request(

tests/nexus/test_signal_link_propagation.py

@@ -37,7 +37,6 @@
     SignalWorkflowInput,
     StartWorkflowInput,
 )
-from temporalio.nexus._link_conversion import _LinkType
 from temporalio.worker._nexus import _NexusTaskCancellation, _NexusWorker
 
 NAMESPACE = "test-namespace"
@@ -316,7 +315,9 @@ async def test_signal_with_start_against_older_server_captures_no_backlink(
 # ── start ─────────────────────────────────────────────────────────────────────────────────
 
 
-async def test_start_uses_server_link_when_present(nexus_ctx: Any) -> None:
+async def test_start_forwards_inbound_links_and_captures_no_backlink(
+    nexus_ctx: Any,
+) -> None:
     server_link = _workflow_event_link(
         WORKFLOW_ID,
         "target-run",
@@ -333,15 +334,18 @@ async def test_start_uses_server_link_when_present(nexus_ctx: Any) -> None:
 
     await impl.start_workflow(_start_input())
 
-    assert len(nexus_ctx.nexus_context.outbound_links) == 1
-    [link] = nexus_ctx.nexus_context.outbound_links
-    assert link.type == _LinkType.WORKFLOW.value
-    assert "wf-target" in link.url
-    # WorkflowExecutionStarted event type round-trips through the URL.
-    assert "WorkflowExecutionStarted" in link.url
+    # Forward: the start request carries the single inbound link.
+    sent = workflow_service.start_workflow_execution.call_args.args[0]
+    assert len(sent.links) == 1
+    assert sent.links[0] == _inbound_nexus_link()
+
+    # Backward: a plain start does not capture a backlink, even when the server returns one.
+    assert nexus_ctx.nexus_context.outbound_links == []
 
 
-async def test_start_fabricates_link_when_server_omits_it(nexus_ctx: Any) -> None:
+async def test_start_against_older_server_captures_no_backlink(
+    nexus_ctx: Any,
+) -> None:
     workflow_service = mock.MagicMock()
     workflow_service.start_workflow_execution = mock.AsyncMock(
         return_value=temporalio.api.workflowservice.v1.StartWorkflowExecutionResponse(
@@ -352,12 +356,13 @@ async def test_start_fabricates_link_when_server_omits_it(nexus_ctx: Any) -> Non
 
     await impl.start_workflow(_start_input())
 
-    assert len(nexus_ctx.nexus_context.outbound_links) == 1
-    [link] = nexus_ctx.nexus_context.outbound_links
-    # Fabricated link points at the started workflow's WorkflowExecutionStarted event.
-    assert "wf-target" in link.url
-    assert "target-run" in link.url
-    assert "WorkflowExecutionStarted" in link.url
+    # Forward direction still works regardless of server version.
+    sent = workflow_service.start_workflow_execution.call_args.args[0]
+    assert len(sent.links) == 1
+    assert sent.links[0] == _inbound_nexus_link()
+
+    # Backward: a plain start never fabricates a backlink.
+    assert nexus_ctx.nexus_context.outbound_links == []
 
 
 async def test_start_outside_nexus_context_does_not_touch_links() -> None:

tests/nexus/test_signal_link_propagation_e2e.py

@@ -59,7 +59,7 @@ class SignalingService:
 
 
 @workflow.defn
-class SignalCalleeWorkflow:
+class CalleeWorkflow:
     def __init__(self) -> None:
         self._received: list[str] = []
         self._expected = 1
@@ -79,7 +79,7 @@ def ping(self, msg: str) -> None:
 
 
 @workflow.defn
-class SignalCallerWorkflow:
+class CallerWorkflow:
     @workflow.run
     async def run(self, mode: str, callee_id: str, task_queue: str) -> str:
         client = workflow.create_nexus_client(
@@ -133,7 +133,7 @@ async def op(self, _ctx: StartOperationContext, input: str) -> str:
         await (
             nexus.client()
             .get_workflow_handle(callee_id)
-            .signal(SignalCalleeWorkflow.ping, "second")
+            .signal(CalleeWorkflow.ping, "second")
         )
         return "ok:sync"
 
@@ -155,7 +155,7 @@ async def _signal_with_start(callee_id: str, payload: str) -> None:
     # signal-with-start exercises the SignalWithStartWorkflowExecutionResponse.signal_link
     # backlink path in temporalio.client._impl.
     await nexus.client().start_workflow(
-        SignalCalleeWorkflow.run,
+        CalleeWorkflow.run,
         2 if payload == "first" else 1,
         id=callee_id,
         task_queue=nexus.info().task_queue,
@@ -251,17 +251,17 @@ async def test_sync_signal_operation_links(
 
     task_queue = str(uuid.uuid4())
     await env.create_nexus_endpoint(make_nexus_endpoint_name(task_queue), task_queue)
-    callee_id = f"sync-callee-{uuid.uuid4()}"
-    caller_id = f"sync-caller-{uuid.uuid4()}"
+    callee_id = f"callee-{uuid.uuid4()}"
+    caller_id = f"caller-{uuid.uuid4()}"
 
     async with Worker(
         client,
         task_queue=task_queue,
         nexus_service_handlers=[SignalingServiceHandler()],
-        workflows=[SignalCallerWorkflow, SignalCalleeWorkflow],
+        workflows=[CallerWorkflow, CalleeWorkflow],
     ):
         caller_handle = await client.start_workflow(
-            SignalCallerWorkflow.run,
+            CallerWorkflow.run,
             args=[MODE_SYNC, callee_id, task_queue],
             id=caller_id,
             task_queue=task_queue,
@@ -307,7 +307,7 @@ async def test_async_signal_operation_links(
         client,
         task_queue=task_queue,
         nexus_service_handlers=[AsyncSignalingServiceHandler()],
-        workflows=[AsyncSignalCallerWorkflow, SignalCalleeWorkflow],
+        workflows=[AsyncSignalCallerWorkflow, CalleeWorkflow],
     ):
         caller_handle = await client.start_workflow(
             AsyncSignalCallerWorkflow.run,