GenAI Conversation Root

src/splunk_ao/logger/logger.py

@@ -1849,6 +1849,9 @@ def add_workflow_span(
             id=uuid.uuid4(),
             step_number=step_number,
         )
+        if isinstance(self.current_parent(), LoggedTrace):
+            span.conversation_root = True
+            span.user_metadata = {"gen_ai.conversation_root": "true", **(span.user_metadata or {})}
         return self._attach_parentable_span(span, status_code)
 
     @nop_sync
@@ -1932,6 +1935,9 @@ def add_agent_span(
             id=uuid.uuid4(),
             step_number=step_number,
         )
+        if isinstance(self.current_parent(), LoggedTrace):
+            span.conversation_root = True
+            span.user_metadata = {"gen_ai.conversation_root": "true", **(span.user_metadata or {})}
         return self._attach_parentable_span(span, status_code)
 
     @nop_sync

src/splunk_ao/otel.py

@@ -17,7 +17,7 @@
 from opentelemetry.trace import Tracer
 from requests import Session
 
-from galileo_core.schemas.logging.span import RetrieverSpan, ToolSpan, WorkflowSpan
+from galileo_core.schemas.logging.span import AgentSpan, RetrieverSpan, ToolSpan, WorkflowSpan
 from galileo_core.schemas.logging.span import Span as GalileoSpan
 from splunk_ao.config import SplunkAOConfig
 from splunk_ao.decorator import (
@@ -34,6 +34,8 @@
 
 logger = logging.getLogger(__name__)
 
+GEN_AI_CONVERSATION_ROOT = "gen_ai.conversation_root"
+
 
 class TracerProvider(Protocol):
     def add_span_processor(self, span_processor: Any) -> None: ...
@@ -361,9 +363,15 @@ def start_splunk_ao_span(galileo_span: GalileoSpan) -> Generator[trace.Span, Any
         tracer_provider = trace.get_tracer_provider()
         _TRACE_PROVIDER_CONTEXT_VAR.set(cast(TracerProvider, tracer_provider))
     tracer = tracer_provider.get_tracer("galileo-tracer")
+    is_conversation_root = (
+        not trace.get_current_span().get_span_context().is_valid
+        and isinstance(galileo_span, WorkflowSpan | AgentSpan)
+    )
     with tracer.start_as_current_span(galileo_span.name) as span:
         yield span
         span.set_attribute("gen_ai.system", "galileo-otel")
+        if is_conversation_root:
+            span.set_attribute(GEN_AI_CONVERSATION_ROOT, value=True)
         # Set dataset attributes for ground truth/reference output support
         _apply_dataset_attributes(
             span, galileo_span.dataset_input, galileo_span.dataset_output, galileo_span.dataset_metadata

src/splunk_ao/schema/logged.py

@@ -61,6 +61,7 @@ class LoggedWorkflowSpan(WorkflowSpan):
     output: IngestOutputType | None = _OUTPUT_FIELD
     redacted_output: IngestOutputType | None = _REDACTED_OUTPUT_FIELD
     spans: list["LoggedSpan"] = Field(default_factory=list)
+    conversation_root: bool | None = Field(default=None)
 
 
 class LoggedAgentSpan(AgentSpan):
@@ -71,6 +72,7 @@ class LoggedAgentSpan(AgentSpan):
     output: IngestOutputType | None = _OUTPUT_FIELD
     redacted_output: IngestOutputType | None = _REDACTED_OUTPUT_FIELD
     spans: list["LoggedSpan"] = Field(default_factory=list)
+    conversation_root: bool | None = Field(default=None)
 
 
 class LoggedLlmSpan(LlmSpan):

tests/schemas/test_logged.py

@@ -270,6 +270,8 @@ def test_full_ingest_request_roundtrip(self) -> None:
                         LoggedAgentSpan(
                             input=[LoggedMessage(content="agent task", role=MessageRole.user)],
                             output="agent done",
+                            conversation_root=True,
+                            user_metadata={"gen_ai.conversation_root": "true"},
                             spans=[
                                 LoggedWorkflowSpan(
                                     input=[LoggedMessage(content="wf", role=MessageRole.user)],
@@ -313,6 +315,8 @@ def test_full_ingest_request_roundtrip(self) -> None:
 
         agent = trace.spans[0]
         assert type(agent) is LoggedAgentSpan
+        assert agent.conversation_root is True
+        assert agent.user_metadata == {"gen_ai.conversation_root": "true"}
         assert isinstance(agent.input, list)
         assert type(agent.input[0]) is LoggedMessage
 

tests/test_crewai_handler.py

@@ -247,6 +247,28 @@ def test_crew_kickoff_started_empty_inputs(crewai_callback, generated_id) -> Non
         assert call_args[1]["input"] == "-"
 
 
+def test_crew_events_inherit_conversation_root_marking(crewai_callback, mock_splunk_ao_logger) -> None:
+    """A root crew workflow is marked while its nested task workflow is not."""
+    crew_id = uuid.uuid4()
+    task_id = uuid.uuid4()
+    crew = MockCrew(crew_id=crew_id)
+    agent = MockAgent(crew=crew)
+    task = MockTask(task_id=task_id, description="Research market trends", agent=agent)
+
+    crewai_callback._handle_crew_kickoff_started(MockSource(id=crew_id), MockEvent(crew_name="Test Crew"))
+    crewai_callback._handle_task_started(MockSource(id=task_id), MockEvent(task=task))
+    crewai_callback._handle_task_completed(MockSource(id=task_id), MockEvent(output=MockOutput("Done")))
+    crewai_callback._handle_crew_kickoff_completed(
+        MockSource(id=crew_id), MockEvent(output=MockOutput("Crew completed successfully"))
+    )
+
+    traces = mock_splunk_ao_logger.traces
+    assert len(traces) == 1
+    root_span = traces[0].spans[0]
+    assert root_span.conversation_root is True
+    assert root_span.spans[0].conversation_root is None
+
+
 @pytest.mark.parametrize("generated_id", [lambda: uuid.uuid4(), lambda: str(uuid.uuid4())])
 def test_crew_kickoff_completed(crewai_callback, generated_id) -> None:
     """Test crew kickoff completed event handling."""

tests/test_decorator.py

@@ -8,7 +8,7 @@
 from galileo_core.schemas.logging.span import AgentSpan, LlmSpan, RetrieverSpan, ToolSpan, WorkflowSpan
 from galileo_core.schemas.shared.document import Document
 from galileo_core.schemas.shared.multimodal import ContentModality
-from splunk_ao import Message, MessageRole, splunk_ao_context, log, start_session
+from splunk_ao import Message, MessageRole, log, splunk_ao_context, start_session
 from splunk_ao.decorator import _session_id_context
 from splunk_ao.schema.content_blocks import DataContentBlock, TextContentBlock
 from tests.testutils.setup import setup_mock_logstreams_client, setup_mock_projects_client, setup_mock_traces_client
@@ -257,6 +257,7 @@ def my_function(arg1, arg2):
     assert payload.traces[0].input == '{"arg1": 1, "arg2": 2}'
     assert payload.traces[0].spans[0].input == '{"arg1": 1, "arg2": 2}'
     assert payload.traces[0].spans[0].output == "3"
+    assert payload.traces[0].spans[0].conversation_root is True
 
 
 @patch("splunk_ao.logger.logger.LogStreams")
@@ -415,6 +416,7 @@ def my_function(arg1: str, arg2: str):
     assert payload.traces[0].spans[0].input == '{"arg1": "arg1", "arg2": "arg2"}'
     assert payload.traces[0].spans[0].output == "arg1"
     assert payload.traces[0].spans[0].agent_type == "planner"
+    assert payload.traces[0].spans[0].conversation_root is True
     assert len(payload.traces[0].spans[0].spans) == 1
     assert isinstance(payload.traces[0].spans[0].spans[0], ToolSpan)
     assert payload.traces[0].spans[0].spans[0].input == '{"arg1": "arg1"}'
@@ -449,6 +451,7 @@ def nested_call(nested_query: str):
     assert len(payload.traces[0].spans[0].spans) == 1
     assert isinstance(payload.traces[0].spans[0], WorkflowSpan)
     assert isinstance(payload.traces[0].spans[0].spans[0], LlmSpan)
+    assert payload.traces[0].spans[0].conversation_root is True
     assert payload.traces[0].input == '{"nested_query": "input"}'
     assert payload.traces[0].spans[0].input == '{"nested_query": "input"}'
     assert payload.traces[0].spans[0].output == output

tests/test_langchain.py

@@ -79,6 +79,7 @@ def test_on_chain_start_end(self, callback: SplunkAOCallback, splunk_ao_logger:
         assert traces[0].spans[0].input == '{"query": "test question"}'
         assert traces[0].spans[0].output == '{"result": "test answer"}'
         assert traces[0].spans[0].step_number is None
+        assert traces[0].spans[0].conversation_root is True
 
     def test_on_chain_start_with_kwargs_serialised_none(
         self, callback: SplunkAOCallback, splunk_ao_logger: SplunkAOLogger
@@ -898,10 +899,12 @@ def test_on_nested_agent_chains(self, callback: SplunkAOCallback, splunk_ao_logg
         outer_span = traces[0].spans[0]
         assert outer_span.type == "workflow"
         assert outer_span.name == "OuterChain"
+        assert outer_span.conversation_root is True
         assert len(outer_span.spans) == 1
         inner_span = outer_span.spans[0]
         assert inner_span.type == "agent"
         assert inner_span.name == "OuterChain:Agent"
+        assert inner_span.conversation_root is None
 
     def test_ai_message_with_list_content(self, callback: SplunkAOCallback, splunk_ao_logger: SplunkAOLogger) -> None:
         """Test AIMessage serialization with content as list of dicts (Responses API format)"""

tests/test_logger_batch.py

@@ -67,17 +67,56 @@ def test_disable_splunk_ao_logger(mock_traces_client: Mock, monkeypatch, caplog,
             total_tokens=13,
             duration_ns=1000,
         )
+        assert logger.add_workflow_span(input="workflow input") is None
+        assert logger.add_agent_span(input="agent input") is None
         logger.conclude(output="Nice try!", duration_ns=1000)
         logger.flush()
 
         assert "Bypassing logging for start_trace. Logging is currently disabled." in caplog.text
         assert "Bypassing logging for add_llm_span. Logging is currently disabled." in caplog.text
+        assert "Bypassing logging for add_workflow_span. Logging is currently disabled." in caplog.text
+        assert "Bypassing logging for add_agent_span. Logging is currently disabled." in caplog.text
         assert "Bypassing logging for conclude. Logging is currently disabled." in caplog.text
         assert "Bypassing logging for flush. Logging is currently disabled." in caplog.text
     mock_traces_client.assert_not_called()
     mock_traces_client.ingest_traces.assert_not_called()
 
 
+@patch("splunk_ao.logger.logger.LogStreams")
+@patch("splunk_ao.logger.logger.Projects")
+@patch("splunk_ao.logger.logger.Traces")
+def test_native_conversation_root_marks_direct_trace_children(
+    mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock
+) -> None:
+    """Workflow and agent spans directly under a trace are conversation roots."""
+    setup_mock_traces_client(mock_traces_client)
+    setup_mock_projects_client(mock_projects_client)
+    setup_mock_logstreams_client(mock_logstreams_client)
+
+    logger = SplunkAOLogger(project="my_project", log_stream="my_log_stream")
+    trace = logger.start_trace(input="trace input")
+    workflow = logger.add_workflow_span(
+        input="workflow input", metadata={"existing": "value", "gen_ai.conversation_root": "caller-value"}
+    )
+    nested_agent = logger.add_agent_span(input="nested agent input")
+    logger.conclude()
+    logger.conclude()
+    sibling_agent = logger.add_agent_span(input="sibling agent input")
+
+    assert workflow.conversation_root is True
+    assert workflow.user_metadata == {"existing": "value", "gen_ai.conversation_root": "caller-value"}
+    assert nested_agent.conversation_root is None
+    assert nested_agent.user_metadata == {}
+    assert sibling_agent.conversation_root is True
+    assert sibling_agent.user_metadata == {"gen_ai.conversation_root": "true"}
+
+    request = TracesIngestRequest(traces=[trace])
+    serialized = request.model_dump(mode="json")
+    assert serialized["traces"][0]["spans"][0]["conversation_root"] is True
+    assert serialized["traces"][0]["spans"][0]["user_metadata"]["gen_ai.conversation_root"] == "caller-value"
+    assert serialized["traces"][0]["spans"][1]["user_metadata"]["gen_ai.conversation_root"] == "true"
+
+
 @patch("splunk_ao.logger.logger.LogStreams")
 @patch("splunk_ao.logger.logger.Projects")
 @patch("splunk_ao.logger.logger.Traces")
@@ -1904,13 +1943,25 @@ def test_start_trace_invalid_redacted_input_type_raises() -> None:
         pytest.param(
             "add_workflow_span",
             {"input": "workflow input"},
-            {"intMeta": "1", "boolMeta": "True", "ratio": "3.14", "name": "test"},
+            {
+                "intMeta": "1",
+                "boolMeta": "True",
+                "ratio": "3.14",
+                "name": "test",
+                "gen_ai.conversation_root": "true",
+            },
             id="workflow_span",
         ),
         pytest.param(
             "add_agent_span",
             {"input": "agent input"},
-            {"intMeta": "1", "boolMeta": "True", "ratio": "3.14", "name": "test"},
+            {
+                "intMeta": "1",
+                "boolMeta": "True",
+                "ratio": "3.14",
+                "name": "test",
+                "gen_ai.conversation_root": "true",
+            },
             id="agent_span",
         ),
     ],

tests/test_logger_distributed.py

@@ -540,7 +540,7 @@ def test_conclude_trace_with_nested_span(
     assert request.spans[0].output is None
     assert request.spans[0].name == "test-workflow-span"
     assert request.spans[0].created_at == created_at
-    assert request.spans[0].user_metadata == metadata
+    assert request.spans[0].user_metadata == {**metadata, "gen_ai.conversation_root": "true"}
     assert len(request.spans[0].spans) == 0
     assert request.spans[0].metrics.duration_ns is None
     assert request.spans[0].step_number == 1
@@ -667,7 +667,7 @@ def test_conclude_all_with_nested_span(
     assert request.spans[0].output is None
     assert request.spans[0].name == "test-workflow-span"
     assert request.spans[0].created_at == created_at
-    assert request.spans[0].user_metadata == metadata
+    assert request.spans[0].user_metadata == {**metadata, "gen_ai.conversation_root": "true"}
     assert len(request.spans[0].spans) == 0
     workflow_span_id = request.spans[0].id
 
@@ -801,7 +801,7 @@ def test_conclude_trace_with_agent_span(
     assert request.spans[0].agent_type == "planner"
     assert request.spans[0].name == "test-agent-span"
     assert request.spans[0].created_at == created_at
-    assert request.spans[0].user_metadata == metadata
+    assert request.spans[0].user_metadata == {**metadata, "gen_ai.conversation_root": "true"}
     assert len(request.spans[0].spans) == 0
     assert request.spans[0].metrics.duration_ns == 1_000_000
     agent_span_id = request.spans[0].id
@@ -955,7 +955,7 @@ def test_trace_with_multiple_nested_spans(
     assert request.spans[0].agent_type == "planner"
     assert request.spans[0].name == "test-agent-span"
     assert request.spans[0].created_at == created_at
-    assert request.spans[0].user_metadata == metadata
+    assert request.spans[0].user_metadata == {**metadata, "gen_ai.conversation_root": "true"}
     assert request.spans[0].metrics.duration_ns is None
     assert len(request.spans[0].spans) == 0
     agent_span_id = request.spans[0].id
@@ -1025,7 +1025,7 @@ def test_trace_with_multiple_nested_spans(
     assert request.spans[0].output is None
     assert request.spans[0].name == "test-workflow-span"
     assert request.spans[0].created_at == created_at
-    assert request.spans[0].user_metadata == metadata
+    assert request.spans[0].user_metadata == {**metadata, "gen_ai.conversation_root": "true"}
     assert request.spans[0].metrics.duration_ns is None
     assert len(request.spans[0].spans) == 0
     workflow_span_id = request.spans[0].id
@@ -1161,7 +1161,7 @@ def test_trace_with_nested_span_and_sibling(
     assert request.spans[0].output is None
     assert request.spans[0].name == "test-workflow-span"
     assert request.spans[0].created_at == created_at
-    assert request.spans[0].user_metadata == metadata
+    assert request.spans[0].user_metadata == {**metadata, "gen_ai.conversation_root": "true"}
     assert len(request.spans[0].spans) == 0
     assert request.spans[0].metrics.duration_ns is None
     workflow_span_id = request.spans[0].id
@@ -1383,7 +1383,7 @@ def test_add_nested_span_and_conclude_existing_trace(
     assert request.spans[0].output == "workflow-output"
     assert request.spans[0].name == "test-workflow-span"
     assert request.spans[0].created_at == created_at
-    assert request.spans[0].user_metadata == metadata
+    assert request.spans[0].user_metadata == {**metadata, "gen_ai.conversation_root": "true"}
     assert request.spans[0].metrics.duration_ns == 1_000_000
     assert request.spans[0].step_number == 1
     workflow_span_id = request.spans[0].id

tests/test_openai_agents.py

@@ -1,4 +1,5 @@
 import os
+import uuid
 from unittest.mock import MagicMock, Mock, patch
 
 import httpx
@@ -21,6 +22,7 @@
 from galileo_core.schemas.logging.span import LlmSpan, ToolSpan
 from splunk_ao.handlers.openai_agents import SplunkAOTracingProcessor
 from splunk_ao.logger.logger import SplunkAOLogger
+from splunk_ao.schema.handlers import Node
 from splunk_ao.utils.openai_agents import _extract_llm_data, _parse_usage
 from tests.testutils.setup import setup_mock_logstreams_client, setup_mock_projects_client, setup_mock_traces_client
 
@@ -132,6 +134,37 @@ async def test_simple_agent(
     assert len(payload.traces[0].spans) == 1
 
 
+@patch("splunk_ao.logger.logger.LogStreams")
+@patch("splunk_ao.logger.logger.Projects")
+@patch("splunk_ao.logger.logger.Traces")
+def test_processor_marks_direct_trace_child_agent(
+    mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock
+) -> None:
+    """The processor inherits native root semantics without an OpenAI API call."""
+    setup_mock_traces_client(mock_traces_client)
+    setup_mock_projects_client(mock_projects_client)
+    setup_mock_logstreams_client(mock_logstreams_client)
+    logger = SplunkAOLogger(project="test", log_stream="test")
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False)
+    logger.start_trace(input="input")
+
+    processor._log_node_tree(
+        Node(
+            node_type="agent",
+            run_id=uuid.uuid4(),
+            span_params={
+                "input": "input",
+                "output": "output",
+                "name": "Agent step",
+                "start_time_iso": "2025-01-01T00:00:00+00:00",
+            },
+        )
+    )
+    logger.conclude(output="output")
+
+    assert logger.traces[0].spans[0].conversation_root is True
+
+
 def _create_mock_response_with_tools(tool_calls: list[dict]) -> dict:
     """Create a mock OpenAI API response with embedded tool calls."""
     return {