fix(handlers): SAO-17638 LLM span double-encoding and workflow input

CHANGELOG.md

@@ -22,6 +22,12 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   (`langchain-google-genai`) does not pass Gemini's breakdown through, so its
   spans keep flat-rate cost.
 
+### Fixed
+
+- `SplunkAOTracingProcessor` (OpenAI Agents) no longer double-encodes Chat Completions LLM span
+  input and output. With Chat Completions models, workflow/agent spans without input now show the
+  user message instead of a `"<Type> Step"` placeholder. The Responses API path is unchanged.
+
 ## [0.4.0]
 
 ### Added

src/splunk_ao/handlers/openai_agents/handler.py

@@ -24,6 +24,7 @@
     SplunkAOCustomSpan,
     _extract_llm_data,
     _extract_tool_data,
+    _extract_user_input,
     _extract_workflow_data,
     _map_span_name,
     _map_span_type,
@@ -240,7 +241,7 @@ def _log_node_tree(self, node: Node, state: _OpenAITraceState, reuse_current: bo
         if node.node_type in ("agent", "chain", "workflow"):
             if not reuse_current:
                 self._splunk_ao_logger.add_workflow_span(
-                    input=input or node.node_type.capitalize() + " Step",
+                    input=input or state.first_input or node.node_type.capitalize() + " Step",
                     output=output,
                     name=name,
                     metadata=metadata,
@@ -354,7 +355,9 @@ def _update_owned_root(self, node: Node, state: _OpenAITraceState) -> None:
         """Apply final OpenAI callback data to the already-live root."""
         if state.owned_root is None:
             return
-        state.owned_root.input = node.span_params.get("input") or node.node_type.capitalize() + " Step"
+        state.owned_root.input = (
+            node.span_params.get("input") or state.first_input or node.node_type.capitalize() + " Step"
+        )
         if node.span_params.get("name") is not None:
             state.owned_root.name = node.span_params["name"]
         if node.span_params.get("tags") is not None:
@@ -395,10 +398,13 @@ def _finish_incremental_span(self, node: Node, state: _OpenAITraceState) -> None
             _logger.warning("Unable to complete OpenAI Agents span %s: no active state", node_id)
             return
         try:
-            if node.node_type in ("agent", "chain", "workflow") and not node.span_params.get("output"):
-                last_child = state.nodes.get(node.children[-1]) if node.children else None
-                if last_child is not None:
-                    node.span_params["output"] = last_child.span_params.get("output", "")
+            if node.node_type in ("agent", "chain", "workflow"):
+                if not node.span_params.get("input") and state.first_input:
+                    node.span_params["input"] = state.first_input
+                if not node.span_params.get("output"):
+                    last_child = state.nodes.get(node.children[-1]) if node.children else None
+                    if last_child is not None:
+                        node.span_params["output"] = last_child.span_params.get("output", "")
             final = finalize_handler_step(node, span_state, openai_agents=True)
             final = self._splunk_ao_logger._replace_handler_step(span_state.step, final)
             span_state.step = final
@@ -462,8 +468,8 @@ def _on_span_start(self, span: Span[Any]) -> None:
                     "status_code": llm_data.get("status_code", 200),
                 }
             )
-            if not state.first_input and initial_params.get("input") != serialize_to_str(None):
-                state.first_input = initial_params.get("input")
+            if not state.first_input:
+                state.first_input = _extract_user_input(initial_params.get("input"))
         elif splunk_ao_type == "tool":
             tool_data = _extract_tool_data(span.span_data)
             initial_params.update(
@@ -580,12 +586,8 @@ def _on_span_end(self, span: Span[Any]) -> None:
             # Ensure input is preserved if it wasn't available at start
             if node.span_params.get("input") is None:
                 node.span_params["input"] = llm_data.get("input")
-                if (
-                    not state.first_input
-                    and node.span_params["input"]
-                    and node.span_params["input"] != serialize_to_str(None)
-                ):
-                    state.first_input = node.span_params["input"]
+            if not state.first_input:
+                state.first_input = _extract_user_input(llm_data.get("input") or node.span_params.get("input"))
 
             # Extract embedded tool calls and merge with existing tool definitions
             if isinstance(span.span_data, ResponseSpanData) and span.span_data.response:

src/splunk_ao/utils/openai_agents.py

@@ -1,4 +1,5 @@
 import logging
+from collections.abc import Sequence
 from typing import Any, Literal
 
 from agents import (
@@ -20,6 +21,16 @@
 _logger = logging.getLogger(__name__)
 
 
+def _extract_user_input(llm_input: Any) -> str | None:
+    """Return the last user message content from a GenerationSpanData input list."""
+    if not isinstance(llm_input, list):
+        return None
+    for message in reversed(llm_input):
+        if isinstance(message, dict) and message.get("role") == "user":
+            return str(message.get("content") or "")
+    return None
+
+
 class SplunkAOCustomSpan(CustomSpanData):
     def __init__(self, span: SplunkAOSpan, data: dict[str, Any]):
         self.span = span
@@ -272,10 +283,19 @@ def _extract_llm_data(span_data: GenerationSpanData | ResponseSpanData) -> dict[
             if hasattr(response, "instructions") and response.instructions:
                 data["metadata"]["instructions"] = response.instructions
 
-    # Serialize complex inputs/outputs for logging
-    # Splunk AO expects input/output as serialized strings for llm spans
-    data["input"] = serialize_to_str(data["input"])
-    data["output"] = serialize_to_str(data["output"])
+    if isinstance(span_data, GenerationSpanData):
+        # GenerationSpanData: input/output are list[dict] — LoggedLlmSpan validators handle conversion.
+        # output is Sequence[Mapping] (a list from the SDK's models, possibly a tuple from a custom model),
+        # but LlmSpanAllowedOutputType only accepts a single dict; take the first choice.
+        if isinstance(data["output"], Sequence) and not isinstance(data["output"], str):
+            choices = list(data["output"])
+            if len(choices) > 1:
+                _logger.debug("GenerationSpanData.output has %d choices; only the first is recorded", len(choices))
+            data["output"] = choices[0] if choices else None
+    else:
+        # ResponseSpanData: output is a list of ResponseOutputItem objects — serialize for ingestion.
+        data["input"] = serialize_to_str(data["input"])
+        data["output"] = serialize_to_str(data["output"])
 
     if data["temperature"] is not None:
         try:

tests/test_openai_agents.py

@@ -1,4 +1,5 @@
 import asyncio
+import json
 import os
 import uuid
 from typing import Any
@@ -112,6 +113,112 @@ def test_openai_agents_children_enqueue_before_trace_end() -> None:
         logger.terminate()
 
 
+def test_workflow_span_input_falls_back_to_user_message() -> None:
+    # Given: a trace with an agent (workflow) span wrapping an LLM span
+    # When: the LLM span carries system + user messages and the workflow span has no input
+    # Then: the workflow span's gen_ai.input.messages reflects the user message, not the system prompt
+    sink = RecordingSink()
+    logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink)
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False)
+    set_trace_processors([processor])
+    try:
+        with agents_tracing.trace("Agent trace", trace_id="trace_wf_input") as openai_trace:
+            with agent_span("SupportAgent", span_id="span_wf", parent=openai_trace) as root_span:
+                with generation_span(
+                    input=[
+                        {"role": "system", "content": "You are a support agent."},
+                        {"role": "user", "content": "Look up customer CUST-123"},
+                    ],
+                    output=[{"role": "assistant", "content": "Found it."}],
+                    model="gpt-4",
+                    span_id="span_llm",
+                    parent=root_span,
+                ):
+                    pass
+
+        workflow_span = next(
+            s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "invoke_workflow"
+        )
+        input_messages = (workflow_span.attributes or {}).get("gen_ai.input.messages", "")
+        assert "Look up customer CUST-123" in input_messages
+        assert "You are a support agent." not in input_messages
+    finally:
+        set_trace_processors([])
+        logger.terminate()
+
+
+def test_llm_span_input_output_not_double_encoded() -> None:
+    # Given: a generation span with structured message list input/output
+    # When: the span is processed
+    # Then: chat span input contains multiple messages (not a single JSON-encoded string)
+    sink = RecordingSink()
+    logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink)
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False)
+    set_trace_processors([processor])
+    try:
+        with agents_tracing.trace("Agent trace", trace_id="trace_llm_io") as openai_trace:
+            with agent_span("Agent", span_id="span_agent", parent=openai_trace) as root_span:
+                with generation_span(
+                    input=[
+                        {"role": "system", "content": "You are helpful."},
+                        {"role": "user", "content": "What is the capital of France?"},
+                    ],
+                    output=[{"role": "assistant", "content": "Paris."}],
+                    model="gpt-4",
+                    span_id="span_gen",
+                    parent=root_span,
+                ):
+                    pass
+
+        llm_span = next(s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "chat")
+        input_messages = json.loads((llm_span.attributes or {}).get("gen_ai.input.messages", "[]"))
+        assert len(input_messages) == 2
+        assert input_messages[0]["role"] == "system"
+        assert input_messages[1]["role"] == "user"
+        output_messages = json.loads((llm_span.attributes or {}).get("gen_ai.output.messages", "[]"))
+        assert output_messages[0]["role"] == "assistant"
+        assert "Paris." in str(output_messages[0])
+    finally:
+        set_trace_processors([])
+        logger.terminate()
+
+
+@patch("splunk_ao.logger.logger.AgentStreams")
+@patch("splunk_ao.logger.logger.Projects")
+@patch("splunk_ao.logger.logger.Traces")
+def test_ingestion_hook_root_input_falls_back_to_user_message(
+    mock_traces_client: Mock, mock_projects_client: Mock, mock_logstreams_client: Mock
+) -> None:
+    # Given: the ingestion-hook path, where the root workflow span starts with a placeholder input
+    # When: an LLM span carrying the user message runs and the trace is committed
+    # Then: the committed root span's input is the user message, not "<Type> Step"
+    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", agent_stream="test", ingestion_hook=lambda _: None)
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False)
+    set_trace_processors([processor])
+    try:
+        with agents_tracing.trace("Agent trace", trace_id="trace_hook_input") as openai_trace:
+            with agent_span("SupportAgent", parent=openai_trace) as root_span:
+                with generation_span(
+                    input=[
+                        {"role": "system", "content": "You are a support agent."},
+                        {"role": "user", "content": "Look up customer CUST-123"},
+                    ],
+                    output=[{"role": "assistant", "content": "Found it."}],
+                    model="gpt-4",
+                    parent=root_span,
+                ):
+                    pass
+
+        root_workflow = logger.traces[0].spans[0]
+        assert root_workflow.input == "Look up customer CUST-123"
+    finally:
+        set_trace_processors([])
+        logger.terminate()
+
+
 def test_openai_agents_public_tracing_lifecycle_exports_and_cleans_state() -> None:
     sink = RecordingSink()
     logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink)
@@ -695,3 +802,95 @@ def model_dump(self):
     assert extracted_data_no_usage.get("num_total_tokens") is None
     assert "input_tokens_details" not in extracted_data_no_usage["metadata"]
     assert "output_tokens_details" not in extracted_data_no_usage["metadata"]
+
+
+def test_generation_span_exports_when_input_is_populated_after_start() -> None:
+    # Given: a generation span starts without input, as OpenAIChatCompletionsModel does
+    # When: OpenAI populates input, output, and usage after the start callback
+    # Then: the completed LLM span is exported with its messages and token counts
+    sink = RecordingSink()
+    logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink)
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False)
+    set_trace_processors([processor])
+    try:
+        with agents_tracing.trace("Delayed generation input") as openai_trace:
+            with agent_span("Agent", parent=openai_trace) as root_span:
+                with generation_span(model="gpt-4", parent=root_span) as span:
+                    span.span_data.input = [{"role": "user", "content": "Hi"}]
+                    span.span_data.output = [{"role": "assistant", "content": "Hello"}]
+                    span.span_data.usage = {"input_tokens": 1, "output_tokens": 2, "total_tokens": 3}
+
+        operations = [(s.attributes or {}).get("gen_ai.operation.name") for s in sink.spans]
+        assert "chat" in operations, f"Missing LLM span; exported operations: {operations}"
+        llm_span = next(s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "chat")
+        attributes = llm_span.attributes or {}
+        input_messages = json.loads(attributes.get("gen_ai.input.messages", "[]"))
+        output_messages = json.loads(attributes.get("gen_ai.output.messages", "[]"))
+        assert input_messages[0]["role"] == "user"
+        assert "Hi" in str(input_messages[0])
+        assert output_messages[0]["role"] == "assistant"
+        assert "Hello" in str(output_messages[0])
+        assert attributes["gen_ai.usage.input_tokens"] == 1
+        assert attributes["gen_ai.usage.output_tokens"] == 2
+        assert processor._trace_states == {}
+    finally:
+        set_trace_processors([])
+        logger.terminate()
+
+
+def test_generation_span_exports_tuple_output() -> None:
+    # Given: GenerationSpanData permits a Sequence of output mappings, including a tuple
+    # When: a generation span finishes with a tuple containing an assistant message
+    # Then: the completed LLM span is exported instead of being discarded during validation
+    sink = RecordingSink()
+    logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink)
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False)
+    set_trace_processors([processor])
+    try:
+        with agents_tracing.trace("Tuple generation output") as openai_trace:
+            with agent_span("Agent", parent=openai_trace) as root_span:
+                with generation_span(
+                    input=[{"role": "user", "content": "Hi"}], model="gpt-4", parent=root_span
+                ) as span:
+                    span.span_data.output = ({"role": "assistant", "content": "Hello"},)
+
+        operations = [(s.attributes or {}).get("gen_ai.operation.name") for s in sink.spans]
+        assert "chat" in operations, f"Missing LLM span; exported operations: {operations}"
+        llm_span = next(s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "chat")
+        output_messages = json.loads((llm_span.attributes or {}).get("gen_ai.output.messages", "[]"))
+        assert output_messages[0]["role"] == "assistant"
+        assert "Hello" in str(output_messages[0])
+        assert processor._trace_states == {}
+    finally:
+        set_trace_processors([])
+        logger.terminate()
+
+
+def test_workflow_input_falls_back_when_generation_input_is_set_after_start() -> None:
+    # Given: a generation span opened without input, then given it inside the span, as OpenAIChatCompletionsModel does
+    # When: the trace completes
+    # Then: the agent workflow span, which has no input of its own, shows the user message from that LLM call
+    sink = RecordingSink()
+    logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink)
+    processor = SplunkAOTracingProcessor(splunk_ao_logger=logger, flush_on_trace_end=False)
+    set_trace_processors([processor])
+    try:
+        with agents_tracing.trace("Delayed generation input") as openai_trace:
+            with agent_span("Agent", parent=openai_trace) as root_span:
+                with generation_span(model="gpt-4", parent=root_span) as gen:
+                    # Mirror OpenAIChatCompletionsModel: input is assigned after the span starts.
+                    gen.span_data.input = [
+                        {"role": "system", "content": "You are a support agent."},
+                        {"role": "user", "content": "Look up customer CUST-123"},
+                    ]
+                    gen.span_data.output = [{"role": "assistant", "content": "Found it."}]
+
+        workflows = [s for s in sink.spans if (s.attributes or {}).get("gen_ai.operation.name") == "invoke_workflow"]
+        assert workflows, "Missing workflow span"
+        for workflow in workflows:
+            input_messages = json.loads((workflow.attributes or {}).get("gen_ai.input.messages", "[]"))
+            assert input_messages[0]["role"] == "user"
+            assert input_messages[0]["parts"][0]["content"] == "Look up customer CUST-123"
+    finally:
+        set_trace_processors([])
+        logger.terminate()