spike: gen_ai.conversation_root — native + OTel ingest paths

examples/agent/langchain-agent-http/.env.example

@@ -0,0 +1,17 @@
+# Splunk Agent Observability — copy this file to .env and fill in your values.
+
+SPLUNK_AO_API_KEY=your-splunk-ao-api-key
+SPLUNK_AO_API_URL=https://agent-observability-api.us0.signalfx.com
+SPLUNK_AO_CONSOLE_URL=https://agent-observability.us0.signalfx.com
+SPLUNK_AO_USE_DIRECT_API=true
+SPLUNK_AO_PROJECT=your-project-name
+SPLUNK_AO_LOG_STREAM=your-log-stream
+
+# OpenAI (standard)
+OPENAI_API_KEY=your-openai-api-key
+OPENAI_MODEL=gpt-4o-mini
+
+# Azure OpenAI (optional — overrides OPENAI_API_KEY / OPENAI_MODEL if set)
+# OPENAI_API_KEY=your-azure-openai-api-key
+# OPENAI_BASE_URL=https://your-resource.cognitiveservices.azure.com/openai/v1/
+# AZURE_OPENAI_API_VERSION=2024-02-01

examples/agent/langchain-agent-http/.gitignore

@@ -0,0 +1,4 @@
+.env
+__pycache__/
+*.pyc
+.venv/

examples/agent/langchain-agent-http/app.py

@@ -0,0 +1,177 @@
+"""FastAPI HTTP wrapper around the langchain-agent example.
+
+Exposes POST /invoke and POST /invoke/nested endpoints that accept a JSON body
+with a `prompt` field, pass it to a LangGraph ReAct agent, and return the response.
+
+This mirrors the "Scenarios" test layout from splunk-otel-python-contrib PR #236:
+  • HTTP span (FastAPI request) → GenAI Workflow span (LangChain/LangGraph agent)
+  • The Workflow span has no GenAI parent, so `gen_ai.conversation_root` is set
+    to True via the auto-detection logic in SplunkAOLogger.add_workflow_span()
+    (native path) and start_splunk_ao_span() (OTel path).
+
+Verification:
+  1. Start the server:
+       uvicorn app:app --reload --port 8080
+  2. Send a request:
+       curl -X POST http://localhost:8080/invoke \\
+            -H "Content-Type: application/json" \\
+            -d '{"prompt": "Say hello to Erin"}'
+  3. Observe the trace in the Splunk AO console at SPLUNK_AO_CONSOLE_URL and
+     confirm the root WorkflowSpan carries gen_ai.conversation_root=true in its
+     user_metadata (and on the LoggedWorkflowSpan.conversation_root field).
+
+Environment variables (see .env.example):
+  SPLUNK_AO_API_KEY, SPLUNK_AO_API_URL, SPLUNK_AO_PROJECT, SPLUNK_AO_LOG_STREAM,
+  OPENAI_API_KEY (or OPENAI_BASE_URL + AZURE_OPENAI_* for Azure).
+"""
+
+import logging
+import os
+
+from dotenv import load_dotenv
+from fastapi import FastAPI, HTTPException
+from langchain.tools import tool
+from langchain_openai import ChatOpenAI
+from langgraph.prebuilt import create_react_agent
+from pydantic import BaseModel
+
+from splunk_ao import splunk_ao_context
+from splunk_ao.handlers.langchain import SplunkAOCallback
+
+load_dotenv()
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+app = FastAPI(
+    title="LangChain Agent HTTP — gen_ai.conversation_root demo",
+    description="Wraps a LangGraph ReAct agent in FastAPI for e2e conversation-root tracing verification.",
+    version="0.1.0",
+)
+
+
+# ---------------------------------------------------------------------------
+# Agent setup — built once at startup, reused across requests.
+# ---------------------------------------------------------------------------
+
+_PROJECT = os.getenv("SPLUNK_AO_PROJECT", "langchain-http-demo")
+_LOG_STREAM = os.getenv("SPLUNK_AO_LOG_STREAM", "langchain-http")
+
+
+@tool
+def greet(name: str) -> str:
+    """Say hello to someone by name."""
+    return f"Hello, {name}! 👋"
+
+
+@tool
+def get_weather(city: str) -> str:
+    """Return a made-up weather report for a city (demo tool)."""
+    return f"It's sunny and 22 °C in {city} right now."
+
+
+_llm = ChatOpenAI(
+    model=os.getenv("OPENAI_MODEL", "gpt-4o-mini"),
+    temperature=0.7,
+)
+
+_agent = create_react_agent(
+    model=_llm,
+    tools=[greet, get_weather],
+)
+
+
+def _invoke_with_tracing(prompt: str) -> str:
+    """Run the agent under a splunk_ao_context so each invocation is traced."""
+    callback = SplunkAOCallback()
+    with splunk_ao_context(project=_PROJECT, log_stream=_LOG_STREAM):
+        result = _agent.invoke(
+            {"messages": [{"role": "user", "content": prompt}]},
+            config={"callbacks": [callback]},
+        )
+    # Last message in the graph output is the final AI response.
+    messages = result.get("messages", [])
+    return messages[-1].content if messages else str(result)
+
+
+# ---------------------------------------------------------------------------
+# Request / response models
+# ---------------------------------------------------------------------------
+
+
+class InvokeRequest(BaseModel):
+    """Payload accepted by POST /invoke."""
+
+    prompt: str
+    """The user prompt to send to the agent (e.g. 'Say hello to Erin')."""
+
+
+class InvokeResponse(BaseModel):
+    """Response returned by POST /invoke."""
+
+    response: str
+    """The agent's final text output."""
+    note: str = (
+        "The root WorkflowSpan for this request should carry "
+        "gen_ai.conversation_root=true (user_metadata) and "
+        "LoggedWorkflowSpan.conversation_root=True."
+    )
+
+
+# ---------------------------------------------------------------------------
+# Routes
+# ---------------------------------------------------------------------------
+
+
+@app.get("/health")
+def health() -> dict:
+    """Liveness probe."""
+    return {"status": "ok"}
+
+
+@app.post("/invoke", response_model=InvokeResponse)
+def invoke_agent(body: InvokeRequest) -> InvokeResponse:
+    """Invoke the LangGraph ReAct agent with a user prompt.
+
+    The HTTP span (this FastAPI handler) is *not* a GenAI span, so the first
+    GenAI WorkflowSpan created by the SplunkAOCallback will be detected as the
+    conversation root (gen_ai.conversation_root = True).
+
+    Scenario mirrors PR #236 test cases:
+      - single_agent_under_http: HTTP request → WorkflowSpan (root = True)
+    """
+    logger.info("Received prompt: %s", body.prompt)
+    try:
+        output = _invoke_with_tracing(body.prompt)
+    except Exception as exc:
+        logger.exception("Agent invocation failed")
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+    logger.info("Agent response: %s", output)
+    return InvokeResponse(response=output)
+
+
+@app.post("/invoke/nested", response_model=InvokeResponse)
+def invoke_nested(body: InvokeRequest) -> InvokeResponse:
+    """Invoke two sequential agent calls in one HTTP request.
+
+    Mirrors the 'two_sequential_agents' scenario from PR #236:
+    each _invoke_with_tracing() call produces a separate trace, and each
+    trace's root WorkflowSpan gets conversation_root=True independently.
+    """
+    logger.info("Received nested prompt: %s", body.prompt)
+    try:
+        out1 = _invoke_with_tracing(body.prompt)
+        out2 = _invoke_with_tracing(f"Summarise in one sentence: {out1}")
+    except Exception as exc:
+        logger.exception("Nested agent invocation failed")
+        raise HTTPException(status_code=500, detail=str(exc)) from exc
+
+    output = f"[Pass 1] {out1} | [Pass 2] {out2}"
+    return InvokeResponse(response=output)
+
+
+if __name__ == "__main__":
+    import uvicorn
+
+    uvicorn.run("app:app", host="0.0.0.0", port=8080, reload=True)

examples/agent/langchain-agent-http/requirements.txt

@@ -0,0 +1,7 @@
+fastapi==0.139.2
+uvicorn[standard]==0.38.0
+python-dotenv==1.2.1
+langchain==1.2.15
+langchain-openai==1.2.0
+openai==2.41.0
+splunk-ao>=0.1.0

src/splunk_ao/logger/logger.py

@@ -1664,6 +1664,11 @@ def add_workflow_span(
             id=uuid.uuid4(),
             step_number=step_number,
         )
+        # Auto-mark as conversation root when this workflow is the first GenAI span
+        # in the trace (its direct parent is the LoggedTrace, not another GenAI span).
+        if isinstance(self.current_parent(), LoggedTrace):
+            span.conversation_root = True
+            span.user_metadata = {**(span.user_metadata or {}), "gen_ai.conversation_root": "true"}
         return self._attach_parentable_span(span, status_code)
 
     @nop_sync
@@ -1747,6 +1752,11 @@ def add_agent_span(
             id=uuid.uuid4(),
             step_number=step_number,
         )
+        # Auto-mark as conversation root when this agent span is the first GenAI span
+        # in the trace (its direct parent is the LoggedTrace, not another GenAI span).
+        if isinstance(self.current_parent(), LoggedTrace):
+            span.conversation_root = True
+            span.user_metadata = {**(span.user_metadata or {}), "gen_ai.conversation_root": "true"}
         return self._attach_parentable_span(span, status_code)
 
     @nop_sync

src/splunk_ao/otel.py

@@ -9,7 +9,7 @@
 
 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 (
@@ -26,6 +26,9 @@
 
 logger = logging.getLogger(__name__)
 
+# Semantic-convention attribute key for marking the invocation-level GenAI root span.
+# Mirrors the constant from opentelemetry-util-genai (splunk-otel-python-contrib PR #236).
+GEN_AI_CONVERSATION_ROOT = "gen_ai.conversation_root"
 
 INSTALL_ERR_MSG = (
     "OpenTelemetry packages are not installed. "
@@ -407,9 +410,18 @@ 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")
+    # Capture root status BEFORE entering the span context so we see the caller's
+    # OTel parent (or absence of one).  A Workflow/Agent span with no valid OTel
+    # parent is, by definition, the conversation root for this trace.
+    _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, 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,9 @@ class LoggedWorkflowSpan(WorkflowSpan):
     output: IngestOutputType | None = _OUTPUT_FIELD
     redacted_output: IngestOutputType | None = _REDACTED_OUTPUT_FIELD
     spans: list["LoggedSpan"] = Field(default_factory=list)
+    # When True, marks this span as the invocation-level GenAI root for the trace
+    # (mirrors gen_ai.conversation_root from the OTel semantic conventions).
+    conversation_root: bool | None = Field(default=None)
 
 
 class LoggedAgentSpan(AgentSpan):
@@ -71,6 +74,9 @@ class LoggedAgentSpan(AgentSpan):
     output: IngestOutputType | None = _OUTPUT_FIELD
     redacted_output: IngestOutputType | None = _REDACTED_OUTPUT_FIELD
     spans: list["LoggedSpan"] = Field(default_factory=list)
+    # When True, marks this span as the invocation-level GenAI root for the trace
+    # (mirrors gen_ai.conversation_root from the OTel semantic conventions).
+    conversation_root: bool | None = Field(default=None)
 
 
 class LoggedLlmSpan(LlmSpan):