From 4114ac33c313cfd48c547e491b165843c3975a5d Mon Sep 17 00:00:00 2001 From: Wrisa Date: Tue, 30 Jun 2026 12:03:39 -0700 Subject: [PATCH 1/3] Add conversation root --- .../tests/test_callback_handler.py | 96 +++++++++++++++++++ .../util/genai/_agent_invocation.py | 6 ++ .../opentelemetry/util/genai/_invocation.py | 27 +++++- .../util/genai/_workflow_invocation.py | 10 +- .../tests/test_handler_agent.py | 61 ++++++++++++ .../tests/test_handler_workflow.py | 47 +++++++++ 6 files changed, 245 insertions(+), 2 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py index be5476e47..e633cc033 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py @@ -1015,3 +1015,99 @@ def test_agent_output_messages_only_last_ai_message(self): assigned = agent_inv.output_messages assert len(assigned) == 1 assert assigned[0].parts[0].content == "the answer is 7" + + +# --------------------------------------------------------------------------- +# conversation_root integration tests (real TelemetryHandler + real spans) +# --------------------------------------------------------------------------- + + +def _make_real_handler(): + """Return a handler wired to a real TelemetryHandler with an in-memory exporter.""" + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( + InMemorySpanExporter, + ) + from opentelemetry.util.genai.handler import TelemetryHandler + + exporter = InMemorySpanExporter() + tracer_provider = TracerProvider() + tracer_provider.add_span_processor(SimpleSpanProcessor(exporter)) + telemetry = TelemetryHandler(tracer_provider=tracer_provider) + handler = OpenTelemetryLangChainCallbackHandler(telemetry) + return handler, exporter + + +class TestConversationRootIntegration: + """Integration tests that verify gen_ai.conversation_root is emitted correctly.""" + + def test_root_agent_has_conversation_root_on_span(self): + """Root agent (parent_run_id=None) gets gen_ai.conversation_root=True.""" + handler, exporter = _make_real_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "travel_planner"}, + inputs={}, + run_id=run_id, + parent_run_id=None, + metadata={"agent_name": "Travel Planner"}, + ) + handler.on_chain_end(outputs={}, run_id=run_id) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].attributes.get("gen_ai.conversation_root") is True + + def test_child_agent_lacks_conversation_root_on_span(self): + """Child agent (has a parent invocation) does NOT get gen_ai.conversation_root.""" + handler, exporter = _make_real_handler() + parent_run_id = _run_id() + child_run_id = _run_id() + + # Start root agent + handler.on_chain_start( + serialized={"name": "travel_planner"}, + inputs={}, + run_id=parent_run_id, + parent_run_id=None, + metadata={"agent_name": "Travel Planner"}, + ) + # Start child agent nested inside the root + handler.on_chain_start( + serialized={"name": "flight_search"}, + inputs={}, + run_id=child_run_id, + parent_run_id=parent_run_id, + metadata={"agent_name": "Flight Search Agent"}, + ) + handler.on_chain_end(outputs={}, run_id=child_run_id) + handler.on_chain_end(outputs={}, run_id=parent_run_id) + + spans = exporter.get_finished_spans() + root_span = next( + s for s in spans if "Travel Planner" in s.name + ) + child_span = next( + s for s in spans if "Flight Search Agent" in s.name + ) + assert root_span.attributes.get("gen_ai.conversation_root") is True + assert "gen_ai.conversation_root" not in child_span.attributes + + def test_root_workflow_has_conversation_root_on_span(self): + """Root workflow (LangGraph, parent_run_id=None) gets gen_ai.conversation_root=True.""" + handler, exporter = _make_real_handler() + run_id = _run_id() + + handler.on_chain_start( + serialized={"name": "LangGraph", "id": ["langgraph"]}, + inputs={}, + run_id=run_id, + parent_run_id=None, + ) + handler.on_chain_end(outputs={}, run_id=run_id) + + spans = exporter.get_finished_spans() + assert len(spans) == 1 + assert spans[0].attributes.get("gen_ai.conversation_root") is True diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py index eebb05c5a..78e21c69f 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py @@ -14,6 +14,7 @@ from opentelemetry.util.genai._invocation import ( Error, GenAIInvocation, + _has_active_span, get_content_attributes, ) from opentelemetry.util.genai.completion_hook import CompletionHook @@ -89,6 +90,11 @@ def __init__( self.finish_reasons: list[str] | None = None + # Auto-mark as conversation root when no active span exists at creation time. + # Can be overridden by the caller after construction. + if not _has_active_span(): + self.conversation_root = True + self.input_tokens: int | None = None self.output_tokens: int | None = None self.cache_creation_input_tokens: int | None = None diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py index 4b1d7e662..5e079a076 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py @@ -18,7 +18,14 @@ ) from opentelemetry.semconv.attributes import error_attributes from opentelemetry.trace import INVALID_SPAN as _INVALID_SPAN -from opentelemetry.trace import Span, SpanKind, Tracer, set_span_in_context +from opentelemetry.trace import ( + NonRecordingSpan, + Span, + SpanKind, + Tracer, + get_current_span, + set_span_in_context, +) from opentelemetry.trace.status import Status, StatusCode from opentelemetry.util.genai.completion_hook import CompletionHook from opentelemetry.util.genai.types import ( @@ -41,6 +48,16 @@ ContextToken: TypeAlias = Token[Context] +def _has_active_span() -> bool: + """Return True if there is an active recording span in the current context. + + Used by WorkflowInvocation and AgentInvocation to auto-detect whether they + are the root of a conversation (no active parent span). + """ + current = get_current_span() + return not isinstance(current, NonRecordingSpan) and current.is_recording() + + class GenAIInvocation(AbstractContextManager["GenAIInvocation"]): """ Base class for all GenAI invocation types. Manages the lifecycle of a single @@ -83,6 +100,10 @@ def __init__( self._span_kind: SpanKind = span_kind self._context_token: ContextToken | None = None self._monotonic_start_s: float | None = None + self.conversation_root: bool | None = None + """When True, marks this span as the root of a conversation (no GenAI parent). + Auto-set by WorkflowInvocation and AgentInvocation when no active span exists + at construction time. Can be explicitly set by callers to override.""" def _start(self, attributes: dict[str, Any] | None = None) -> None: """Start the invocation span and attach it to the current context. @@ -147,6 +168,10 @@ def _finish(self, error: Error | None = None) -> None: return try: self._apply_finish(error) + if self.conversation_root is not None: + self.span.set_attribute( + "gen_ai.conversation_root", self.conversation_root + ) finally: try: detach(self._context_token) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py index e7d6ff7e4..b6cf7832b 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py @@ -11,7 +11,11 @@ gen_ai_attributes as GenAI, ) from opentelemetry.trace import SpanKind, Tracer -from opentelemetry.util.genai._invocation import Error, GenAIInvocation +from opentelemetry.util.genai._invocation import ( + Error, + GenAIInvocation, + _has_active_span, +) from opentelemetry.util.genai.completion_hook import CompletionHook from opentelemetry.util.genai.metrics import InvocationMetricsRecorder from opentelemetry.util.genai.types import ( @@ -55,6 +59,10 @@ def __init__( self.name = name self.input_messages: list[InputMessage] = [] self.output_messages: list[OutputMessage] = [] + # Auto-mark as conversation root when no active span exists at creation time. + # Can be overridden by the caller after construction. + if not _has_active_span(): + self.conversation_root = True self._start(self._get_base_attributes()) def _get_base_attributes(self) -> dict[str, Any]: diff --git a/util/opentelemetry-util-genai/tests/test_handler_agent.py b/util/opentelemetry-util-genai/tests/test_handler_agent.py index 9075b291e..5de776cd9 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_agent.py +++ b/util/opentelemetry-util-genai/tests/test_handler_agent.py @@ -544,6 +544,67 @@ def get_description(self): assert captured_attributes[server_attributes.SERVER_PORT] == 8080 +class TestAgentConversationRoot(unittest.TestCase): + def setUp(self): + self.span_exporter = InMemorySpanExporter() + tracer_provider = TracerProvider() + tracer_provider.add_span_processor( + SimpleSpanProcessor(self.span_exporter) + ) + self.handler = TelemetryHandler(tracer_provider=tracer_provider) + + def test_root_agent_gets_conversation_root_true(self): + """Agent with no active parent span is auto-marked as conversation root.""" + invocation = self.handler.invoke_local_agent("openai", agent_name="Root Agent") + self.assertTrue(invocation.conversation_root) + invocation.stop() + + spans = self.span_exporter.get_finished_spans() + self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), True) + + def test_child_agent_does_not_get_conversation_root(self): + """Agent created inside an active parent span is NOT marked as root.""" + with self.handler.invoke_local_agent("openai", agent_name="Parent Agent") as _parent: + child = self.handler.invoke_local_agent("openai", agent_name="Child Agent") + self.assertIsNone(child.conversation_root) + child.stop() + + spans = self.span_exporter.get_finished_spans() + child_span = next(s for s in spans if s.name == "invoke_agent Child Agent") + self.assertNotIn("gen_ai.conversation_root", child_span.attributes) + + def test_explicit_conversation_root_false_is_respected(self): + """Explicitly setting conversation_root=False is not overridden.""" + invocation = self.handler.invoke_local_agent("openai") + invocation.conversation_root = False + invocation.stop() + + spans = self.span_exporter.get_finished_spans() + self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), False) + + def test_conversation_root_not_emitted_when_none(self): + """When conversation_root is None (child), attribute is not emitted.""" + with self.handler.invoke_local_agent("openai", agent_name="Parent") as _parent: + child = self.handler.invoke_local_agent("openai", agent_name="Child") + self.assertIsNone(child.conversation_root) + child.stop() + + spans = self.span_exporter.get_finished_spans() + child_span = next(s for s in spans if s.name == "invoke_agent Child") + self.assertNotIn("gen_ai.conversation_root", child_span.attributes) + + def test_remote_root_agent_gets_conversation_root_true(self): + """Remote agent with no active parent span is auto-marked as conversation root.""" + invocation = self.handler.invoke_remote_agent( + "openai", agent_name="Remote Root" + ) + self.assertTrue(invocation.conversation_root) + invocation.stop() + + spans = self.span_exporter.get_finished_spans() + self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), True) + + class TestAgentInvocationMetrics(TestBase): def test_local_agent_records_duration_and_tokens(self) -> None: handler = TelemetryHandler( diff --git a/util/opentelemetry-util-genai/tests/test_handler_workflow.py b/util/opentelemetry-util-genai/tests/test_handler_workflow.py index a304f5b82..c6e08df94 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_workflow.py +++ b/util/opentelemetry-util-genai/tests/test_handler_workflow.py @@ -266,3 +266,50 @@ def test_workflow_context_manager_with_messages(self) -> None: spans[0].attributes[GenAI.GEN_AI_OPERATION_NAME], "invoke_workflow", ) + + +class TelemetryHandlerWorkflowConversationRootTest(_WorkflowTestBase): + # ------------------------------------------------------------------ + # conversation_root auto-detection + # ------------------------------------------------------------------ + + def test_root_workflow_gets_conversation_root_true(self) -> None: + """Workflow with no active parent span is auto-marked as conversation root.""" + invocation = self.handler.workflow(name="root_wf") + self.assertTrue(invocation.conversation_root) + invocation.stop() + + spans = self._get_finished_spans() + self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), True) + + def test_child_workflow_does_not_get_conversation_root(self) -> None: + """Workflow created inside an active parent span is NOT marked as root.""" + with self.handler.workflow(name="parent_wf") as parent: + child = self.handler.workflow(name="child_wf") + self.assertIsNone(child.conversation_root) + child.stop() + + spans = self._get_finished_spans() + child_span = next(s for s in spans if s.name == "invoke_workflow child_wf") + self.assertNotIn("gen_ai.conversation_root", child_span.attributes) + + def test_explicit_conversation_root_false_is_respected(self) -> None: + """Explicitly setting conversation_root=False is not overridden.""" + invocation = self.handler.workflow(name="wf") + invocation.conversation_root = False + invocation.stop() + + spans = self._get_finished_spans() + self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), False) + + def test_conversation_root_not_emitted_when_none(self) -> None: + """When conversation_root is None, attribute is not emitted on the span.""" + with self.handler.workflow(name="parent") as _parent: + child = self.handler.workflow(name="child") + # conversation_root is None for child (has active parent) + self.assertIsNone(child.conversation_root) + child.stop() + + spans = self._get_finished_spans() + child_span = next(s for s in spans if s.name == "invoke_workflow child") + self.assertNotIn("gen_ai.conversation_root", child_span.attributes) From 7d611ba53fe1177dfea81aaa280af02e6dfd9a12 Mon Sep 17 00:00:00 2001 From: wrisa Date: Wed, 1 Jul 2026 05:38:47 -0700 Subject: [PATCH 2/3] Revert "Add conversation root" This reverts commit 4114ac33c313cfd48c547e491b165843c3975a5d. --- .../tests/test_callback_handler.py | 96 ------------------- .../util/genai/_agent_invocation.py | 6 -- .../opentelemetry/util/genai/_invocation.py | 27 +----- .../util/genai/_workflow_invocation.py | 10 +- .../tests/test_handler_agent.py | 61 ------------ .../tests/test_handler_workflow.py | 47 --------- 6 files changed, 2 insertions(+), 245 deletions(-) diff --git a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py index e633cc033..be5476e47 100644 --- a/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py +++ b/instrumentation/opentelemetry-instrumentation-genai-langchain/tests/test_callback_handler.py @@ -1015,99 +1015,3 @@ def test_agent_output_messages_only_last_ai_message(self): assigned = agent_inv.output_messages assert len(assigned) == 1 assert assigned[0].parts[0].content == "the answer is 7" - - -# --------------------------------------------------------------------------- -# conversation_root integration tests (real TelemetryHandler + real spans) -# --------------------------------------------------------------------------- - - -def _make_real_handler(): - """Return a handler wired to a real TelemetryHandler with an in-memory exporter.""" - from opentelemetry.sdk.trace import TracerProvider - from opentelemetry.sdk.trace.export import SimpleSpanProcessor - from opentelemetry.sdk.trace.export.in_memory_span_exporter import ( - InMemorySpanExporter, - ) - from opentelemetry.util.genai.handler import TelemetryHandler - - exporter = InMemorySpanExporter() - tracer_provider = TracerProvider() - tracer_provider.add_span_processor(SimpleSpanProcessor(exporter)) - telemetry = TelemetryHandler(tracer_provider=tracer_provider) - handler = OpenTelemetryLangChainCallbackHandler(telemetry) - return handler, exporter - - -class TestConversationRootIntegration: - """Integration tests that verify gen_ai.conversation_root is emitted correctly.""" - - def test_root_agent_has_conversation_root_on_span(self): - """Root agent (parent_run_id=None) gets gen_ai.conversation_root=True.""" - handler, exporter = _make_real_handler() - run_id = _run_id() - - handler.on_chain_start( - serialized={"name": "travel_planner"}, - inputs={}, - run_id=run_id, - parent_run_id=None, - metadata={"agent_name": "Travel Planner"}, - ) - handler.on_chain_end(outputs={}, run_id=run_id) - - spans = exporter.get_finished_spans() - assert len(spans) == 1 - assert spans[0].attributes.get("gen_ai.conversation_root") is True - - def test_child_agent_lacks_conversation_root_on_span(self): - """Child agent (has a parent invocation) does NOT get gen_ai.conversation_root.""" - handler, exporter = _make_real_handler() - parent_run_id = _run_id() - child_run_id = _run_id() - - # Start root agent - handler.on_chain_start( - serialized={"name": "travel_planner"}, - inputs={}, - run_id=parent_run_id, - parent_run_id=None, - metadata={"agent_name": "Travel Planner"}, - ) - # Start child agent nested inside the root - handler.on_chain_start( - serialized={"name": "flight_search"}, - inputs={}, - run_id=child_run_id, - parent_run_id=parent_run_id, - metadata={"agent_name": "Flight Search Agent"}, - ) - handler.on_chain_end(outputs={}, run_id=child_run_id) - handler.on_chain_end(outputs={}, run_id=parent_run_id) - - spans = exporter.get_finished_spans() - root_span = next( - s for s in spans if "Travel Planner" in s.name - ) - child_span = next( - s for s in spans if "Flight Search Agent" in s.name - ) - assert root_span.attributes.get("gen_ai.conversation_root") is True - assert "gen_ai.conversation_root" not in child_span.attributes - - def test_root_workflow_has_conversation_root_on_span(self): - """Root workflow (LangGraph, parent_run_id=None) gets gen_ai.conversation_root=True.""" - handler, exporter = _make_real_handler() - run_id = _run_id() - - handler.on_chain_start( - serialized={"name": "LangGraph", "id": ["langgraph"]}, - inputs={}, - run_id=run_id, - parent_run_id=None, - ) - handler.on_chain_end(outputs={}, run_id=run_id) - - spans = exporter.get_finished_spans() - assert len(spans) == 1 - assert spans[0].attributes.get("gen_ai.conversation_root") is True diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py index 78e21c69f..eebb05c5a 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py @@ -14,7 +14,6 @@ from opentelemetry.util.genai._invocation import ( Error, GenAIInvocation, - _has_active_span, get_content_attributes, ) from opentelemetry.util.genai.completion_hook import CompletionHook @@ -90,11 +89,6 @@ def __init__( self.finish_reasons: list[str] | None = None - # Auto-mark as conversation root when no active span exists at creation time. - # Can be overridden by the caller after construction. - if not _has_active_span(): - self.conversation_root = True - self.input_tokens: int | None = None self.output_tokens: int | None = None self.cache_creation_input_tokens: int | None = None diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py index 5e079a076..4b1d7e662 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py @@ -18,14 +18,7 @@ ) from opentelemetry.semconv.attributes import error_attributes from opentelemetry.trace import INVALID_SPAN as _INVALID_SPAN -from opentelemetry.trace import ( - NonRecordingSpan, - Span, - SpanKind, - Tracer, - get_current_span, - set_span_in_context, -) +from opentelemetry.trace import Span, SpanKind, Tracer, set_span_in_context from opentelemetry.trace.status import Status, StatusCode from opentelemetry.util.genai.completion_hook import CompletionHook from opentelemetry.util.genai.types import ( @@ -48,16 +41,6 @@ ContextToken: TypeAlias = Token[Context] -def _has_active_span() -> bool: - """Return True if there is an active recording span in the current context. - - Used by WorkflowInvocation and AgentInvocation to auto-detect whether they - are the root of a conversation (no active parent span). - """ - current = get_current_span() - return not isinstance(current, NonRecordingSpan) and current.is_recording() - - class GenAIInvocation(AbstractContextManager["GenAIInvocation"]): """ Base class for all GenAI invocation types. Manages the lifecycle of a single @@ -100,10 +83,6 @@ def __init__( self._span_kind: SpanKind = span_kind self._context_token: ContextToken | None = None self._monotonic_start_s: float | None = None - self.conversation_root: bool | None = None - """When True, marks this span as the root of a conversation (no GenAI parent). - Auto-set by WorkflowInvocation and AgentInvocation when no active span exists - at construction time. Can be explicitly set by callers to override.""" def _start(self, attributes: dict[str, Any] | None = None) -> None: """Start the invocation span and attach it to the current context. @@ -168,10 +147,6 @@ def _finish(self, error: Error | None = None) -> None: return try: self._apply_finish(error) - if self.conversation_root is not None: - self.span.set_attribute( - "gen_ai.conversation_root", self.conversation_root - ) finally: try: detach(self._context_token) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py index b6cf7832b..e7d6ff7e4 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py @@ -11,11 +11,7 @@ gen_ai_attributes as GenAI, ) from opentelemetry.trace import SpanKind, Tracer -from opentelemetry.util.genai._invocation import ( - Error, - GenAIInvocation, - _has_active_span, -) +from opentelemetry.util.genai._invocation import Error, GenAIInvocation from opentelemetry.util.genai.completion_hook import CompletionHook from opentelemetry.util.genai.metrics import InvocationMetricsRecorder from opentelemetry.util.genai.types import ( @@ -59,10 +55,6 @@ def __init__( self.name = name self.input_messages: list[InputMessage] = [] self.output_messages: list[OutputMessage] = [] - # Auto-mark as conversation root when no active span exists at creation time. - # Can be overridden by the caller after construction. - if not _has_active_span(): - self.conversation_root = True self._start(self._get_base_attributes()) def _get_base_attributes(self) -> dict[str, Any]: diff --git a/util/opentelemetry-util-genai/tests/test_handler_agent.py b/util/opentelemetry-util-genai/tests/test_handler_agent.py index 5de776cd9..9075b291e 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_agent.py +++ b/util/opentelemetry-util-genai/tests/test_handler_agent.py @@ -544,67 +544,6 @@ def get_description(self): assert captured_attributes[server_attributes.SERVER_PORT] == 8080 -class TestAgentConversationRoot(unittest.TestCase): - def setUp(self): - self.span_exporter = InMemorySpanExporter() - tracer_provider = TracerProvider() - tracer_provider.add_span_processor( - SimpleSpanProcessor(self.span_exporter) - ) - self.handler = TelemetryHandler(tracer_provider=tracer_provider) - - def test_root_agent_gets_conversation_root_true(self): - """Agent with no active parent span is auto-marked as conversation root.""" - invocation = self.handler.invoke_local_agent("openai", agent_name="Root Agent") - self.assertTrue(invocation.conversation_root) - invocation.stop() - - spans = self.span_exporter.get_finished_spans() - self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), True) - - def test_child_agent_does_not_get_conversation_root(self): - """Agent created inside an active parent span is NOT marked as root.""" - with self.handler.invoke_local_agent("openai", agent_name="Parent Agent") as _parent: - child = self.handler.invoke_local_agent("openai", agent_name="Child Agent") - self.assertIsNone(child.conversation_root) - child.stop() - - spans = self.span_exporter.get_finished_spans() - child_span = next(s for s in spans if s.name == "invoke_agent Child Agent") - self.assertNotIn("gen_ai.conversation_root", child_span.attributes) - - def test_explicit_conversation_root_false_is_respected(self): - """Explicitly setting conversation_root=False is not overridden.""" - invocation = self.handler.invoke_local_agent("openai") - invocation.conversation_root = False - invocation.stop() - - spans = self.span_exporter.get_finished_spans() - self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), False) - - def test_conversation_root_not_emitted_when_none(self): - """When conversation_root is None (child), attribute is not emitted.""" - with self.handler.invoke_local_agent("openai", agent_name="Parent") as _parent: - child = self.handler.invoke_local_agent("openai", agent_name="Child") - self.assertIsNone(child.conversation_root) - child.stop() - - spans = self.span_exporter.get_finished_spans() - child_span = next(s for s in spans if s.name == "invoke_agent Child") - self.assertNotIn("gen_ai.conversation_root", child_span.attributes) - - def test_remote_root_agent_gets_conversation_root_true(self): - """Remote agent with no active parent span is auto-marked as conversation root.""" - invocation = self.handler.invoke_remote_agent( - "openai", agent_name="Remote Root" - ) - self.assertTrue(invocation.conversation_root) - invocation.stop() - - spans = self.span_exporter.get_finished_spans() - self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), True) - - class TestAgentInvocationMetrics(TestBase): def test_local_agent_records_duration_and_tokens(self) -> None: handler = TelemetryHandler( diff --git a/util/opentelemetry-util-genai/tests/test_handler_workflow.py b/util/opentelemetry-util-genai/tests/test_handler_workflow.py index c6e08df94..a304f5b82 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_workflow.py +++ b/util/opentelemetry-util-genai/tests/test_handler_workflow.py @@ -266,50 +266,3 @@ def test_workflow_context_manager_with_messages(self) -> None: spans[0].attributes[GenAI.GEN_AI_OPERATION_NAME], "invoke_workflow", ) - - -class TelemetryHandlerWorkflowConversationRootTest(_WorkflowTestBase): - # ------------------------------------------------------------------ - # conversation_root auto-detection - # ------------------------------------------------------------------ - - def test_root_workflow_gets_conversation_root_true(self) -> None: - """Workflow with no active parent span is auto-marked as conversation root.""" - invocation = self.handler.workflow(name="root_wf") - self.assertTrue(invocation.conversation_root) - invocation.stop() - - spans = self._get_finished_spans() - self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), True) - - def test_child_workflow_does_not_get_conversation_root(self) -> None: - """Workflow created inside an active parent span is NOT marked as root.""" - with self.handler.workflow(name="parent_wf") as parent: - child = self.handler.workflow(name="child_wf") - self.assertIsNone(child.conversation_root) - child.stop() - - spans = self._get_finished_spans() - child_span = next(s for s in spans if s.name == "invoke_workflow child_wf") - self.assertNotIn("gen_ai.conversation_root", child_span.attributes) - - def test_explicit_conversation_root_false_is_respected(self) -> None: - """Explicitly setting conversation_root=False is not overridden.""" - invocation = self.handler.workflow(name="wf") - invocation.conversation_root = False - invocation.stop() - - spans = self._get_finished_spans() - self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), False) - - def test_conversation_root_not_emitted_when_none(self) -> None: - """When conversation_root is None, attribute is not emitted on the span.""" - with self.handler.workflow(name="parent") as _parent: - child = self.handler.workflow(name="child") - # conversation_root is None for child (has active parent) - self.assertIsNone(child.conversation_root) - child.stop() - - spans = self._get_finished_spans() - child_span = next(s for s in spans if s.name == "invoke_workflow child") - self.assertNotIn("gen_ai.conversation_root", child_span.attributes) From e22d3d45d79d45ad263bdad90c96d1b6007dff2b Mon Sep 17 00:00:00 2001 From: Wrisa Date: Wed, 1 Jul 2026 05:59:58 -0700 Subject: [PATCH 3/3] removed recording span logic and added context scoped attributes --- .../util/genai/_agent_invocation.py | 23 +++- .../opentelemetry/util/genai/_invocation.py | 32 +++++- .../util/genai/_workflow_invocation.py | 22 +++- .../util/genai/context_attributes.py | 69 +++++++++++ .../tests/test_context_attributes.py | 78 +++++++++++++ .../tests/test_handler_agent.py | 76 ++++++++++++ .../tests/test_handler_workflow.py | 108 ++++++++++++++++++ 7 files changed, 404 insertions(+), 4 deletions(-) create mode 100644 util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py create mode 100644 util/opentelemetry-util-genai/tests/test_context_attributes.py diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py index eebb05c5a..f31d91496 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_agent_invocation.py @@ -16,6 +16,10 @@ GenAIInvocation, get_content_attributes, ) +from opentelemetry.util.genai.context_attributes import ( + get_context_scoped_attributes, + set_context_scoped_attributes, +) from opentelemetry.util.genai.completion_hook import CompletionHook from opentelemetry.util.genai.metrics import InvocationMetricsRecorder from opentelemetry.util.genai.types import ( @@ -89,6 +93,20 @@ def __init__( self.finish_reasons: list[str] | None = None + # Use the context-scoped attribute key to determine whether a GenAI + # root already exists in the current context. This correctly ignores + # non-GenAI parents (HTTP, gRPC, etc.) — unlike checking OTel span + # parentage directly. + csa = get_context_scoped_attributes() + if "gen_ai.conversation_root" not in csa: + self.conversation_root = True + + # Propagate the marker so nested invocations do NOT mark themselves root. + extra_ctx = set_context_scoped_attributes( + {"gen_ai.conversation_root": True} + ) + self._start_extra_ctx = extra_ctx + self.input_tokens: int | None = None self.output_tokens: int | None = None self.cache_creation_input_tokens: int | None = None @@ -99,7 +117,10 @@ def __init__( self.system_instruction: list[MessagePart] = [] self.tool_definitions: list[ToolDefinition] | None = None - self._start(self._get_base_attributes()) + self._start( + self._get_base_attributes(), + extra_context=self._start_extra_ctx, + ) def _get_base_attributes(self) -> dict[str, Any]: """Return sampling-relevant attributes available at span creation time.""" diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py index 4b1d7e662..d1800d276 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_invocation.py @@ -13,6 +13,10 @@ from opentelemetry._logs import Logger, LogRecord from opentelemetry.context import Context, attach, detach +from opentelemetry.util.genai.context_attributes import ( + get_context_scoped_attributes, + set_context_scoped_attributes, +) from opentelemetry.semconv._incubating.attributes import ( gen_ai_attributes as GenAI, ) @@ -83,19 +87,39 @@ def __init__( self._span_kind: SpanKind = span_kind self._context_token: ContextToken | None = None self._monotonic_start_s: float | None = None + self.conversation_root: bool | None = None + """When True, marks this span as the root GenAI span of a conversation. + + Auto-set by WorkflowInvocation and AgentInvocation when no + ``gen_ai.conversation_root`` key is found in the current OTel context + (i.e. no enclosing GenAI root span exists). Uses a context-scoped + attribute key rather than raw OTel span parentage so that non-GenAI + parents (HTTP, gRPC, etc.) are correctly ignored. + + Can be explicitly set to override auto-detection. + """ - def _start(self, attributes: dict[str, Any] | None = None) -> None: + def _start( + self, + attributes: dict[str, Any] | None = None, + extra_context: Context | None = None, + ) -> None: """Start the invocation span and attach it to the current context. Args: attributes: Initial span attributes available for sampling decisions. + extra_context: Optional context that already contains additional + values (e.g. context-scoped attributes set by a subclass before + calling _start). When provided, the span is set into this context + rather than the bare current context, so both the span and any + extra values are attached together. """ self.span = self._tracer.start_span( name=self._span_name, kind=self._span_kind, attributes=attributes, ) - self._span_context = set_span_in_context(self.span) + self._span_context = set_span_in_context(self.span, extra_context) self._monotonic_start_s = timeit.default_timer() self._context_token = attach(self._span_context) @@ -147,6 +171,10 @@ def _finish(self, error: Error | None = None) -> None: return try: self._apply_finish(error) + if self.conversation_root is not None: + self.span.set_attribute( + "gen_ai.conversation_root", self.conversation_root + ) finally: try: detach(self._context_token) diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py index e7d6ff7e4..58a2b502a 100644 --- a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/_workflow_invocation.py @@ -12,6 +12,10 @@ ) from opentelemetry.trace import SpanKind, Tracer from opentelemetry.util.genai._invocation import Error, GenAIInvocation +from opentelemetry.util.genai.context_attributes import ( + get_context_scoped_attributes, + set_context_scoped_attributes, +) from opentelemetry.util.genai.completion_hook import CompletionHook from opentelemetry.util.genai.metrics import InvocationMetricsRecorder from opentelemetry.util.genai.types import ( @@ -55,7 +59,23 @@ def __init__( self.name = name self.input_messages: list[InputMessage] = [] self.output_messages: list[OutputMessage] = [] - self._start(self._get_base_attributes()) + + # Use the context-scoped attribute key to determine whether a GenAI + # root already exists in the current context. This correctly ignores + # non-GenAI parents (HTTP, gRPC, etc.) — unlike checking OTel span + # parentage directly. + csa = get_context_scoped_attributes() + if "gen_ai.conversation_root" not in csa: + # No enclosing GenAI root — this invocation is the root. + self.conversation_root = True + + # Propagate the marker to child spans via context-scoped attributes, + # so any nested WorkflowInvocation or AgentInvocation sees it and + # does NOT mark itself as root. + extra_ctx = set_context_scoped_attributes( + {"gen_ai.conversation_root": True} + ) + self._start(self._get_base_attributes(), extra_context=extra_ctx) def _get_base_attributes(self) -> dict[str, Any]: """Return sampling-relevant attributes available at span creation time.""" diff --git a/util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py new file mode 100644 index 000000000..e273960a0 --- /dev/null +++ b/util/opentelemetry-util-genai/src/opentelemetry/util/genai/context_attributes.py @@ -0,0 +1,69 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +""" +Context-scoped attributes for GenAI instrumentation. + +Follows the API shape proposed by OTel spec PR #4931 (Context-scoped Attributes). +Attributes stored here are process-local — they are never serialised into W3C +Baggage headers or any outbound propagation format. + +Currently used to propagate ``gen_ai.conversation_root`` from a root +WorkflowInvocation or AgentInvocation to child spans so that the root can be +identified without relying on OTel span parentage (which includes non-GenAI +parents such as HTTP spans). +""" + +from __future__ import annotations + +from typing import Any + +from opentelemetry import context as otel_context +from opentelemetry.context import Context + +# Private context key — never leaked outside this module. +_GENAI_CONTEXT_ATTRS_KEY = otel_context.create_key( + "opentelemetry.util.genai.context_scoped_attrs" +) + + +def set_context_scoped_attributes( + attrs: dict[str, Any], + context: Context | None = None, +) -> Context: + """Return a new Context with *attrs* merged in (existing keys win). + + Keys already present in the context are **not** overwritten — lower-priority + semantics matching the CSA spec: the first writer (outermost scope) wins. + + Args: + attrs: Attributes to add to the context. + context: Base context to merge into. Defaults to the current context. + + Returns: + A new Context containing the merged attributes. The caller is + responsible for attaching it if needed. + """ + ctx = context if context is not None else otel_context.get_current() + existing: dict[str, Any] = ( + otel_context.get_value(_GENAI_CONTEXT_ATTRS_KEY, context=ctx) or {} + ) + # Existing keys win — new attrs only fill in gaps. + merged = {**attrs, **existing} + return otel_context.set_value(_GENAI_CONTEXT_ATTRS_KEY, merged, ctx) + + +def get_context_scoped_attributes( + context: Context | None = None, +) -> dict[str, Any]: + """Return context-scoped GenAI attributes, or an empty dict. + + Args: + context: Context to read from. Defaults to the current context. + + Returns: + A dict of attributes previously set via + :func:`set_context_scoped_attributes`, or ``{}`` if none are present. + """ + ctx = context if context is not None else otel_context.get_current() + return otel_context.get_value(_GENAI_CONTEXT_ATTRS_KEY, context=ctx) or {} diff --git a/util/opentelemetry-util-genai/tests/test_context_attributes.py b/util/opentelemetry-util-genai/tests/test_context_attributes.py new file mode 100644 index 000000000..9519e824c --- /dev/null +++ b/util/opentelemetry-util-genai/tests/test_context_attributes.py @@ -0,0 +1,78 @@ +# Copyright The OpenTelemetry Authors +# SPDX-License-Identifier: Apache-2.0 + +from __future__ import annotations + +from unittest import TestCase + +from opentelemetry import context as otel_context +from opentelemetry.util.genai.context_attributes import ( + get_context_scoped_attributes, + set_context_scoped_attributes, +) + + +class TestSetContextScopedAttributes(TestCase): + def test_returns_new_context(self) -> None: + original = otel_context.get_current() + new_ctx = set_context_scoped_attributes({"key": "value"}, original) + self.assertIsNot(new_ctx, original) + + def test_values_readable_from_new_context(self) -> None: + ctx = set_context_scoped_attributes({"gen_ai.conversation_root": True}) + attrs = get_context_scoped_attributes(ctx) + self.assertEqual(attrs["gen_ai.conversation_root"], True) + + def test_multiple_attributes_stored(self) -> None: + ctx = set_context_scoped_attributes({"a": "1", "b": "2"}) + attrs = get_context_scoped_attributes(ctx) + self.assertEqual(attrs["a"], "1") + self.assertEqual(attrs["b"], "2") + + def test_existing_key_not_overwritten(self) -> None: + """Lower-priority semantics: a key already in context is not replaced.""" + ctx = set_context_scoped_attributes({"gen_ai.conversation_root": True}) + ctx2 = set_context_scoped_attributes( + {"gen_ai.conversation_root": False}, ctx + ) + attrs = get_context_scoped_attributes(ctx2) + # Original value wins + self.assertEqual(attrs["gen_ai.conversation_root"], True) + + def test_new_key_added_alongside_existing(self) -> None: + ctx = set_context_scoped_attributes({"first": "a"}) + ctx2 = set_context_scoped_attributes({"second": "b"}, ctx) + attrs = get_context_scoped_attributes(ctx2) + self.assertEqual(attrs["first"], "a") + self.assertEqual(attrs["second"], "b") + + def test_defaults_to_current_context(self) -> None: + ctx = set_context_scoped_attributes({"implicit": "yes"}) + token = otel_context.attach(ctx) + try: + attrs = get_context_scoped_attributes() + self.assertEqual(attrs["implicit"], "yes") + finally: + otel_context.detach(token) + + +class TestGetContextScopedAttributes(TestCase): + def test_empty_context_returns_empty_dict(self) -> None: + fresh_ctx = otel_context.get_current() + attrs = get_context_scoped_attributes(fresh_ctx) + self.assertEqual(attrs, {}) + + def test_no_argument_uses_current_context(self) -> None: + ctx = set_context_scoped_attributes({"k": "v"}) + token = otel_context.attach(ctx) + try: + attrs = get_context_scoped_attributes() + self.assertEqual(attrs["k"], "v") + finally: + otel_context.detach(token) + + def test_returns_same_dict_instance(self) -> None: + ctx = set_context_scoped_attributes({"x": "1"}) + attrs1 = get_context_scoped_attributes(ctx) + attrs2 = get_context_scoped_attributes(ctx) + self.assertIs(attrs1, attrs2) diff --git a/util/opentelemetry-util-genai/tests/test_handler_agent.py b/util/opentelemetry-util-genai/tests/test_handler_agent.py index 9075b291e..a209ec143 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_agent.py +++ b/util/opentelemetry-util-genai/tests/test_handler_agent.py @@ -544,6 +544,82 @@ def get_description(self): assert captured_attributes[server_attributes.SERVER_PORT] == 8080 +class TestAgentConversationRoot(unittest.TestCase): + def setUp(self): + self.span_exporter = InMemorySpanExporter() + tracer_provider = TracerProvider() + tracer_provider.add_span_processor( + SimpleSpanProcessor(self.span_exporter) + ) + self.handler = TelemetryHandler(tracer_provider=tracer_provider) + + def test_root_agent_gets_conversation_root_true(self): + """Agent with no enclosing GenAI context is auto-marked as root.""" + invocation = self.handler.invoke_local_agent("openai", agent_name="Root Agent") + assert invocation.conversation_root is True + invocation.stop() + + spans = self.span_exporter.get_finished_spans() + assert spans[0].attributes.get("gen_ai.conversation_root") is True + + def test_child_agent_does_not_get_conversation_root(self): + """Agent nested inside a parent GenAI span is NOT marked as root.""" + with self.handler.invoke_local_agent("openai", agent_name="Parent Agent"): + child = self.handler.invoke_local_agent("openai", agent_name="Child Agent") + assert child.conversation_root is None + child.stop() + + spans = self.span_exporter.get_finished_spans() + child_span = next(s for s in spans if s.name == "invoke_agent Child Agent") + assert "gen_ai.conversation_root" not in child_span.attributes + + def test_root_agent_under_non_genai_span_gets_conversation_root(self): + """Agent under a non-GenAI OTel span (e.g. HTTP) is still marked as root. + + The context-scoped attribute key is only set by GenAI invocations, + so a plain OTel parent span does not suppress conversation_root. + """ + from opentelemetry.sdk.trace import TracerProvider as _TP + tracer = _TP().get_tracer("test") + with tracer.start_as_current_span("HTTP POST /invoke"): + invocation = self.handler.invoke_local_agent("openai", agent_name="Agent") + assert invocation.conversation_root is True + invocation.stop() + + spans = self.span_exporter.get_finished_spans() + genai_span = next(s for s in spans if "invoke_agent" in s.name) + assert genai_span.attributes.get("gen_ai.conversation_root") is True + + def test_agent_nested_under_workflow_is_not_root(self): + """Agent created inside a workflow does not get conversation_root.""" + with self.handler.workflow(name="my_workflow"): + agent = self.handler.invoke_local_agent("openai", agent_name="Sub Agent") + assert agent.conversation_root is None + agent.stop() + + spans = self.span_exporter.get_finished_spans() + agent_span = next(s for s in spans if "invoke_agent" in s.name) + assert "gen_ai.conversation_root" not in agent_span.attributes + + def test_explicit_conversation_root_false_is_respected(self): + """Explicitly setting conversation_root=False is not overridden.""" + invocation = self.handler.invoke_local_agent("openai") + invocation.conversation_root = False + invocation.stop() + + spans = self.span_exporter.get_finished_spans() + assert spans[0].attributes.get("gen_ai.conversation_root") is False + + def test_remote_root_agent_gets_conversation_root_true(self): + """Remote agent with no enclosing GenAI context is auto-marked as root.""" + invocation = self.handler.invoke_remote_agent("openai", agent_name="Remote Root") + assert invocation.conversation_root is True + invocation.stop() + + spans = self.span_exporter.get_finished_spans() + assert spans[0].attributes.get("gen_ai.conversation_root") is True + + class TestAgentInvocationMetrics(TestBase): def test_local_agent_records_duration_and_tokens(self) -> None: handler = TelemetryHandler( diff --git a/util/opentelemetry-util-genai/tests/test_handler_workflow.py b/util/opentelemetry-util-genai/tests/test_handler_workflow.py index a304f5b82..9cb7213b6 100644 --- a/util/opentelemetry-util-genai/tests/test_handler_workflow.py +++ b/util/opentelemetry-util-genai/tests/test_handler_workflow.py @@ -266,3 +266,111 @@ def test_workflow_context_manager_with_messages(self) -> None: spans[0].attributes[GenAI.GEN_AI_OPERATION_NAME], "invoke_workflow", ) + + +class TelemetryHandlerWorkflowConversationRootTest(_WorkflowTestBase): + # ------------------------------------------------------------------ + # conversation_root via context-scoped attributes + # ------------------------------------------------------------------ + + def test_root_workflow_gets_conversation_root_true(self) -> None: + """Workflow with no enclosing GenAI context is auto-marked as root.""" + invocation = self.handler.workflow(name="root_wf") + self.assertTrue(invocation.conversation_root) + invocation.stop() + + spans = self._get_finished_spans() + self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), True) + + def test_child_workflow_does_not_get_conversation_root(self) -> None: + """Workflow nested inside another workflow is NOT marked as root.""" + with self.handler.workflow(name="parent_wf"): + child = self.handler.workflow(name="child_wf") + self.assertIsNone(child.conversation_root) + child.stop() + + spans = self._get_finished_spans() + child_span = next(s for s in spans if s.name == "invoke_workflow child_wf") + self.assertNotIn("gen_ai.conversation_root", child_span.attributes) + + def test_root_workflow_under_non_genai_span_gets_conversation_root(self) -> None: + """Workflow under a non-GenAI OTel span (e.g. HTTP) is still marked as root. + + The context-scoped attribute key is only written by GenAI invocations, + so a plain OTel parent span does not suppress conversation_root. + """ + from opentelemetry.sdk.trace import TracerProvider + tracer = TracerProvider().get_tracer("test") + with tracer.start_as_current_span("HTTP GET /plan"): + invocation = self.handler.workflow(name="wf_under_http") + self.assertTrue(invocation.conversation_root) + invocation.stop() + + spans = self._get_finished_spans() + genai_span = next(s for s in spans if "invoke_workflow" in s.name) + self.assertEqual(genai_span.attributes.get("gen_ai.conversation_root"), True) + + def test_explicit_conversation_root_false_is_respected(self) -> None: + """Explicitly setting conversation_root=False is not overridden.""" + invocation = self.handler.workflow(name="wf") + invocation.conversation_root = False + invocation.stop() + + spans = self._get_finished_spans() + self.assertEqual(spans[0].attributes.get("gen_ai.conversation_root"), False) + + def test_conversation_root_not_emitted_when_none(self) -> None: + """When conversation_root is None (child), attribute is not emitted.""" + with self.handler.workflow(name="parent"): + child = self.handler.workflow(name="child") + self.assertIsNone(child.conversation_root) + child.stop() + + spans = self._get_finished_spans() + child_span = next(s for s in spans if s.name == "invoke_workflow child") + self.assertNotIn("gen_ai.conversation_root", child_span.attributes) + + def test_context_key_cleared_after_workflow_ends(self) -> None: + """After the root workflow ends, the context key is detached.""" + from opentelemetry.util.genai.context_attributes import ( + get_context_scoped_attributes, + ) + invocation = self.handler.workflow(name="wf") + invocation.stop() + + # After stop, context_token is detached — key is gone from current context + attrs = get_context_scoped_attributes() + self.assertNotIn("gen_ai.conversation_root", attrs) + + def test_non_root_span_types_never_get_conversation_root(self) -> None: + """chat, execute_tool, retrieval, embedding spans are never marked as root. + + Only WorkflowInvocation and AgentInvocation participate in root + detection. All other invocation types must never emit conversation_root. + """ + from unittest.mock import patch + from opentelemetry.util.genai.types import ContentCapturingMode + + # chat (InferenceInvocation) — even when called at the top level + llm = self.handler.inference("openai", request_model="gpt-4") + llm.stop() + + # execute_tool (ToolInvocation) + tool = self.handler.tool("search_flights") + tool.stop() + + # embeddings (EmbeddingInvocation) + emb = self.handler.embedding("openai", request_model="text-embedding-3") + emb.stop() + + # retrieval (RetrievalInvocation) + ret = self.handler.retrieval(data_source_id="kb-prod") + ret.stop() + + spans = self._get_finished_spans() + for span in spans: + self.assertNotIn( + "gen_ai.conversation_root", + span.attributes, + msg=f"Span '{span.name}' should not have gen_ai.conversation_root", + )