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