Fix System Nexus tracing headers

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

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

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)

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(  # pyright: ignore[reportUnusedFunction]
+    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,

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",

temporalio/worker/_interceptor.py

@@ -348,6 +348,35 @@ 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
+
+    @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 +510,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:

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]:

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:

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