From 5e088af185ccf6db976df75dbd03052373986cd9 Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Tue, 4 Aug 2026 15:44:21 -0500 Subject: [PATCH 1/4] Adding QueryWorkflowResponse Link for Nexus --- CHANGELOG.md | 2 + temporalio/client/_impl.py | 3 + temporalio/nexus/_operation_context.py | 6 +- tests/nexus/test_link_propagation.py | 85 ++++++++++++++++++++++++-- tests/nexus/test_temporal_operation.py | 80 ++++++++++++++++++++++++ 5 files changed, 169 insertions(+), 7 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index b3c6f7ae5..20ddd464f 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -64,6 +64,8 @@ to include examples, links to docs, or any other relevant information. variable names a worker will read. - **Experimental**: `temporalio.contrib.openai_agents.AllowAllWorkerEnvVars` allowlists every environment variable name on the worker. +- Added Nexus operation link propagation for Workflow Queries issued from operation handlers. The + queried Workflow link returned by the server is attached to the caller's Nexus operation event. ### Changed diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index e4c46cf80..ce2ef8a8a 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -436,6 +436,9 @@ async def query_workflow(self, input: QueryWorkflowInput) -> Any: raise WorkflowQueryFailedError(err.message) else: raise + nexus_ctx = temporalio.nexus._operation_context._try_start_operation_context() + if nexus_ctx is not None and resp.HasField("link"): + nexus_ctx._add_response_link(resp.link) if resp.HasField("query_rejected"): raise WorkflowQueryRejectedError( WorkflowExecutionStatus(resp.query_rejected.status) diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py index cdc8140dd..fa1a92fcf 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -300,9 +300,9 @@ def _add_start_workflow_response_link( 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, it is - converted to a Nexus link and added to the operation's outbound links. + ``link`` is the ``common.v1.Link`` returned by a Temporal RPC (or ``None`` against a + server that did not return one). When present, it is converted to a Nexus link and added + to the operation's outbound links. This is only safe to call from the single thread/task that runs the operation handler. """ diff --git a/tests/nexus/test_link_propagation.py b/tests/nexus/test_link_propagation.py index c332cab04..72379b032 100644 --- a/tests/nexus/test_link_propagation.py +++ b/tests/nexus/test_link_propagation.py @@ -1,9 +1,9 @@ """Unit tests for Nexus link propagation. -These exercise link propagation when a Nexus operation handler signals a workflow or -starts a workflow, workflow update, or activity against a mocked workflow service. -End-to-end signal backlinks require a server with EnableCHASMSignalBacklinks enabled -and are not covered here. +These exercise link propagation when a Nexus operation handler queries or signals a +workflow, or starts a workflow, workflow update, or activity against a mocked workflow +service. End-to-end signal backlinks require a server with +EnableCHASMSignalBacklinks enabled and are not covered here. """ from __future__ import annotations @@ -38,6 +38,7 @@ import temporalio.nexus._token from temporalio.client._impl import _ClientImpl from temporalio.client._interceptor import ( + QueryWorkflowInput, SignalWorkflowInput, StartActivityInput, StartWorkflowInput, @@ -68,6 +69,19 @@ def _workflow_event_link( ) +def _workflow_link( + workflow_id: str, run_id: str, *, reason: str +) -> temporalio.api.common.v1.Link: + return temporalio.api.common.v1.Link( + workflow=temporalio.api.common.v1.Link.Workflow( + namespace=NAMESPACE, + workflow_id=workflow_id, + run_id=run_id, + reason=reason, + ) + ) + + def _inbound_nexus_link() -> temporalio.api.common.v1.Link: return _workflow_event_link( "caller-wf", @@ -137,6 +151,20 @@ def _signal_input() -> SignalWorkflowInput: ) +def _query_input() -> QueryWorkflowInput: + return QueryWorkflowInput( + id=WORKFLOW_ID, + run_id=None, + query="query-done", + args=[], + reject_condition=None, + headers={}, + ret_type=bool, + rpc_metadata={}, + rpc_timeout=None, + ) + + def _start_input(start_signal: str | None = None) -> StartWorkflowInput: return StartWorkflowInput( workflow="TestWorkflow", @@ -222,9 +250,56 @@ def _outbound_link_urls(ctx: Any) -> list[str]: return [link.url for link in ctx.nexus_context.outbound_links] +def test_response_link_captures_workflow_link( + nexus_ctx: _TemporalStartOperationContext, +) -> None: + nexus_ctx._add_response_link( + _workflow_link(WORKFLOW_ID, "target-run", reason="Query processed") + ) + + assert nexus_ctx.nexus_context.outbound_links == [ + nexusrpc.Link( + type=temporalio.api.common.v1.Link.Workflow.DESCRIPTOR.full_name, + url=( + "temporal:///namespaces/test-namespace/workflows/" + "wf-target/target-run?reason=Query+processed" + ), + ) + ] + + # ── signal ──────────────────────────────────────────────────────────────────────────────── +# Query responses differ from Signal responses by linking to the Workflow rather than an event. +async def test_query_captures_response_workflow_link( + nexus_ctx: _TemporalStartOperationContext, +) -> None: + payloads = await temporalio.converter.DataConverter.default.encode([False]) + workflow_service = mock.MagicMock() + workflow_service.query_workflow = mock.AsyncMock( + return_value=temporalio.api.workflowservice.v1.QueryWorkflowResponse( + query_result=temporalio.api.common.v1.Payloads(payloads=payloads), + link=_workflow_link(WORKFLOW_ID, "target-run", reason="Query processed"), + ) + ) + impl = _make_client_impl(workflow_service) + + result = await impl.query_workflow(_query_input()) + + assert result is False + assert nexus_ctx.nexus_context.outbound_links == [ + nexusrpc.Link( + type=temporalio.api.common.v1.Link.Workflow.DESCRIPTOR.full_name, + url=( + "temporal:///namespaces/test-namespace/workflows/" + "wf-target/target-run?reason=Query+processed" + ), + ) + ] + + +# Signal responses link to the event that accepted the Signal. async def test_signal_forwards_inbound_links_and_captures_response_backlink( nexus_ctx: _TemporalStartOperationContext, ) -> None: @@ -521,6 +596,8 @@ async def test_start_outside_nexus_context_leaves_on_conflict_options_unset() -> # ── workflow update ────────────────────────────────────────────────────────────── + + def _workflow_update_response( link: temporalio.api.common.v1.Link | None = None, ) -> temporalio.api.workflowservice.v1.UpdateWorkflowExecutionResponse: diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index 85948deb3..f91ed5338 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -116,6 +116,7 @@ class TestService: sync_result: Operation[Input, str] custom_cancel: Operation[str, None] update_op: Operation[Input, str] + query_op: Operation[str, bool] echo_activity: Operation[Input, str] error_activity: Operation[Input, None] blocking_activity: Operation[str, None] @@ -292,6 +293,17 @@ async def update_op( update_id=input.update_id, ) + @nexus.temporal_operation + async def query_op( + self, + _ctx: nexus.TemporalStartOperationContext, + client: nexus.TemporalNexusClient, + input: str, + ) -> nexus.TemporalOperationResult[bool]: + handle = client.client.get_workflow_handle(input) + result = await handle.query(BlockingWorkflow.query_done) + return nexus.TemporalOperationResult.sync(result) + @nexus.temporal_operation async def echo_activity( self, @@ -824,6 +836,74 @@ async def run(self) -> None: async def unblock(self): self.done = True + @workflow.query + def query_done(self) -> bool: + return self.done + + +@workflow.defn +class QueryWorkflowCaller: + @workflow.run + async def run(self, input: Input) -> bool: + client = workflow.create_nexus_client( + service=TestService, endpoint=make_nexus_endpoint_name(input.task_queue) + ) + return await client.execute_operation(TestService.query_op, input.value) + + +async def test_temporal_operation_query_workflow( + client: Client, env: WorkflowEnvironment +) -> None: + task_queue = str(uuid.uuid4()) + endpoint_name = make_nexus_endpoint_name(task_queue) + await env.create_nexus_endpoint(endpoint_name, task_queue) + target_workflow_id = f"query-target-{uuid.uuid4()}" + + async with Worker( + env.client, + task_queue=task_queue, + nexus_service_handlers=[TestServiceHandler()], + workflows=[BlockingWorkflow, QueryWorkflowCaller], + ): + target_handle = await client.start_workflow( + BlockingWorkflow.run, + id=target_workflow_id, + task_queue=task_queue, + ) + caller_handle = await client.start_workflow( + QueryWorkflowCaller.run, + Input(value=target_workflow_id, task_queue=task_queue), + id=f"query-caller-{uuid.uuid4()}", + task_queue=task_queue, + ) + + try: + assert not await caller_handle.result() + + caller_history = await caller_handle.fetch_history() + completed_event = next( + event + for event in caller_history.events + if event.event_type == EventType.EVENT_TYPE_NEXUS_OPERATION_COMPLETED + ) + + target_history = await target_handle.fetch_history() + assert not any(event.links for event in target_history.events) + + if not completed_event.links: + pytest.skip("server did not return a Workflow Query response link") + assert target_handle.result_run_id is not None + assert Link( + workflow=Link.Workflow( + namespace=client.namespace, + workflow_id=target_workflow_id, + run_id=target_handle.result_run_id, + reason="Query processed", + ) + ) in list(completed_event.links) + finally: + await target_handle.cancel() + @workflow.defn class CancelBlockingWorkflowCaller: From 338c5c9c24480529e1354ac13911ed99a77c9d30 Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Mon, 24 Aug 2026 11:06:59 -0500 Subject: [PATCH 2/4] Fixing formatting checks --- tests/nexus/test_link_propagation.py | 2 -- 1 file changed, 2 deletions(-) diff --git a/tests/nexus/test_link_propagation.py b/tests/nexus/test_link_propagation.py index 72379b032..eb9360c8c 100644 --- a/tests/nexus/test_link_propagation.py +++ b/tests/nexus/test_link_propagation.py @@ -596,8 +596,6 @@ async def test_start_outside_nexus_context_leaves_on_conflict_options_unset() -> # ── workflow update ────────────────────────────────────────────────────────────── - - def _workflow_update_response( link: temporalio.api.common.v1.Link | None = None, ) -> temporalio.api.workflowservice.v1.UpdateWorkflowExecutionResponse: From 9d128d86d532ed8b015a00d3091764aeeac0e1ed Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Mon, 24 Aug 2026 13:48:25 -0500 Subject: [PATCH 3/4] Moving link logic to _operation_context --- temporalio/client/_impl.py | 6 +++--- temporalio/nexus/_operation_context.py | 9 +++++++++ 2 files changed, 12 insertions(+), 3 deletions(-) diff --git a/temporalio/client/_impl.py b/temporalio/client/_impl.py index ce2ef8a8a..78471baf7 100644 --- a/temporalio/client/_impl.py +++ b/temporalio/client/_impl.py @@ -436,9 +436,9 @@ async def query_workflow(self, input: QueryWorkflowInput) -> Any: raise WorkflowQueryFailedError(err.message) else: raise - nexus_ctx = temporalio.nexus._operation_context._try_start_operation_context() - if nexus_ctx is not None and resp.HasField("link"): - nexus_ctx._add_response_link(resp.link) + temporalio.nexus._operation_context._apply_query_workflow_response_to_nexus_context( + resp + ) if resp.HasField("query_rejected"): raise WorkflowQueryRejectedError( WorkflowExecutionStatus(resp.query_rejected.status) diff --git a/temporalio/nexus/_operation_context.py b/temporalio/nexus/_operation_context.py index fa1a92fcf..c44088f21 100644 --- a/temporalio/nexus/_operation_context.py +++ b/temporalio/nexus/_operation_context.py @@ -777,6 +777,15 @@ def _apply_start_workflow_update_response_to_nexus_context( # pyright: ignore[r nexus_ctx._add_response_link(resp.link) +def _apply_query_workflow_response_to_nexus_context( # pyright: ignore[reportUnusedFunction] + resp: temporalio.api.workflowservice.v1.QueryWorkflowResponse, +) -> None: + """Apply a workflow query response link to the current Nexus context.""" + nexus_ctx = _try_start_operation_context() + if nexus_ctx is not None and resp.HasField("link"): + nexus_ctx._add_response_link(resp.link) + + def _apply_nexus_context_to_signal_workflow_request( # pyright: ignore[reportUnusedFunction] req: temporalio.api.workflowservice.v1.SignalWorkflowExecutionRequest, ) -> None: From c93e5f6ece997ff327ef8876efd7d4d0954b599b Mon Sep 17 00:00:00 2001 From: Joshua Frenchwood Date: Mon, 24 Aug 2026 14:20:26 -0500 Subject: [PATCH 4/4] remove skip for query_workflow operation test --- tests/nexus/test_temporal_operation.py | 14 +++++++++++--- 1 file changed, 11 insertions(+), 3 deletions(-) diff --git a/tests/nexus/test_temporal_operation.py b/tests/nexus/test_temporal_operation.py index f91ed5338..983e9ea03 100644 --- a/tests/nexus/test_temporal_operation.py +++ b/tests/nexus/test_temporal_operation.py @@ -44,6 +44,9 @@ # See https://github.com/temporalio/sdk-python/issues/1704. pytestmark = pytest.mark.requires_local_server +# Query response links require a newer server than the shared test environment. +_QUERY_LINK_DEV_SERVER_DOWNLOAD_VERSION = "v1.8.3-server-1.32.0-162.0" + @dataclass class Input: @@ -851,7 +854,14 @@ async def run(self, input: Input) -> bool: return await client.execute_operation(TestService.query_op, input.value) -async def test_temporal_operation_query_workflow( +async def test_temporal_operation_query_workflow() -> None: + async with await WorkflowEnvironment.start_local( + dev_server_download_version=_QUERY_LINK_DEV_SERVER_DOWNLOAD_VERSION + ) as env: + await _assert_temporal_operation_query_workflow(env.client, env) + + +async def _assert_temporal_operation_query_workflow( client: Client, env: WorkflowEnvironment ) -> None: task_queue = str(uuid.uuid4()) @@ -890,8 +900,6 @@ async def test_temporal_operation_query_workflow( target_history = await target_handle.fetch_history() assert not any(event.links for event in target_history.events) - if not completed_event.links: - pytest.skip("server did not return a Workflow Query response link") assert target_handle.result_run_id is not None assert Link( workflow=Link.Workflow(