diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py index 06ffd9b0f..54f8a987d 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -766,9 +766,9 @@ 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: @@ -776,15 +776,10 @@ def _apply_nexus_context_to_start_activity_request( # pyright: ignore[reportUnu 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] diff --git a/tests/nexus/test_link_propagation.py b/tests/nexus/test_link_propagation.py index 2c9f6eec7..a7b78426c 100644 --- a/tests/nexus/test_link_propagation.py +++ b/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( diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index d966c5cc6..70254542d 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/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 ):