From d25b184a138ee6d6d666a2c5f773c734411bf03b Mon Sep 17 00:00:00 2001 From: DABH Date: Mon, 14 Sep 2026 12:33:04 -0500 Subject: [PATCH] Stop starting replica traces in the OpenAI Agents OTel interceptor With use_otel_instrumentation=True the receiving side recreated the caller's Agents SDK trace and span and started them, so OpenInference registered never-finished copies under the caller's IDs. In a shared process these 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. Restore the trace and span without starting them and attach the propagated OTel span context as current so receiving-side spans parent to the caller's span. Remove the now unused id seeding and start_traces flag. Fixes #1852 --- CHANGELOG.md | 6 + .../openai_agents/_otel_trace_interceptor.py | 65 ++++----- .../openai_agents/_temporal_openai_agents.py | 1 - .../openai_agents/_trace_interceptor.py | 15 +- .../openai_agents/test_openai_tracing.py | 128 ++++++++++++++---- 5 files changed, 139 insertions(+), 76 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f1a526e7e..3c2134de9 100644 --- a/CHANGELOG.md +++ b/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 diff --git a/temporalio/contrib/openai_agents/_otel_trace_interceptor.py b/temporalio/contrib/openai_agents/_otel_trace_interceptor.py index 63f8f9d83..1218c0719 100644 --- a/temporalio/contrib/openai_agents/_otel_trace_interceptor.py +++ b/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) diff --git a/temporalio/contrib/openai_agents/_temporal_openai_agents.py b/temporalio/contrib/openai_agents/_temporal_openai_agents.py index 6023ad090..3931ca211 100644 --- a/temporalio/contrib/openai_agents/_temporal_openai_agents.py +++ b/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 diff --git a/temporalio/contrib/openai_agents/_trace_interceptor.py b/temporalio/contrib/openai_agents/_trace_interceptor.py index 66297e20b..04a056722 100644 --- a/temporalio/contrib/openai_agents/_trace_interceptor.py +++ b/temporalio/contrib/openai_agents/_trace_interceptor.py @@ -84,7 +84,6 @@ 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. @@ -92,12 +91,9 @@ def __init__( 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, diff --git a/tests/contrib/openai_agents/test_openai_tracing.py b/tests/contrib/openai_agents/test_openai_tracing.py index 28b804cc1..8e8508058 100644 --- a/tests/contrib/openai_agents/test_openai_tracing.py +++ b/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"