Stop starting replica traces in the OpenAI Agents OTel interceptor

CHANGELOG.md

@@ -86,6 +86,12 @@ to include examples, links to docs, or any other relevant information.
 - Nexus-context workflow/activity starts no longer set `on_conflict_options` when there are no links
   or callbacks to attach.
 - The workflow sandbox now passes `pydantic_core` through by default, alongside `pydantic`.
+- `OpenAIAgentsPlugin(use_otel_instrumentation=True)`: spans started in workflows and
+  activities now parent directly to the caller's OpenTelemetry span instead of to a copy of
+  the caller's Agents SDK trace and span started on the worker. The copies were never finished
+  and, when client and worker shared a process, displaced the caller's own spans, leaving
+  `temporal:startWorkflow` and `temporal:startActivity` spans with a parent that was never
+  exported and dropping the client's root span.
 
 ### Security
 

temporalio/contrib/openai_agents/_otel_trace_interceptor.py

@@ -5,10 +5,14 @@
 from typing import Any
 
 import opentelemetry.trace
+from opentelemetry.context import attach
+from opentelemetry.trace import (
+    NonRecordingSpan,
+    SpanContext,
+    TraceFlags,
+    set_span_in_context,
+)
 
-import temporalio.converter
-
-from ..opentelemetry._id_generator import TemporalIdGenerator
 from ._trace_interceptor import (
     OpenAIAgentsContextPropagationInterceptor,
     _InputWithHeaders,
@@ -20,22 +24,6 @@ class OTelOpenAIAgentsContextPropagationInterceptor(
 ):
     """OTEL-aware variant that enhances headers with OpenTelemetry span context."""
 
-    def __init__(
-        self,
-        otel_id_generator: TemporalIdGenerator,
-        payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter,
-        add_temporal_spans: bool = True,
-    ) -> None:
-        """Initialize OTEL-aware context propagation interceptor.
-
-        Args:
-            otel_id_generator: Generator for OTEL-compatible IDs.
-            payload_converter: Converter for serializing trace context.
-            add_temporal_spans: Whether to add Temporal-specific spans.
-        """
-        super().__init__(payload_converter, add_temporal_spans, start_traces=True)
-        self._otel_id_generator = otel_id_generator
-
     def header_contents(self) -> dict[str, Any]:
         """Get header contents enhanced with OpenTelemetry span context.
 
@@ -66,23 +54,22 @@ def context_from_header(
         otel_span_id = span_info.get("otelSpanId")
         otel_trace_id = span_info.get("otelTraceId")
 
-        # Seed the trace id before the trace is reconstructed so the workflow's root
-        # OTEL span shares the caller's trace id rather than generating a new one.
-        if otel_trace_id and self._otel_id_generator:
-            self._otel_id_generator.seed_trace_id(otel_trace_id)
-
-        # If only a trace was propagated from the caller, we need to seed for trace context
-        if otel_span_id and self._otel_id_generator and span_info.get("spanId") is None:
-            self._otel_id_generator.seed_span_id(otel_span_id)
-
-        super().trace_context_from_header_contents(span_info)
-
-        # If a span was propagated from the caller, we need to seed for span context
-        if (
-            otel_span_id
-            and self._otel_id_generator
-            and span_info.get("spanId") is not None
-        ):
-            self._otel_id_generator.seed_span_id(otel_span_id)
-
-        super().span_context_from_header_contents(span_info)
+        # Parent OTEL spans started here to the caller's span. The Agents SDK trace
+        # and span restored below are not started, so OpenInference never registers
+        # copies of them under the caller's IDs.
+        if otel_span_id and otel_trace_id:
+            attach(
+                set_span_in_context(
+                    NonRecordingSpan(
+                        SpanContext(
+                            trace_id=otel_trace_id,
+                            span_id=otel_span_id,
+                            is_remote=True,
+                            trace_flags=TraceFlags(TraceFlags.SAMPLED),
+                        )
+                    )
+                )
+            )
+
+        self.trace_context_from_header_contents(span_info)
+        self.span_context_from_header_contents(span_info)

temporalio/contrib/openai_agents/_temporal_openai_agents.py

@@ -426,7 +426,6 @@ def workflow_runner(runner: WorkflowRunner | None) -> WorkflowRunner:
 
             interceptor = OTelOpenAIAgentsContextPropagationInterceptor(
                 add_temporal_spans=add_temporal_spans,
-                otel_id_generator=provider.id_generator(),
             )
 
         @asynccontextmanager

temporalio/contrib/openai_agents/_trace_interceptor.py

@@ -84,20 +84,16 @@ def __init__(
         self,
         payload_converter: temporalio.converter.PayloadConverter = temporalio.converter.default().payload_converter,
         add_temporal_spans: bool = True,
-        start_traces: bool = False,
     ) -> None:
         """Initialize the interceptor with a payload converter.
 
         Args:
             payload_converter: The payload converter to use for serializing/deserializing
                 trace context. Defaults to the default Temporal payload converter.
             add_temporal_spans: Whether to add temporal-specific spans to traces.
-            start_traces: Whether to start new traces if none exist. This will cause duplication if the underlying
-                trace provider actually process start events. Primarily designed for use with Open Telemetry integration.
         """
         super().__init__()
         self._payload_converter = payload_converter
-        self._start_traces = start_traces
         self._add_temporal_spans = add_temporal_spans
 
     def intercept_client(
@@ -188,11 +184,7 @@ def trace_context_from_header_contents(self, span_info: dict[str, Any]):
                 span_info["traceName"],
                 trace_id=span_info["traceId"],
             )
-
-            if self._start_traces:
-                current_trace.start(mark_as_current=True)
-            else:
-                Scope.set_current_trace(current_trace)
+            Scope.set_current_trace(current_trace)
 
     def span_context_from_header_contents(self, span_info: dict[str, Any]):
         """Initialize span context from header contents.
@@ -205,10 +197,7 @@ def span_context_from_header_contents(self, span_info: dict[str, Any]):
             current_span = get_trace_provider().create_span(
                 span_data=CustomSpanData(name="", data={}), span_id=span_info["spanId"]
             )
-            if self._start_traces:
-                current_span.start(mark_as_current=True)
-            else:
-                Scope.set_current_span(current_span)
+            Scope.set_current_span(current_span)
 
     def context_from_header(
         self,

tests/contrib/openai_agents/test_openai_tracing.py

@@ -3,7 +3,7 @@
 from typing import Any
 
 import opentelemetry.trace
-from agents import Span, Trace, TracingProcessor, custom_span, trace
+from agents import Agent, Runner, Span, Trace, TracingProcessor, custom_span, trace
 from agents.tracing import get_trace_provider
 from opentelemetry.sdk.trace import ReadableSpan
 from opentelemetry.sdk.trace.export import SimpleSpanProcessor
@@ -14,7 +14,10 @@
 from temporalio.contrib.openai_agents import _temporal_openai_agents
 from temporalio.contrib.openai_agents.testing import (
     AgentEnvironment,
+    ResponseBuilders,
+    TestModel,
 )
+from temporalio.contrib.openai_agents.workflow import activity_as_tool
 from temporalio.contrib.opentelemetry import create_tracer_provider
 from temporalio.worker.workflow_sandbox import (
     SandboxedWorkflowRunner,
@@ -677,15 +680,18 @@ async def test_otel_tracing_in_runner(
             ResearchWorkflow,
             max_cached_workflows=0,
         ) as worker:
-            with trace("Research workflow"):
-                workflow_handle = await client.start_workflow(
-                    ResearchWorkflow.run,
-                    "Caribbean vacation spots in April, optimizing for surfing, hiking and water sports",
-                    id=f"research-workflow-{uuid.uuid4()}",
-                    task_queue=worker.task_queue,
-                    execution_timeout=timedelta(seconds=120),
-                )
-                await workflow_handle.result()
+            # The worker instruments on its run task, which has not run yet; without
+            # this the client's trace would get no OTEL root span.
+            with env.openai_agents_plugin.tracing_context():
+                with trace("Research workflow"):
+                    workflow_handle = await client.start_workflow(
+                        ResearchWorkflow.run,
+                        "Caribbean vacation spots in April, optimizing for surfing, hiking and water sports",
+                        id=f"research-workflow-{uuid.uuid4()}",
+                        task_queue=worker.task_queue,
+                        execution_timeout=timedelta(seconds=120),
+                    )
+                    await workflow_handle.result()
 
     spans = exporter.get_finished_spans()
     print("OTEL tracing in runner spans:")
@@ -858,21 +864,23 @@ async def test_sdk_trace_to_otel_span_parenting(
                 SandboxRestrictions.default.with_passthrough_modules("opentelemetry")
             ),
         ) as worker:
-            # Start SDK trace in client, then start workflow within that trace
-            with trace("Client SDK trace"):
-                workflow_handle = await new_client.start_workflow(
-                    OtelSpanWorkflow.run,
-                    id=f"sdk-trace-otel-span-workflow-{uuid.uuid4()}",
-                    task_queue=worker.task_queue,
-                    execution_timeout=timedelta(seconds=120),
-                )
-                workflow_id = workflow_handle.id
+            # The worker instruments on its run task, which has not run yet; without
+            # this the client's trace would get no OTEL root span.
+            with env.openai_agents_plugin.tracing_context():
+                with trace("Client SDK trace"):
+                    workflow_handle = await new_client.start_workflow(
+                        OtelSpanWorkflow.run,
+                        id=f"sdk-trace-otel-span-workflow-{uuid.uuid4()}",
+                        task_queue=worker.task_queue,
+                        execution_timeout=timedelta(seconds=120),
+                    )
+                    workflow_id = workflow_handle.id
 
-                # Wait for workflow to be ready
-                async def ready() -> bool:
-                    return await workflow_handle.query(OtelSpanWorkflow.ready)
+                    # Wait for workflow to be ready
+                    async def ready() -> bool:
+                        return await workflow_handle.query(OtelSpanWorkflow.ready)
 
-                await assert_eq_eventually(True, ready)
+                    await assert_eq_eventually(True, ready)
 
     # Second worker: Complete the workflow with fresh objects (new instrumentation)
     async with AgentEnvironment(
@@ -953,3 +961,77 @@ async def ready() -> bool:
     assert len(span_ids) == len(set(span_ids)), (
         f"All spans should have unique IDs, got: {span_ids}"
     )
+
+
+@activity.defn
+async def lookup_activity(query: str) -> str:
+    return f"result for {query}"
+
+
+@workflow.defn
+class ActivityToolWorkflow:
+    @workflow.run
+    async def run(self) -> str:
+        agent = Agent[str](
+            name="Tool agent",
+            instructions="Use the tool.",
+            tools=[
+                activity_as_tool(
+                    lookup_activity, start_to_close_timeout=timedelta(seconds=10)
+                )
+            ],
+        )
+        return str((await Runner.run(agent, input="look up x")).final_output)
+
+
+async def test_otel_spans_single_process_parents_exported(
+    client: Client,
+    reset_otel_tracer_provider: Any,  # type: ignore[reportUnusedParameter]
+):
+    """Client, workflow and activity in one process with Temporal spans enabled.
+
+    Every exported span must have an exported parent, and the client's root span
+    must be the one exported (not a replica recreated on the worker side).
+    """
+    exporter = set_test_tracer_provider()
+
+    async with AgentEnvironment(
+        model=TestModel.returning_responses(
+            [
+                ResponseBuilders.tool_call('{"query":"x"}', "lookup_activity"),
+                ResponseBuilders.output_message("done"),
+            ]
+        ),
+        use_otel_instrumentation=True,
+    ) as env:
+        client = env.applied_on_client(client)
+
+        async with new_worker(
+            client, ActivityToolWorkflow, activities=[lookup_activity]
+        ) as worker:
+            with env.openai_agents_plugin.tracing_context():
+                with trace("Client trace"):
+                    root = opentelemetry.trace.get_current_span()
+                    root.set_attribute("test.marker", "client root")
+                    result = await client.execute_workflow(
+                        ActivityToolWorkflow.run,
+                        id=f"otel-single-process-{uuid.uuid4()}",
+                        task_queue=worker.task_queue,
+                        execution_timeout=timedelta(seconds=120),
+                    )
+    assert result == "done"
+
+    spans = exporter.get_finished_spans()
+    print_otel_spans(spans)
+    by_id = {span.context.span_id: span for span in spans if span.context}
+
+    assert "temporal:executeActivity" in {span.name for span in spans}
+    assert [
+        span.name for span in spans if span.parent and span.parent.span_id not in by_id
+    ] == [], "spans whose parent was never exported"
+
+    exported_root = by_id.get(root.get_span_context().span_id)
+    assert exported_root is not None, "client root span was not exported"
+    assert exported_root.parent is None
+    assert exported_root.attributes is not None
+    assert exported_root.attributes["test.marker"] == "client root"