Prototype showing gen_ai.conversation_root span attribute to mark the root GenAI span of a conversation

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."""

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)

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."""

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 {}

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)

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(