Fix duplicate links for Nexus backing SAA

temporalio/nexus/_operation_context.py

@@ -766,25 +766,20 @@ def _apply_nexus_context_to_start_activity_request(  # pyright: ignore[reportUnu
     """Apply the current Nexus operation context to an activity start request.
 
     This is a no-op outside a Nexus operation context. Within one, it attaches
-    the Nexus request ID and inbound links and configures conflict handling to
-    preserve the Nexus metadata. Completion callbacks are added only when the
-    activity is backing the Nexus operation.
+    the Nexus request ID and configures conflict handling to preserve the Nexus
+    metadata. Inbound links are attached to the completion callback when the
+    activity backs the operation and to the request otherwise.
     """
     nexus_ctx = _try_start_operation_context()
     if nexus_ctx is not None:
         req.on_conflict_options.attach_request_id = True
         req.on_conflict_options.attach_completion_callbacks = True
         req.on_conflict_options.attach_links = True
 
-        # Add request_id and all Nexus links if we're in a Nexus context, backing or otherwise
         req.request_id = nexus_ctx.nexus_context.request_id
         request_links = nexus_ctx._get_request_links()
 
-        # Links are duplicated on request for compatibility with older server versions.
-        req.links.extend(request_links)
-
         if _in_nexus_backing_start_context():
-            # Add callbacks only if we're in a backing Nexus context
             callbacks = nexus_ctx._get_callbacks(
                 OperationToken(
                     type=OperationTokenType.ACTIVITY,
@@ -802,6 +797,8 @@ def _apply_nexus_context_to_start_activity_request(  # pyright: ignore[reportUnu
                 )
                 for callback in callbacks
             )
+        else:
+            req.links.extend(request_links)
 
 
 def _apply_start_activity_response_to_nexus_context(  # pyright: ignore[reportUnusedFunction]

tests/nexus/test_link_propagation.py

@@ -531,8 +531,7 @@ async def test_backing_activity_start_gets_nexus_request_fields() -> None:
             _start_activity_input()
         )
 
-    assert len(req.links) == 1
-    assert req.links[0] == _inbound_nexus_link()
+    assert len(req.links) == 0
     assert req.request_id == "req-id"
     assert len(req.completion_callbacks) == 1
     operation_token = temporalio.nexus._token.OperationToken.decode(

tests/nexus/test_temporal_operation.py

@@ -16,6 +16,7 @@
 
 import temporalio.exceptions
 from temporalio import activity, nexus, workflow
+from temporalio.api.activity.v1 import ActivityExecutionInfo
 from temporalio.api.common.v1 import Link
 from temporalio.client import (
     ActivityExecutionStatus,
@@ -33,7 +34,11 @@
 from temporalio.testing import WorkflowEnvironment
 from temporalio.worker import Worker
 from tests.helpers import EventType, assert_event_subsequence, assert_eventually
-from tests.helpers.nexus import make_nexus_endpoint_name
+from tests.helpers.nexus import (
+    assert_links_match,
+    expected_nexus_operation_link,
+    make_nexus_endpoint_name,
+)
 
 # Cloud CI's namespace credentials cannot manage Nexus endpoints.
 # See https://github.com/temporalio/sdk-python/issues/1704.
@@ -1132,6 +1137,67 @@ async def test_temporal_operation_start_activity(
         assert result == "test"
 
 
+async def test_temporal_operation_backing_activity_does_not_duplicate_links(
+    client: Client, env: WorkflowEnvironment
+):
+    if env.supports_time_skipping:
+        pytest.skip(
+            "Standalone Nexus Operation tests don't work with time-skipping server"
+        )
+
+    task_queue = str(uuid.uuid4())
+    endpoint_name = make_nexus_endpoint_name(task_queue)
+    await env.create_nexus_endpoint(endpoint_name, task_queue)
+    activity_id = f"link-activity-{uuid.uuid4()}"
+
+    @service_handler
+    class LinkActivityHandler:
+        @nexus.temporal_operation
+        async def echo_activity(
+            self,
+            _ctx: nexus.TemporalStartOperationContext,
+            client: nexus.TemporalNexusClient,
+            input: Input,
+        ) -> nexus.TemporalOperationResult[str]:
+            return await client.start_activity(
+                echo_activity,
+                input,
+                id=activity_id,
+                start_to_close_timeout=timedelta(seconds=5),
+            )
+
+    async with Worker(
+        env.client,
+        task_queue=task_queue,
+        nexus_service_handlers=[LinkActivityHandler()],
+        activities=[echo_activity],
+    ):
+        nexus_client = client.create_nexus_client(LinkActivityHandler, endpoint_name)
+        operation_handle = await nexus_client.start_operation(
+            LinkActivityHandler.echo_activity,
+            Input(value="test", task_queue=task_queue),
+            id=str(uuid.uuid4()),
+        )
+
+        assert await operation_handle.result() == "test"
+        activity_description = await client.get_activity_handle(activity_id).describe()
+        assert isinstance(activity_description.raw_info, ActivityExecutionInfo)
+        assert operation_handle.run_id is not None
+        callback_links = [
+            link
+            for callback in activity_description.raw_callbacks
+            for link in callback.info.callback.links
+        ]
+        assert_links_match(
+            [*activity_description.raw_info.links, *callback_links],
+            expected_nexus_operation_link(
+                namespace=client.namespace,
+                operation_id=operation_handle.operation_id,
+                run_id=operation_handle.run_id,
+            ),
+        )
+
+
 async def test_temporal_operation_start_activity_raises_error(
     client: Client, env: WorkflowEnvironment
 ):