From e2047dd2c5ef1950dcc0539ad84467061be550dd Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 27 Jul 2026 11:31:33 -0700 Subject: [PATCH 1/5] Fix System Nexus tracing headers --- CHANGELOG.md | 7 +++ .../contrib/opentelemetry/_interceptor.py | 10 ++++ .../opentelemetry/_otel_interceptor.py | 10 ++++ temporalio/worker/__init__.py | 2 + temporalio/worker/_interceptor.py | 50 +++++++++++++++++++ temporalio/worker/_workflow_instance.py | 42 +++++++++++++++- 6 files changed, 120 insertions(+), 1 deletion(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index 17d90a43d..6135dd6fc 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -153,11 +153,18 @@ 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 - Marked system Nexus envelope payloads so nested payloads can be detected and visited after the envelope is already stored as a payload. +- 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/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 eec5f903c..5be3f61e0 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -76,6 +76,7 @@ StartChildWorkflowInput, StartLocalActivityInput, StartNexusOperationInput, + StartSystemNexusOperationInput, WorkflowInboundInterceptor, WorkflowOutboundInterceptor, ) @@ -1724,7 +1725,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, @@ -2171,6 +2186,26 @@ async def operation_handle_fn() -> OutputT: ) return handle + 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. @@ -3157,6 +3192,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]: From 10fb379bf4f40e7f267f80a9613ed301860df244 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 27 Jul 2026 12:35:59 -0700 Subject: [PATCH 2/5] Silence System Nexus helper type warning --- temporalio/nexus/system/__init__.py | 18 +++++++++++++++--- 1 file changed, 15 insertions(+), 3 deletions(-) diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index 730dd4258..d68d51ee2 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -8,10 +8,12 @@ import contextlib import contextvars -from collections.abc import Iterator, Sequence +from collections.abc import Iterator, Mapping, Sequence from dataclasses import dataclass from typing import Any +from google.protobuf.message import Message + import temporalio.api.common.v1 import temporalio.common import temporalio.converter @@ -132,13 +134,23 @@ 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) + + def _is_system_payload(payload: temporalio.api.common.v1.Payload) -> bool: return ( payload.metadata.get(_SYSTEM_PAYLOAD_METADATA_KEY) == _SYSTEM_PAYLOAD_METADATA_VALUE ) - - async def maybe_visit_payload( payload: temporalio.api.common.v1.Payload, visitor_functions: VisitorFunctions, From b26919f1bef09743723d9384a5151f5a245bbf46 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Mon, 27 Jul 2026 12:39:06 -0700 Subject: [PATCH 3/5] 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.""" From ff318a56c265bbdeb5f7a5bd38eedc36c4bf577f Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 20 Aug 2026 09:51:30 -0700 Subject: [PATCH 4/5] Adapt System Nexus tracing to generated request types --- temporalio/converter/_payload_converter.py | 14 ++++- temporalio/nexus/system/__init__.py | 31 ++++------ temporalio/worker/_workflow_instance.py | 57 +++++++++++++----- .../opentelemetry/test_opentelemetry.py | 60 +++++++++++++++++++ .../test_opentelemetry_plugin.py | 57 ++++++++++++++++++ tests/nexus/test_temporal_system_nexus.py | 11 +++- 6 files changed, 195 insertions(+), 35 deletions(-) diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index a8bc35e28..ac1a921a9 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -613,7 +613,9 @@ def wrap(payload_converter: PayloadConverter) -> PayloadConverter: return _TemporalTransferTypePayloadConverter(payload_converter) def to_payloads( - self, values: Sequence[Any] + self, + values: Sequence[Any], + headers: Mapping[str, temporalio.api.common.v1.Payload] | None = None, ) -> list[temporalio.api.common.v1.Payload]: """See base class.""" transfer_type_values: list[Any] = [] @@ -621,6 +623,16 @@ def to_payloads( converter = _get_transfer_type_converter(type(value)) if converter is not None: value = converter.to_transfer_type(value) + if ( + headers + and isinstance(value, google.protobuf.message.Message) + and "header" in value.DESCRIPTOR.fields_by_name + ): + # System Nexus starts with generated models, so headers can only be + # applied after conversion to a request protobuf and before encoding. + temporalio.common._apply_headers( + headers, getattr(value, "header").fields + ) transfer_type_values.append(value) return self._inner_payload_converter.to_payloads(transfer_type_values) diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index d68d51ee2..ddf223b68 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -12,8 +12,6 @@ from dataclasses import dataclass from typing import Any -from google.protobuf.message import Message - import temporalio.api.common.v1 import temporalio.common import temporalio.converter @@ -93,27 +91,31 @@ class _SystemNexusPayloadConverter(temporalio.converter.PayloadConverter): """Payload converter for system Nexus outer envelopes.""" _user_converters: _SystemNexusUserConverters - _outer_payload_converter: temporalio.converter.PayloadConverter + _outer_payload_converter: _TemporalTransferTypePayloadConverter + _headers: Mapping[str, temporalio.api.common.v1.Payload] | None def __init__( self, user_payload_converter: temporalio.converter.PayloadConverter, user_failure_converter: temporalio.converter.FailureConverter, + headers: Mapping[str, temporalio.api.common.v1.Payload] | None = None, ) -> None: """Create a payload converter for system Nexus outer envelopes.""" self._user_converters = _SystemNexusUserConverters( user_payload_converter, user_failure_converter ) - self._outer_payload_converter = _TemporalTransferTypePayloadConverter.wrap( + + self._outer_payload_converter = _TemporalTransferTypePayloadConverter( _SystemNexusOuterPayloadConverter() ) + self._headers = headers def to_payloads( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: """See base class.""" with _user_converter_context(self._user_converters): - return self._outer_payload_converter.to_payloads(values) + return self._outer_payload_converter.to_payloads(values, self._headers) def from_payloads( self, @@ -134,23 +136,13 @@ 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) - - def _is_system_payload(payload: temporalio.api.common.v1.Payload) -> bool: return ( payload.metadata.get(_SYSTEM_PAYLOAD_METADATA_KEY) == _SYSTEM_PAYLOAD_METADATA_VALUE ) + + async def maybe_visit_payload( payload: temporalio.api.common.v1.Payload, visitor_functions: VisitorFunctions, @@ -175,9 +167,12 @@ async def maybe_visit_payload( def _get_payload_converter( # pyright: ignore[reportUnusedFunction] user_payload_converter: temporalio.converter.PayloadConverter, user_failure_converter: temporalio.converter.FailureConverter, + headers: Mapping[str, temporalio.api.common.v1.Payload] | None = None, ) -> temporalio.converter.PayloadConverter: """Return the fixed payload converter for system Nexus outer envelopes.""" - return _SystemNexusPayloadConverter(user_payload_converter, user_failure_converter) + return _SystemNexusPayloadConverter( + user_payload_converter, user_failure_converter, headers + ) def _get_serialization_context( # pyright: ignore[reportUnusedFunction] diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index 5be3f61e0..b82e9a0f0 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -2189,22 +2189,51 @@ async def operation_handle_fn() -> OutputT: 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, + nexus_input = 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, + ) + handle: _NexusOperationHandle[OutputT] + + async def operation_handle_fn() -> OutputT: + return cast( + OutputT, + await self._await_temporal_operation( + handle._result_fut, + lambda _err, command: handle._apply_cancel_command(command), + ), ) + + payload_converter = temporalio.nexus.system._get_payload_converter( + self._workflow_context_payload_converter, + self._workflow_context_failure_converter, + input.headers, ) + handle = _NexusOperationHandle( + self, + self._next_seq("nexus_operation"), + nexus_input, + operation_handle_fn(), + payload_converter, + ) + handle._apply_schedule_command() + self._pending_nexus_operations[handle._seq] = handle + + await self._await_temporal_operation( + handle._start_fut, + lambda _err, command: handle._apply_cancel_command(command), + reraise_on_workflow_cancellation=True, + ) + return handle #### Miscellaneous helpers #### # These are in alphabetical order. diff --git a/tests/contrib/opentelemetry/test_opentelemetry.py b/tests/contrib/opentelemetry/test_opentelemetry.py index 1bab931ac..55e652925 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry.py +++ b/tests/contrib/opentelemetry/test_opentelemetry.py @@ -88,6 +88,34 @@ class TracingWorkflowActionActivity: fail_on_non_replay_before_complete: bool = False +@workflow.defn +class LegacySignalWithStartHeaderWorkflow: + 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 LegacySignalWithStartCallerWorkflow: + @workflow.run + async def run(self, target_id: str, task_queue: str) -> str: + handle = await workflow.signal_with_start_workflow( + LegacySignalWithStartHeaderWorkflow.run, + id=target_id, + task_queue=task_queue, + signal=LegacySignalWithStartHeaderWorkflow.notify, + ) + return handle.id + + @dataclass class TracingWorkflowActionContinueAsNew: param: TracingWorkflowParam @@ -229,6 +257,38 @@ def update_validator(self) -> None: pass +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=[ + LegacySignalWithStartCallerWorkflow, + LegacySignalWithStartHeaderWorkflow, + ], + 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( + LegacySignalWithStartCallerWorkflow.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 + + async def test_opentelemetry_tracing(client: Client, env: WorkflowEnvironment): # TODO(cretz): Fix if env.supports_time_skipping: diff --git a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py index 12d0c5972..6cf8132ec 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py +++ b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py @@ -122,6 +122,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() @@ -169,6 +197,35 @@ 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 + + @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 c75fd64ff..d9f651f2f 100644 --- a/tests/nexus/test_temporal_system_nexus.py +++ b/tests/nexus/test_temporal_system_nexus.py @@ -36,6 +36,7 @@ from temporalio.worker import ( Interceptor, StartNexusOperationInput, + StartSystemNexusOperationInput, Worker, WorkflowInboundInterceptor, WorkflowInterceptorClassInput, @@ -253,6 +254,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] @@ -269,8 +276,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 = trace_input.input assert request.id == "system-nexus-workflow-id" assert request.signal == "test-signal" From be4814a52ea1972d87712c02fc0ea493005935a2 Mon Sep 17 00:00:00 2001 From: Tim Conley Date: Thu, 20 Aug 2026 10:45:49 -0700 Subject: [PATCH 5/5] Propagate tracing in system Nexus request headers --- scripts/nex_gen_support.py | 15 +++++ .../contrib/opentelemetry/_interceptor.py | 12 ++-- .../opentelemetry/_otel_interceptor.py | 3 +- temporalio/converter/_payload_converter.py | 14 +--- temporalio/nexus/system/__init__.py | 16 ++--- .../nexus/system/workflow_service/__init__.py | 2 +- .../workflow_service/_support/__init__.py | 2 +- .../_support/nex_gen_support.py | 15 +++++ .../nexus/system/workflow_service/models.py | 10 ++- .../workflow_service/operations/__init__.py | 2 +- .../operations/signal_with_start_workflow.py | 2 +- .../nexus/system/workflow_service/services.py | 2 +- temporalio/worker/_interceptor.py | 6 -- temporalio/worker/_workflow_instance.py | 65 +++++-------------- .../opentelemetry/test_opentelemetry.py | 2 + .../test_opentelemetry_plugin.py | 2 + 16 files changed, 80 insertions(+), 90 deletions(-) diff --git a/scripts/nex_gen_support.py b/scripts/nex_gen_support.py index ef6769fa1..fad21151e 100644 --- a/scripts/nex_gen_support.py +++ b/scripts/nex_gen_support.py @@ -167,6 +167,21 @@ def memo_to_proto( return message +def header_from_proto( + proto: common_pb2.Header, +) -> collections.abc.Mapping[str, object]: + return {key: _payload_to_value(value) for key, value in proto.fields.items()} + + +def header_to_proto( + header: collections.abc.Mapping[str, object], +) -> common_pb2.Header: + message = common_pb2.Header() + for key, value in header.items(): + message.fields[key].CopyFrom(_value_to_payload(value)) + return message + + def duration_from_proto(proto: google.protobuf.duration_pb2.Duration) -> timedelta: return proto.ToTimedelta() diff --git a/temporalio/contrib/opentelemetry/_interceptor.py b/temporalio/contrib/opentelemetry/_interceptor.py index 8bb8564f0..38321a2bc 100644 --- a/temporalio/contrib/opentelemetry/_interceptor.py +++ b/temporalio/contrib/opentelemetry/_interceptor.py @@ -833,11 +833,13 @@ async def start_nexus_operation( 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, - ) + if hasattr(input.input, "headers"): + input.input.headers = input.input.headers or {} + self.root._completed_span( + f"StartNexusOperation:{input.service}/{input.operation_name}", + kind=opentelemetry.trace.SpanKind.CLIENT, + add_to_outbound=cast(_InputWithHeaders, input.input), + ) return await super().start_system_nexus_operation(input) diff --git a/temporalio/contrib/opentelemetry/_otel_interceptor.py b/temporalio/contrib/opentelemetry/_otel_interceptor.py index 98875c849..457ff982d 100644 --- a/temporalio/contrib/opentelemetry/_otel_interceptor.py +++ b/temporalio/contrib/opentelemetry/_otel_interceptor.py @@ -608,5 +608,6 @@ async def start_system_nexus_operation( f"StartNexusOperation:{input.service}/{input.operation_name}", kind=opentelemetry.trace.SpanKind.CLIENT, ): - input.headers = _context_to_headers(input.headers) + if hasattr(input.input, "headers"): + input.input.headers = _context_to_headers(input.input.headers or {}) return await super().start_system_nexus_operation(input) diff --git a/temporalio/converter/_payload_converter.py b/temporalio/converter/_payload_converter.py index ac1a921a9..a8bc35e28 100644 --- a/temporalio/converter/_payload_converter.py +++ b/temporalio/converter/_payload_converter.py @@ -613,9 +613,7 @@ def wrap(payload_converter: PayloadConverter) -> PayloadConverter: return _TemporalTransferTypePayloadConverter(payload_converter) def to_payloads( - self, - values: Sequence[Any], - headers: Mapping[str, temporalio.api.common.v1.Payload] | None = None, + self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: """See base class.""" transfer_type_values: list[Any] = [] @@ -623,16 +621,6 @@ def to_payloads( converter = _get_transfer_type_converter(type(value)) if converter is not None: value = converter.to_transfer_type(value) - if ( - headers - and isinstance(value, google.protobuf.message.Message) - and "header" in value.DESCRIPTOR.fields_by_name - ): - # System Nexus starts with generated models, so headers can only be - # applied after conversion to a request protobuf and before encoding. - temporalio.common._apply_headers( - headers, getattr(value, "header").fields - ) transfer_type_values.append(value) return self._inner_payload_converter.to_payloads(transfer_type_values) diff --git a/temporalio/nexus/system/__init__.py b/temporalio/nexus/system/__init__.py index ddf223b68..60843aa1e 100644 --- a/temporalio/nexus/system/__init__.py +++ b/temporalio/nexus/system/__init__.py @@ -8,7 +8,7 @@ import contextlib import contextvars -from collections.abc import Iterator, Mapping, Sequence +from collections.abc import Iterator, Sequence from dataclasses import dataclass from typing import Any @@ -91,31 +91,28 @@ class _SystemNexusPayloadConverter(temporalio.converter.PayloadConverter): """Payload converter for system Nexus outer envelopes.""" _user_converters: _SystemNexusUserConverters - _outer_payload_converter: _TemporalTransferTypePayloadConverter - _headers: Mapping[str, temporalio.api.common.v1.Payload] | None + _outer_payload_converter: temporalio.converter.PayloadConverter def __init__( self, user_payload_converter: temporalio.converter.PayloadConverter, user_failure_converter: temporalio.converter.FailureConverter, - headers: Mapping[str, temporalio.api.common.v1.Payload] | None = None, ) -> None: """Create a payload converter for system Nexus outer envelopes.""" self._user_converters = _SystemNexusUserConverters( user_payload_converter, user_failure_converter ) - self._outer_payload_converter = _TemporalTransferTypePayloadConverter( + self._outer_payload_converter = _TemporalTransferTypePayloadConverter.wrap( _SystemNexusOuterPayloadConverter() ) - self._headers = headers def to_payloads( self, values: Sequence[Any] ) -> list[temporalio.api.common.v1.Payload]: """See base class.""" with _user_converter_context(self._user_converters): - return self._outer_payload_converter.to_payloads(values, self._headers) + return self._outer_payload_converter.to_payloads(values) def from_payloads( self, @@ -167,12 +164,9 @@ async def maybe_visit_payload( def _get_payload_converter( # pyright: ignore[reportUnusedFunction] user_payload_converter: temporalio.converter.PayloadConverter, user_failure_converter: temporalio.converter.FailureConverter, - headers: Mapping[str, temporalio.api.common.v1.Payload] | None = None, ) -> temporalio.converter.PayloadConverter: """Return the fixed payload converter for system Nexus outer envelopes.""" - return _SystemNexusPayloadConverter( - user_payload_converter, user_failure_converter, headers - ) + return _SystemNexusPayloadConverter(user_payload_converter, user_failure_converter) def _get_serialization_context( # pyright: ignore[reportUnusedFunction] diff --git a/temporalio/nexus/system/workflow_service/__init__.py b/temporalio/nexus/system/workflow_service/__init__.py index 092383b35..f757a41dc 100644 --- a/temporalio/nexus/system/workflow_service/__init__.py +++ b/temporalio/nexus/system/workflow_service/__init__.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_support/__init__.py b/temporalio/nexus/system/workflow_service/_support/__init__.py index 530c33e80..8fe5fb12a 100644 --- a/temporalio/nexus/system/workflow_service/_support/__init__.py +++ b/temporalio/nexus/system/workflow_service/_support/__init__.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py index ef6769fa1..fad21151e 100644 --- a/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py +++ b/temporalio/nexus/system/workflow_service/_support/nex_gen_support.py @@ -167,6 +167,21 @@ def memo_to_proto( return message +def header_from_proto( + proto: common_pb2.Header, +) -> collections.abc.Mapping[str, object]: + return {key: _payload_to_value(value) for key, value in proto.fields.items()} + + +def header_to_proto( + header: collections.abc.Mapping[str, object], +) -> common_pb2.Header: + message = common_pb2.Header() + for key, value in header.items(): + message.fields[key].CopyFrom(_value_to_payload(value)) + return message + + def duration_from_proto(proto: google.protobuf.duration_pb2.Duration) -> timedelta: return proto.ToTimedelta() diff --git a/temporalio/nexus/system/workflow_service/models.py b/temporalio/nexus/system/workflow_service/models.py index d03e7c3b4..653db044d 100644 --- a/temporalio/nexus/system/workflow_service/models.py +++ b/temporalio/nexus/system/workflow_service/models.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations @@ -17,6 +17,8 @@ from ._support import ( duration_from_proto, duration_to_proto, + header_from_proto, + header_to_proto, memo_from_proto, memo_to_proto, payload_from_proto, @@ -133,6 +135,9 @@ def from_transfer_type( if proto.HasField("user_metadata") else None, namespace=proto.namespace, + headers=header_from_proto(proto.header) + if proto.HasField("header") + else None, ) @typing_extensions.override @@ -191,6 +196,8 @@ def to_transfer_type( ) ) message.namespace = value.namespace + if value.headers is not None: + message.header.CopyFrom(header_to_proto(value.headers)) return message @@ -226,6 +233,7 @@ class SignalWithStartWorkflowRequest: start_delay: datetime.timedelta | None = None user_metadata: UserMetadata | None = None namespace: str = dataclasses.field(default_factory=workflow_namespace) + headers: collections.abc.Mapping[str, typing.Any] | None = None class _UserMetadataTransferTypeConverter( diff --git a/temporalio/nexus/system/workflow_service/operations/__init__.py b/temporalio/nexus/system/workflow_service/operations/__init__.py index 67c9cc56b..26c5cc532 100644 --- a/temporalio/nexus/system/workflow_service/operations/__init__.py +++ b/temporalio/nexus/system/workflow_service/operations/__init__.py @@ -1,3 +1,3 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py index ac75f9fb9..77782589b 100644 --- a/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py +++ b/temporalio/nexus/system/workflow_service/operations/signal_with_start_workflow.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/nexus/system/workflow_service/services.py b/temporalio/nexus/system/workflow_service/services.py index e2e901157..7b6c58f0e 100644 --- a/temporalio/nexus/system/workflow_service/services.py +++ b/temporalio/nexus/system/workflow_service/services.py @@ -1,4 +1,4 @@ -# Generated by nex-gen. DO NOT EDIT! +# Generated by nexgen. DO NOT EDIT! from __future__ import annotations diff --git a/temporalio/worker/_interceptor.py b/temporalio/worker/_interceptor.py index 76209ab56..b886a59b6 100644 --- a/temporalio/worker/_interceptor.py +++ b/temporalio/worker/_interceptor.py @@ -355,12 +355,6 @@ class StartSystemNexusOperationInput(Generic[InputT, OutputT]): 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 @property diff --git a/temporalio/worker/_workflow_instance.py b/temporalio/worker/_workflow_instance.py index b82e9a0f0..9e9a4f100 100644 --- a/temporalio/worker/_workflow_instance.py +++ b/temporalio/worker/_workflow_instance.py @@ -1726,18 +1726,15 @@ async def workflow_start_nexus_operation( summary: str | None, ) -> temporalio.workflow.NexusOperationHandle[OutputT]: if temporalio.nexus.system.is_system_endpoint(endpoint): + # System operations have no caller-configurable Nexus options. Do not + # expose the normal operation's scheduling, cancellation, headers, or + # summary arguments to System Nexus interceptors. 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( @@ -2189,51 +2186,23 @@ async def operation_handle_fn() -> OutputT: async def _outbound_start_system_nexus_operation( self, input: StartSystemNexusOperationInput[Any, OutputT] ) -> _NexusOperationHandle[OutputT]: - nexus_input = 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, - ) - handle: _NexusOperationHandle[OutputT] - - async def operation_handle_fn() -> OutputT: - return cast( - OutputT, - await self._await_temporal_operation( - handle._result_fut, - lambda _err, command: handle._apply_cancel_command(command), + 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=None, + schedule_to_start_timeout=None, + start_to_close_timeout=None, + cancellation_type=( + temporalio.workflow.NexusOperationCancellationType.WAIT_COMPLETED ), + headers=None, + summary=None, ) - - payload_converter = temporalio.nexus.system._get_payload_converter( - self._workflow_context_payload_converter, - self._workflow_context_failure_converter, - input.headers, - ) - handle = _NexusOperationHandle( - self, - self._next_seq("nexus_operation"), - nexus_input, - operation_handle_fn(), - payload_converter, ) - handle._apply_schedule_command() - self._pending_nexus_operations[handle._seq] = handle - - await self._await_temporal_operation( - handle._start_fut, - lambda _err, command: handle._apply_cancel_command(command), - reraise_on_workflow_cancellation=True, - ) - return handle #### Miscellaneous helpers #### # These are in alphabetical order. diff --git a/tests/contrib/opentelemetry/test_opentelemetry.py b/tests/contrib/opentelemetry/test_opentelemetry.py index 55e652925..3253f4f2b 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry.py +++ b/tests/contrib/opentelemetry/test_opentelemetry.py @@ -257,6 +257,8 @@ def update_validator(self) -> None: pass +# Cloud namespaces created by CI do not have the System Nexus dynamic config. +@pytest.mark.requires_local_server async def test_legacy_otel_workflow_signal_with_start_propagates_trace_headers( client: Client, env: WorkflowEnvironment ): diff --git a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py index 6cf8132ec..8d922aeaa 100644 --- a/tests/contrib/opentelemetry/test_opentelemetry_plugin.py +++ b/tests/contrib/opentelemetry/test_opentelemetry_plugin.py @@ -197,6 +197,8 @@ async def test_otel_tracing_basic(client: Client, reset_otel_tracer_provider: An ) +# Cloud namespaces created by CI do not have the System Nexus dynamic config. +@pytest.mark.requires_local_server async def test_otel_workflow_signal_with_start_propagates_trace_headers( client: Client, env: WorkflowEnvironment,