fix(converter): keep only last message from full-history agent output (HYBIM-988)

CHANGELOG.md

@@ -7,6 +7,20 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Agent and workflow output conversion now removes confirmed repeated input
+  history and keeps only the last message as the terminal output; intermediate
+  tool-call and tool-response messages are no longer included in
+  `gen_ai.output.messages` for full-history spans. The trim is now gated on the
+  prefix-match dedup check so parallel tool-call outputs (e.g. a ToolNode with
+  multiple simultaneous calls) are never collapsed to one message when the output
+  does not echo the input.
+- `gen_ai.output.messages` assistant entries whose parts contain a `tool_call`
+  but carry no explicit `finish_reason` now receive `"tool_call"` instead of
+  `"unknown"`. Explicit source finish reasons and tool-response messages are
+  unaffected.
+
 ## [0.2.1] - 2026-08-07
 
 ### Fixed

src/splunk_ao/converter/attribute_mapping.py

@@ -268,7 +268,13 @@ def _orchestration_messages(value: Any, default_role: str) -> tuple[list[dict[st
 def _with_finish_reasons(messages: list[dict[str, Any]], finish_reason: str | None = None) -> list[dict[str, Any]]:
     for message in messages:
         source_finish_reason = message.get("finish_reason")
-        message["finish_reason"] = finish_reason or source_finish_reason or "unknown"
+        # Infer "tool_call" when the message has tool_call parts but no explicit finish reason.
+        inferred_finish_reason = (
+            "tool_call"
+            if any(part.get("type") == "tool_call" for part in message.get("parts", []) if isinstance(part, Mapping))
+            else "unknown"
+        )
+        message["finish_reason"] = finish_reason or source_finish_reason or inferred_finish_reason
     return messages
 
 
@@ -445,8 +451,24 @@ def _set_orchestration_content(attrs: MutableMapping[str, AttributeValue], span:
     output_messages, full_history = _orchestration_messages(span.output, "assistant")
     if output_messages is None:
         return
-    if full_history and input_messages is not None and output_messages[: len(input_messages)] == input_messages:
+    # The trim is gated on a confirmed prefix match so that it only fires when the output
+    # is genuinely accumulated input history. Without this gate, any span whose output is a
+    # top-level {"messages": [...]} container would be reduced to a single message — a
+    # LangGraph ToolNode emitting one ToolMessage per parallel tool call would lose all but
+    # the last. Consequence: with a checkpointer the input is only the new turn while the
+    # output carries the whole persisted thread, so the prefix never matches and no
+    # reduction fires. That case remains unhandled.
+    if full_history and input_messages and output_messages[: len(input_messages)] == input_messages:
         output_messages = output_messages[len(input_messages) :]
+        # Prefer the last non-user message: the terminal message may legitimately be a tool
+        # response (return_direct) or a tool-call AIMessage, but a trailing user turn is an
+        # input, not this span's output.
+        terminal = next(
+            (index for index in reversed(range(len(output_messages))) if output_messages[index].get("role") != "user"),
+            None,
+        )
+        if terminal is not None:
+            output_messages = output_messages[terminal : terminal + 1]
     attrs["gen_ai.output.messages"] = _json_string(_with_finish_reasons(output_messages))
 
 

tests/test_attribute_mapping.py

@@ -1,6 +1,6 @@
 import json
 from types import SimpleNamespace
-from typing import cast
+from typing import Any, cast
 from uuid import uuid4
 
 import pytest
@@ -418,6 +418,273 @@ def test_orchestration_output_omits_repeated_input_history() -> None:
     ]
 
 
+@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
+def test_orchestration_full_history_keeps_last_terminal_message(
+    span_type: type[WorkflowSpan] | type[AgentSpan],
+) -> None:
+    user = {"role": "user", "content": "Give me two alternatives"}
+    first = {"role": "assistant", "content": "First alternative"}
+    second = {"role": "assistant", "content": "Second alternative"}
+    span_kwargs: dict[str, Any] = {
+        "name": "planner",
+        "input": json.dumps({"messages": [user]}),
+        "output": json.dumps({"messages": [user, first, second]}),
+    }
+    if span_type is AgentSpan:
+        span_kwargs["agent_type"] = AgentType.planner
+
+    attrs = build_span_attributes(span_type(**span_kwargs))
+
+    assert json.loads(attrs["gen_ai.output.messages"]) == [
+        _text_message("assistant", "Second alternative", finish_reason="unknown")
+    ]
+
+
+@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
+def test_orchestration_full_history_removes_confirmed_input_prefix(
+    span_type: type[WorkflowSpan] | type[AgentSpan],
+) -> None:
+    # Given: the input is the complete history immediately before the final assistant response.
+    user = {"role": "user", "content": "What is the dosage?"}
+    tool_call = {
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [{"id": "call-1", "function": {"name": "search", "arguments": '{"query":"dosage"}'}}],
+    }
+    tool_response = {"role": "tool", "content": "10 mg daily", "tool_call_id": "call-1"}
+    final = {"role": "assistant", "content": "The common dosage is 10 mg daily."}
+    input_history = [user, tool_call, tool_response]
+    span_kwargs: dict[str, Any] = {
+        "name": "healthcare",
+        "input": json.dumps({"messages": input_history}),
+        "output": json.dumps({"messages": [*input_history, final]}),
+    }
+    if span_type is AgentSpan:
+        span_kwargs["agent_type"] = AgentType.default
+
+    # When: the orchestration content is converted.
+    attrs = build_span_attributes(span_type(**span_kwargs))
+
+    # Then: only the newly produced terminal response is exported as output.
+    assert json.loads(attrs["gen_ai.output.messages"]) == [
+        _text_message("assistant", "The common dosage is 10 mg daily.", finish_reason="unknown")
+    ]
+
+
+def test_orchestration_tool_call_assistant_message_infers_tool_call_finish_reason() -> None:
+    # Given: a workflow emits an assistant tool call without a source finish reason.
+    output = {
+        "update": {
+            "messages": [
+                {
+                    "role": "assistant",
+                    "content": "",
+                    "tool_calls": [{"id": "call-1", "function": {"name": "search", "arguments": '{"query":"dosage"}'}}],
+                }
+            ]
+        }
+    }
+
+    # When: the workflow output is converted.
+    attrs = build_span_attributes(WorkflowSpan(name="tools", output=json.dumps(output)))
+
+    # Then: finish_reason is inferred as "tool_call" from the parts.
+    output_message = json.loads(attrs["gen_ai.output.messages"])[0]
+    assert output_message["finish_reason"] == "tool_call"
+    assert output_message["parts"][0]["type"] == "tool_call"
+
+
+def test_orchestration_tool_response_uses_unknown_finish_reason() -> None:
+    # Given: a workflow emits a tool response, which has no model-generation finish reason.
+    output = {
+        "update": {"messages": [{"role": "tool", "content": {"dosage": "10 mg daily"}, "tool_call_id": "call-1"}]}
+    }
+
+    # When: the workflow output is converted.
+    attrs = build_span_attributes(WorkflowSpan(name="tools", output=json.dumps(output)))
+
+    # Then: its valid tool response structure is preserved without inventing a model finish reason.
+    output_message = json.loads(attrs["gen_ai.output.messages"])[0]
+    assert output_message["finish_reason"] == "unknown"
+    assert output_message["parts"] == [
+        {"type": "tool_call_response", "id": "call-1", "response": {"dosage": "10 mg daily"}}
+    ]
+
+
+def test_orchestration_preserves_explicit_finish_reason_for_tool_call() -> None:
+    # Given: the source supplies its own finish reason for a message containing a tool call.
+    output = {
+        "update": {
+            "messages": [
+                {
+                    "role": "assistant",
+                    "content": "",
+                    "finish_reason": "provider_tool_calls",
+                    "tool_calls": [{"id": "call-1", "function": {"name": "search", "arguments": '{"query":"dosage"}'}}],
+                }
+            ]
+        }
+    }
+
+    # When: the workflow output is converted.
+    attrs = build_span_attributes(WorkflowSpan(name="tools", output=json.dumps(output)))
+
+    # Then: inference does not overwrite source telemetry.
+    assert json.loads(attrs["gen_ai.output.messages"])[0]["finish_reason"] == "provider_tool_calls"
+
+
+def test_orchestration_full_history_with_tool_call_keeps_last_message() -> None:
+    # LangGraph accumulated state: user → tool-call AI (empty content) → tool response → final AI
+    # The first post-dedup message has empty content; the UI would show "—" without the fix.
+    user = {"role": "user", "content": "What is the dosage of Lisinopril?"}
+    ai_toolcall = {
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [{"id": "tc1", "function": {"name": "rag_search", "arguments": '{"query":"Lisinopril dosage"}'}}],
+    }
+    tool_resp = {"role": "tool", "content": "Lisinopril: 10mg daily", "tool_call_id": "tc1"}
+    ai_final = {"role": "assistant", "content": "Common dosage is 10mg once daily."}
+
+    span = AgentSpan(
+        name="Agent",
+        agent_type=AgentType.default,
+        input=json.dumps({"messages": [user]}),
+        output=json.dumps({"messages": [user, ai_toolcall, tool_resp, ai_final]}),
+    )
+
+    attrs = build_span_attributes(span)
+
+    output_messages = json.loads(attrs["gen_ai.output.messages"])
+    assert len(output_messages) == 1
+    assert output_messages[0]["role"] == "assistant"
+    assert output_messages[0]["parts"][0]["content"] == "Common dosage is 10mg once daily."
+
+
+def test_orchestration_full_history_multi_turn_keeps_last_message() -> None:
+    # Multi-turn: output contains the full conversation history after multiple exchanges.
+    # Only the last message should be kept regardless of role.
+    user1 = {"role": "user", "content": "Hello"}
+    ai1 = {"role": "assistant", "content": "Hi, how can I help?"}
+    user2 = {"role": "user", "content": "What is Lisinopril?"}
+    ai2 = {"role": "assistant", "content": "Lisinopril is a blood pressure medication."}
+
+    span = AgentSpan(
+        name="Agent",
+        agent_type=AgentType.default,
+        input=json.dumps({"messages": [user1]}),
+        output=json.dumps({"messages": [user1, ai1, user2, ai2]}),
+    )
+
+    attrs = build_span_attributes(span)
+
+    output_messages = json.loads(attrs["gen_ai.output.messages"])
+    assert len(output_messages) == 1
+    assert output_messages[0]["parts"][0]["content"] == "Lisinopril is a blood pressure medication."
+
+
+def test_orchestration_full_history_multiple_tool_rounds_keeps_last_message() -> None:
+    # Two tool call rounds before the final answer — last message is still the only output.
+    user = {"role": "user", "content": "Compare Lisinopril and Amlodipine"}
+    tc1_ai = {
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [{"id": "tc1", "function": {"name": "search", "arguments": '{"query":"Lisinopril"}'}}],
+    }
+    tc1_resp = {"role": "tool", "content": "Lisinopril: ACE inhibitor", "tool_call_id": "tc1"}
+    tc2_ai = {
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [{"id": "tc2", "function": {"name": "search", "arguments": '{"query":"Amlodipine"}'}}],
+    }
+    tc2_resp = {"role": "tool", "content": "Amlodipine: calcium channel blocker", "tool_call_id": "tc2"}
+    ai_final = {
+        "role": "assistant",
+        "content": "Lisinopril is an ACE inhibitor; Amlodipine is a calcium channel blocker.",
+    }
+
+    span = AgentSpan(
+        name="Agent",
+        agent_type=AgentType.default,
+        input=json.dumps({"messages": [user]}),
+        output=json.dumps({"messages": [user, tc1_ai, tc1_resp, tc2_ai, tc2_resp, ai_final]}),
+    )
+
+    attrs = build_span_attributes(span)
+
+    output_messages = json.loads(attrs["gen_ai.output.messages"])
+    assert len(output_messages) == 1
+    assert "Amlodipine" in output_messages[0]["parts"][0]["content"]
+
+
+def test_orchestration_message_container_without_input_prefix_match_not_reduced() -> None:
+    # A WorkflowSpan (e.g. ToolNode) returning multiple messages whose output does NOT
+    # prefix-match the input state — dedup gate never fires, so all messages must survive.
+    # This is the parallel-tool-call shape: two ToolMessages from a single ToolNode invocation.
+    tool_msg_1 = {"role": "tool", "content": "Lisinopril: 10 mg daily", "tool_call_id": "tc1"}
+    tool_msg_2 = {"role": "tool", "content": "Amlodipine: 5 mg daily", "tool_call_id": "tc2"}
+    span = WorkflowSpan(
+        name="tools",
+        input=json.dumps({"messages": [{"role": "user", "content": "Compare dosages"}]}),
+        output=json.dumps({"messages": [tool_msg_1, tool_msg_2]}),
+    )
+
+    attrs = build_span_attributes(span)
+
+    output_messages = json.loads(attrs["gen_ai.output.messages"])
+    assert len(output_messages) == 2
+    assert output_messages[0]["parts"][0]["response"] == "Lisinopril: 10 mg daily"
+    assert output_messages[1]["parts"][0]["response"] == "Amlodipine: 5 mg daily"
+
+
+def test_orchestration_full_history_ends_on_tool_message_keeps_last() -> None:
+    # return_direct=True tool: run ends on a tool response, no final assistant message.
+    # The dedup gate fires (prefix matches) but the last message is a tool, not assistant.
+    # Must return the tool message rather than an empty list.
+    user = {"role": "user", "content": "Get patient P001"}
+    ai_toolcall = {
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [{"id": "tc1", "function": {"name": "get_patient", "arguments": '{"id":"P001"}'}}],
+    }
+    tool_resp = {"role": "tool", "content": "George Rivera, Lisinopril 10mg", "tool_call_id": "tc1"}
+    span = AgentSpan(
+        name="Agent",
+        agent_type=AgentType.default,
+        input=json.dumps({"messages": [user, ai_toolcall]}),
+        output=json.dumps({"messages": [user, ai_toolcall, tool_resp]}),
+    )
+
+    attrs = build_span_attributes(span)
+
+    output_messages = json.loads(attrs["gen_ai.output.messages"])
+    assert len(output_messages) == 1
+    assert output_messages[0]["parts"][0]["response"] == "George Rivera, Lisinopril 10mg"
+
+
+def test_orchestration_full_history_ends_on_tool_call_ai_message_keeps_last() -> None:
+    # interrupt_before=["tools"]: run ends on a tool-call AIMessage with empty content.
+    # The last message is assistant role but content="" — must not return empty list.
+    user = {"role": "user", "content": "Search for Lisinopril"}
+    ai_toolcall = {
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [{"id": "tc1", "function": {"name": "search", "arguments": '{"query":"Lisinopril"}'}}],
+    }
+    span = AgentSpan(
+        name="Agent",
+        agent_type=AgentType.default,
+        input=json.dumps({"messages": [user]}),
+        output=json.dumps({"messages": [user, ai_toolcall]}),
+    )
+
+    attrs = build_span_attributes(span)
+
+    output_messages = json.loads(attrs["gen_ai.output.messages"])
+    assert len(output_messages) == 1
+    assert output_messages[0]["role"] == "assistant"
+    assert output_messages[0]["finish_reason"] == "tool_call"
+
+
 def test_orchestration_preserves_schema_valid_parts_and_tool_calls() -> None:
     span = WorkflowSpan(
         name="tool-workflow",