Fix System Nexus tracing headers

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
 

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()
 

temporalio/contrib/opentelemetry/_interceptor.py

@@ -830,6 +830,18 @@ 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]:
+        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)
+
 
 def _carrier_to_nexus_headers(
     carrier: _CarrierDict, initial: Mapping[str, str] | None = None

temporalio/contrib/opentelemetry/_otel_interceptor.py

@@ -600,3 +600,14 @@ 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,
+        ):
+            if hasattr(input.input, "headers"):
+                input.input.headers = _context_to_headers(input.input.headers or {})
+            return await super().start_system_nexus_operation(input)

temporalio/nexus/system/__init__.py

@@ -8,7 +8,7 @@
 
 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
 
@@ -102,6 +102,7 @@
         self._user_converters = _SystemNexusUserConverters(
             user_payload_converter, user_failure_converter
         )
+
         self._outer_payload_converter = _TemporalTransferTypePayloadConverter.wrap(
             _SystemNexusOuterPayloadConverter()
         )

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
 

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
 

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()
 

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(

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

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
 

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
 

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,29 @@ 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
+    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 +504,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,
 )
@@ -1724,7 +1725,18 @@ 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):
+            # 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,
+                )
+            )
         return await self._outbound.start_nexus_operation(
             StartNexusOperationInput(
                 endpoint=endpoint,
@@ -2171,6 +2183,27 @@ async def operation_handle_fn() -> OutputT:
         )
         return handle
 
+    async def _outbound_start_system_nexus_operation(
+        self, input: StartSystemNexusOperationInput[Any, OutputT]
+    ) -> _NexusOperationHandle[OutputT]:
+        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,
+            )
+        )
+
     #### Miscellaneous helpers ####
     # These are in alphabetical order.
 
@@ -3157,6 +3190,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.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: