From b3cf13237c6603c3c1eac412738558452f7c32a6 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 27 Jul 2026 11:31:33 -0700 Subject: [PATCH 1/3] Fix System Nexus tracing headers --- CHANGELOG.md | 6 ++ .../contrib/opentelemetry/_interceptor.py | 10 ++ .../opentelemetry/_otel_interceptor.py | 10 ++ temporalio/nexus/system/__init__.py | 16 +++- temporalio/worker/__init__.py | 2 + temporalio/worker/_interceptor.py | 50 ++++++++++ temporalio/worker/_workflow_instance.py | 42 ++++++++- .../test_opentelemetry_plugin.py | 94 ++++++++++++++++++- tests/nexus/test_temporal_system_nexus.py | 19 +++- 9 files changed, 244 insertions(+), 5 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 0de28b563..eb17d47a2 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -57,9 +57,15 @@ to include examples, links to docs, or any other relevant information. `temporalio.client`) instead of setting `payload_limits` on `DataConverter`. Config fields were renamed to `payloads_warn_size` and `memo_warn_size`, and the deprecated `PayloadSizeWarning` was removed. +- `WorkflowOutboundInterceptor.start_nexus_operation` no longer receives Temporal System Nexus + operations. Custom interceptors that need to observe or modify these operations must implement + `start_system_nexus_operation` instead. ### Fixed +- Fixed OpenTelemetry context propagation when a workflow uses signal-with-start. Trace context is + now added to the called workflow's headers instead of the System Nexus transport headers. + ### Security ## [1.30.0] - 2026-07-01 diff --git a/temporalio/contrib/opentelemetry/_interceptor.py b/temporalio/contrib/opentelemetry/_interceptor.py index eb22f8be6..8bb8564f0 100644 --- a/temporalio/contrib/opentelemetry/_interceptor.py +++ b/temporalio/contrib/opentelemetry/_interceptor.py @@ -830,6 +830,16 @@ async def start_nexus_operation( return await super().start_nexus_operation(input) + async def start_system_nexus_operation( + self, input: temporalio.worker.StartSystemNexusOperationInput[Any, Any] + ) -> temporalio.workflow.NexusOperationHandle[Any]: + self.root._completed_span( + f"StartNexusOperation:{input.service}/{input.operation_name}", + kind=opentelemetry.trace.SpanKind.CLIENT, + add_to_outbound=input, + ) + return await super().start_system_nexus_operation(input) + def _carrier_to_nexus_headers( carrier: _CarrierDict, initial: Mapping[str, str] | None = None diff --git a/temporalio/contrib/opentelemetry/_otel_interceptor.py b/temporalio/contrib/opentelemetry/_otel_interceptor.py index c120fcd03..98875c849 100644 --- a/temporalio/contrib/opentelemetry/_otel_interceptor.py +++ b/temporalio/contrib/opentelemetry/_otel_interceptor.py @@ -600,3 +600,13 @@ async def start_nexus_operation( ): input.headers = _context_to_nexus_headers(input.headers or {}) return await super().start_nexus_operation(input) + + async def start_system_nexus_operation( + self, input: temporalio.worker.StartSystemNexusOperationInput[Any, Any] + ) -> temporalio.workflow.NexusOperationHandle[Any]: + with self._workflow_maybe_span( + f"StartNexusOperation:{input.service}/{input.operation_name}", + kind=opentelemetry.trace.SpanKind.CLIENT, + ): + input.headers = _context_to_headers(input.headers) + return await super().start_system_nexus_operation(input) diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 14a43cb72..e32cf2d88 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -8,9 +8,11 @@ import contextlib import contextvars -from collections.abc import Iterator, Sequence +from collections.abc import Iterator, Mapping, Sequence from typing import Any +from google.protobuf.message import Message + import temporalio.api.common.v1 import temporalio.converter from temporalio.bridge._visitor_functions import VisitorFunctions @@ -94,6 +96,18 @@ def is_system_endpoint(endpoint: str) -> bool: return endpoint == TEMPORAL_SYSTEM_ENDPOINT +def _apply_headers_to_request( + request: Message, + headers: Mapping[str, temporalio.api.common.v1.Payload], +) -> None: + """Apply headers to a system request when it supports Temporal headers.""" + if not headers or "header" not in request.DESCRIPTOR.fields_by_name: + return + request_header = getattr(request, "header") + for key, payload in headers.items(): + request_header.fields[key].CopyFrom(payload) + + async def _maybe_visit_payload( # pyright: ignore[reportUnusedFunction] endpoint: str, payload: temporalio.api.common.v1.Payload, diff --git a/temporalio/worker/__init__.py b/temporalio/worker/__init__.py index 4f6efe68c..ada34ab71 100644 --- a/temporalio/worker/__init__.py +++ b/temporalio/worker/__init__.py @@ -21,6 +21,7 @@ StartChildWorkflowInput, StartLocalActivityInput, StartNexusOperationInput, + StartSystemNexusOperationInput, WorkflowInboundInterceptor, WorkflowInterceptorClassInput, WorkflowOutboundInterceptor, @@ -100,6 +101,7 @@ "StartChildWorkflowInput", "StartLocalActivityInput", "StartNexusOperationInput", + "StartSystemNexusOperationInput", "WorkflowInterceptorClassInput", "ExecuteNexusOperationStartInput", "ExecuteNexusOperationCancelInput", diff --git a/temporalio/worker/_interceptor.py b/temporalio/worker/_interceptor.py index 4acf3c5d1..866d06ea0 100644 --- a/temporalio/worker/_interceptor.py +++ b/temporalio/worker/_interceptor.py @@ -348,6 +348,50 @@ def operation_name(self) -> str: raise ValueError(f"Operation is not a Nexus operation: {self.operation}") +@dataclass +class StartSystemNexusOperationInput(Generic[InputT, OutputT]): + """Input for :py:meth:`WorkflowOutboundInterceptor.start_system_nexus_operation`.""" + + service: str + operation: nexusrpc.Operation[InputT, OutputT] | str | Callable[..., Any] + input: InputT + schedule_to_close_timeout: timedelta | None + schedule_to_start_timeout: timedelta | None + start_to_close_timeout: timedelta | None + cancellation_type: temporalio.workflow.NexusOperationCancellationType + headers: Mapping[str, temporalio.api.common.v1.Payload] + summary: str | None + output_type: type[OutputT] | None = None + + def __post_init__(self) -> None: + """Initialize operation-specific attributes after dataclass creation.""" + if isinstance(self.operation, nexusrpc.Operation): + self.output_type = self.operation.output_type + elif callable(self.operation): + _, op = temporalio.nexus._util.get_operation_factory(self.operation) + if isinstance(op, nexusrpc.Operation): + self.output_type = op.output_type + else: + raise ValueError( + f"Operation callable is not a Nexus operation: {self.operation}" + ) + elif not isinstance(self.operation, str): + raise ValueError(f"Operation is not a Nexus operation: {self.operation}") + + @property + def operation_name(self) -> str: + """Get the name of the Nexus operation.""" + if isinstance(self.operation, nexusrpc.Operation): + return self.operation.name + elif isinstance(self.operation, str): + return self.operation + elif callable(self.operation): + _, op = temporalio.nexus._util.get_operation_factory(self.operation) + if isinstance(op, nexusrpc.Operation): + return op.name + raise ValueError(f"Operation is not a Nexus operation: {self.operation}") + + @dataclass class StartLocalActivityInput: """Input for :py:meth:`WorkflowOutboundInterceptor.start_local_activity`.""" @@ -481,6 +525,12 @@ async def start_nexus_operation( """Called for every :py:func:`temporalio.workflow.NexusClient.start_operation` call.""" return await self.next.start_nexus_operation(input) + async def start_system_nexus_operation( + self, input: StartSystemNexusOperationInput[InputT, OutputT] + ) -> temporalio.workflow.NexusOperationHandle[OutputT]: + """Called for every Temporal System Nexus operation started by a workflow.""" + return await self.next.start_system_nexus_operation(input) + @dataclass class ExecuteNexusOperationStartInput: diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 76b3304ef..ca887aa77 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -76,6 +76,7 @@ StartChildWorkflowInput, StartLocalActivityInput, StartNexusOperationInput, + StartSystemNexusOperationInput, WorkflowInboundInterceptor, WorkflowOutboundInterceptor, ) @@ -1677,7 +1678,21 @@ async def workflow_start_nexus_operation( headers: Mapping[str, str] | None, summary: str | None, ) -> temporalio.workflow.NexusOperationHandle[OutputT]: - # start_nexus_operation + if temporalio.nexus.system.is_system_endpoint(endpoint): + return await self._outbound.start_system_nexus_operation( + StartSystemNexusOperationInput( + service=service, + operation=operation, + input=input, + output_type=output_type, + schedule_to_close_timeout=schedule_to_close_timeout, + schedule_to_start_timeout=schedule_to_start_timeout, + start_to_close_timeout=start_to_close_timeout, + cancellation_type=cancellation_type, + headers={}, + summary=summary, + ) + ) return await self._outbound.start_nexus_operation( StartNexusOperationInput( endpoint=endpoint, @@ -2166,6 +2181,26 @@ async def operation_handle_fn() -> OutputT: if self._cancel_reason is not None or self._deleting: raise + async def _outbound_start_system_nexus_operation( + self, input: StartSystemNexusOperationInput[Any, OutputT] + ) -> _NexusOperationHandle[OutputT]: + temporalio.nexus.system._apply_headers_to_request(input.input, input.headers) + return await self._outbound_start_nexus_operation( + StartNexusOperationInput( + endpoint=temporalio.nexus.system.TEMPORAL_SYSTEM_ENDPOINT, + service=input.service, + operation=input.operation, + input=input.input, + output_type=input.output_type, + schedule_to_close_timeout=input.schedule_to_close_timeout, + schedule_to_start_timeout=input.schedule_to_start_timeout, + start_to_close_timeout=input.start_to_close_timeout, + cancellation_type=input.cancellation_type, + headers=None, + summary=input.summary, + ) + ) + #### Miscellaneous helpers #### # These are in alphabetical order. @@ -3095,6 +3130,11 @@ async def start_nexus_operation( ) -> _NexusOperationHandle[OutputT]: return await self._instance._outbound_start_nexus_operation(input) + async def start_system_nexus_operation( + self, input: StartSystemNexusOperationInput[Any, OutputT] + ) -> _NexusOperationHandle[OutputT]: + return await self._instance._outbound_start_system_nexus_operation(input) + def start_local_activity( self, input: StartLocalActivityInput ) -> temporalio.workflow.ActivityHandle[Any]: diff --git a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py index 3fd50e89b..15e153269 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py +++ b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py @@ -6,6 +6,7 @@ import nexusrpc import opentelemetry.trace import pytest +from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( InMemorySpanExporter, @@ -17,9 +18,14 @@ import temporalio.contrib.opentelemetry.workflow from temporalio import activity, nexus, workflow from temporalio.client import Client, WorkflowFailureError -from temporalio.contrib.opentelemetry import OpenTelemetryPlugin, create_tracer_provider +from temporalio.contrib.opentelemetry import ( + OpenTelemetryPlugin, + TracingInterceptor, + create_tracer_provider, +) from temporalio.exceptions import ApplicationError from temporalio.testing import WorkflowEnvironment +from temporalio.worker import UnsandboxedWorkflowRunner, Worker # Import the dump_spans function from the original opentelemetry test from tests.contrib.opentelemetry.test_opentelemetry import dump_spans @@ -81,6 +87,34 @@ async def run(self): return +@workflow.defn +class SignalWithStartHeaderWorkflow: + def __init__(self) -> None: + self._signaled = False + + @workflow.run + async def run(self) -> bool: + await workflow.wait_condition(lambda: self._signaled) + return "_tracer-data" in workflow.info().headers + + @workflow.signal + def notify(self) -> None: + self._signaled = True + + +@workflow.defn +class SignalWithStartCallerWorkflow: + @workflow.run + async def run(self, target_id: str, task_queue: str) -> str: + handle = await workflow.signal_with_start_workflow( + SignalWithStartHeaderWorkflow.run, + id=target_id, + task_queue=task_queue, + signal=SignalWithStartHeaderWorkflow.notify, + ) + return handle.id + + async def test_otel_tracing_basic(client: Client, reset_otel_tracer_provider: Any): # type: ignore[reportUnusedParameter] exporter = InMemorySpanExporter() provider = create_tracer_provider() @@ -128,6 +162,64 @@ async def test_otel_tracing_basic(client: Client, reset_otel_tracer_provider: An ) +async def test_otel_workflow_signal_with_start_propagates_trace_headers( + client: Client, + env: WorkflowEnvironment, + reset_otel_tracer_provider: Any, # type: ignore[reportUnusedParameter] +): + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + provider = create_tracer_provider() + opentelemetry.trace.set_tracer_provider(provider) + config = client.config() + config["plugins"] = [OpenTelemetryPlugin()] + client = Client(**config) + + async with new_worker( + client, SignalWithStartCallerWorkflow, SignalWithStartHeaderWorkflow + ) as worker: + target_id = f"signal-with-start-target-{uuid.uuid4()}" + with get_tracer(__name__).start_as_current_span("signal-with-start"): + caller = await client.start_workflow( + SignalWithStartCallerWorkflow.run, + args=[target_id, worker.task_queue], + id=f"signal-with-start-caller-{uuid.uuid4()}", + task_queue=worker.task_queue, + execution_timeout=timedelta(seconds=3), + ) + assert await caller.result() == target_id + assert await client.get_workflow_handle(target_id).result() is True + + +async def test_legacy_otel_workflow_signal_with_start_propagates_trace_headers( + client: Client, env: WorkflowEnvironment +): + if env.supports_time_skipping: + pytest.skip("Nexus tests don't work with the Java test server") + provider = TracerProvider() + tracer = provider.get_tracer(__name__) + config = client.config() + config["interceptors"] = [TracingInterceptor(tracer)] + client = Client(**config) + + async with Worker( + client, + task_queue=f"signal-with-start-{uuid.uuid4()}", + workflows=[SignalWithStartCallerWorkflow, SignalWithStartHeaderWorkflow], + workflow_runner=UnsandboxedWorkflowRunner(), + ) as worker: + target_id = f"signal-with-start-target-{uuid.uuid4()}" + with tracer.start_as_current_span("signal-with-start"): + caller = await client.start_workflow( + SignalWithStartCallerWorkflow.run, + args=[target_id, worker.task_queue], + id=f"signal-with-start-caller-{uuid.uuid4()}", + task_queue=worker.task_queue, + ) + assert await caller.result() == target_id + assert await client.get_workflow_handle(target_id).result() is True + + @workflow.defn class ComprehensiveWorkflow: def __init__(self) -> None: diff --git a/tests/nexus/test_temporal_system_nexus.py b/tests/nexus/test_temporal_system_nexus.py index eb8ee603c..4381a90fa 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -25,6 +25,7 @@ from temporalio.worker import ( Interceptor, StartNexusOperationInput, + StartSystemNexusOperationInput, Worker, WorkflowInboundInterceptor, WorkflowInterceptorClassInput, @@ -116,6 +117,12 @@ async def start_nexus_operation( interceptor_traces.append(("workflow.start_nexus_operation", input)) return await super().start_nexus_operation(input) + async def start_system_nexus_operation( + self, input: StartSystemNexusOperationInput[Any, Any] + ) -> workflow.NexusOperationHandle[Any]: + interceptor_traces.append(("workflow.start_system_nexus_operation", input)) + return await super().start_system_nexus_operation(input) + def _assert_stored_payloads_include( driver: InMemoryTestDriver, expected_payload_data: set[bytes] @@ -132,8 +139,8 @@ def _assert_stored_payloads_include( def _assert_start_nexus_operation_interceptor_trace() -> None: assert len(interceptor_traces) == 1 trace_name, trace_value = interceptor_traces.pop() - assert trace_name == "workflow.start_nexus_operation" - trace_input = cast(StartNexusOperationInput[Any, Any], trace_value) + assert trace_name == "workflow.start_system_nexus_operation" + trace_input = cast(StartSystemNexusOperationInput[Any, Any], trace_value) request = cast( workflowservice_pb2.SignalWithStartWorkflowExecutionRequest, trace_input.input, @@ -143,6 +150,14 @@ def _assert_start_nexus_operation_interceptor_trace() -> None: assert request.workflow_type.name == "test-workflow" +def test_system_nexus_headers_are_ignored_when_request_has_no_header_field() -> None: + request = workflowservice_pb2.GetSystemInfoRequest() + nexus_system._apply_headers_to_request( + request, {"trace": temporalio.api.common.v1.Payload(data=b"context")} + ) + assert request == workflowservice_pb2.GetSystemInfoRequest() + + class _MarkingPayloadVisitor: def __init__(self) -> None: self.visited_payload_count = 0 From b63ac76232cfbc2b93aa71af87b77c4f7ee564b2 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 27 Jul 2026 12:35:59 -0700 Subject: [PATCH 2/3] Silence System Nexus helper type warning --- temporalio/nexus/system/__init__.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index e32cf2d88..4ba383c06 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -96,7 +96,7 @@ def is_system_endpoint(endpoint: str) -> bool: return endpoint == TEMPORAL_SYSTEM_ENDPOINT -def _apply_headers_to_request( +def _apply_headers_to_request( # pyright: ignore[reportUnusedFunction] request: Message, headers: Mapping[str, temporalio.api.common.v1.Payload], ) -> None: From 53f4c6726d527362244dc457dd2bf78c192d1e0e Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 27 Jul 2026 12:39:06 -0700 Subject: [PATCH 3/3] Simplify System Nexus interceptor input --- temporalio/worker/_interceptor.py | 15 --------------- 1 file changed, 15 deletions(-) diff --git a/temporalio/worker/_interceptor.py b/temporalio/worker/_interceptor.py index 866d06ea0..76209ab56 100644 --- a/temporalio/worker/_interceptor.py +++ b/temporalio/worker/_interceptor.py @@ -363,21 +363,6 @@ class StartSystemNexusOperationInput(Generic[InputT, OutputT]): summary: str | None output_type: type[OutputT] | None = None - def __post_init__(self) -> None: - """Initialize operation-specific attributes after dataclass creation.""" - if isinstance(self.operation, nexusrpc.Operation): - self.output_type = self.operation.output_type - elif callable(self.operation): - _, op = temporalio.nexus._util.get_operation_factory(self.operation) - if isinstance(op, nexusrpc.Operation): - self.output_type = op.output_type - else: - raise ValueError( - f"Operation callable is not a Nexus operation: {self.operation}" - ) - elif not isinstance(self.operation, str): - raise ValueError(f"Operation is not a Nexus operation: {self.operation}") - @property def operation_name(self) -> str: """Get the name of the Nexus operation."""