fix(converter): export only the current turn as workflow and agent input (SAO-19572)

CHANGELOG.md

@@ -7,6 +7,16 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+### Fixed
+
+- Agent and workflow spans no longer repeat earlier conversation turns in `gen_ai.input.messages`. When an app
+  re-sends the whole conversation, either as the graph state (for example a LangGraph `{"messages": [...]}` state) or
+  as a bare message list (for example the LangChain agent middleware), the exported input now starts at the current
+  user message, so continuous conversations no longer grow each trace's input with previous inputs and outputs
+  (SAO-19572). Tool-call and tool-response messages of the current turn are kept. A workflow or agent input that is
+  deliberately a message list, such as few-shot examples, is trimmed the same way. LLM span inputs still carry the
+  full prompt sent to the model.
+
 ## [0.7.0]
 
 ### Added

src/splunk_ao/converter/attribute_mapping.py

@@ -469,10 +469,34 @@ def _matching_child_finish_reason(span: WorkflowSpan | AgentSpan, output_message
     return None
 
 
+def _current_turn_messages(messages: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    """Return the messages from the last user message onward, or all of them when there is no user message."""
+    for index in range(len(messages) - 1, -1, -1):
+        if messages[index].get("role") == "user":
+            return messages[index:]
+    return messages
+
+
+def _is_bare_message_list(value: Any) -> bool:
+    """Return True when the value is a non-empty list of role-tagged messages, not a state or update container."""
+    parsed = _parse_json_string(value)
+    return (
+        isinstance(parsed, Sequence)
+        and not isinstance(parsed, str | bytes | bytearray)
+        and _is_message_sequence(parsed)
+    )
+
+
 def _set_orchestration_content(attrs: MutableMapping[str, AttributeValue], span: WorkflowSpan | AgentSpan) -> None:
-    input_messages, _ = _orchestration_messages(span.input, "user")
+    input_messages, input_full_history = _orchestration_messages(span.input, "user")
     if input_messages is not None:
-        attrs["gen_ai.input.messages"] = _json_string(input_messages)
+        # An app that re-sends the whole conversation, either as graph state ({"messages": [...]}) or as a
+        # bare message list (for example the LangChain agent middleware), would otherwise repeat every
+        # earlier turn in each new span's input. Export only the current turn; the untrimmed input_messages
+        # stay available below for the output history dedup.
+        trim_input = input_full_history or _is_bare_message_list(span.input)
+        exported_input = _current_turn_messages(input_messages) if trim_input else input_messages
+        attrs["gen_ai.input.messages"] = _json_string(exported_input)
 
     if span.output is None:
         return

tests/test_attribute_mapping.py

@@ -457,6 +457,189 @@ def test_orchestration_output_omits_repeated_input_history() -> None:
     ]
 
 
+def _orchestration_span(span_type: type[WorkflowSpan] | type[AgentSpan], **kwargs: Any) -> WorkflowSpan | AgentSpan:
+    span_kwargs: dict[str, Any] = {"name": "chatbot", **kwargs}
+    if span_type is AgentSpan:
+        span_kwargs["agent_type"] = AgentType.default
+    return span_type(**span_kwargs)
+
+
+@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
+def test_orchestration_input_keeps_only_current_turn_of_full_history(
+    span_type: type[WorkflowSpan] | type[AgentSpan],
+) -> None:
+    # Given: an app that re-sends the whole conversation (3 user turns) as the graph state on every request
+    user1 = {"role": "user", "content": "Look up patient P001"}
+    ai1 = {"role": "assistant", "content": "Patient P001 is George Rivera."}
+    user2 = {"role": "user", "content": "What is the dosage of Lisinopril?"}
+    ai2 = {"role": "assistant", "content": "10-40 mg once daily."}
+    user3 = {"role": "user", "content": "Look up patient P001 again"}
+    span = _orchestration_span(
+        span_type,
+        input=json.dumps({"messages": [user1, ai1, user2, ai2, user3]}),
+        output=json.dumps({"messages": [{"role": "assistant", "content": "Patient P001 is George Rivera."}]}),
+    )
+
+    # When: the orchestration content is converted
+    attrs = build_span_attributes(span)
+
+    # Then: only the current user turn is exported as input, not the earlier questions and replies
+    assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "Look up patient P001 again")]
+
+
+@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
+def test_orchestration_input_keeps_tool_loop_messages_of_current_turn(
+    span_type: type[WorkflowSpan] | type[AgentSpan],
+) -> None:
+    # Given: a graph node revisited mid-turn, whose state holds an earlier turn plus the current tool loop
+    user1 = {"role": "user", "content": "Hello"}
+    ai1 = {"role": "assistant", "content": "Hi, how can I help?"}
+    user2 = {"role": "user", "content": "Look up patient P001"}
+    tool_call = {
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [{"id": "call-1", "function": {"name": "get_patient_info", "arguments": '{"id":"P001"}'}}],
+    }
+    tool_response = {"role": "tool", "content": "George Rivera", "tool_call_id": "call-1"}
+    span = _orchestration_span(
+        span_type,
+        input=json.dumps({"messages": [user1, ai1, user2, tool_call, tool_response]}),
+        output=json.dumps({"messages": [{"role": "assistant", "content": "Patient P001 is George Rivera."}]}),
+    )
+
+    # When: the orchestration content is converted
+    attrs = build_span_attributes(span)
+
+    # Then: the input starts at the current user turn and keeps its tool call and tool response
+    input_messages = json.loads(attrs["gen_ai.input.messages"])
+    assert [message["role"] for message in input_messages] == ["user", "assistant", "tool"]
+    assert input_messages[0] == _text_message("user", "Look up patient P001")
+
+
+@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
+def test_orchestration_input_without_user_message_is_unchanged(span_type: type[WorkflowSpan] | type[AgentSpan]) -> None:
+    # Given: a state container with no user message to anchor the current turn on
+    ai = {"role": "assistant", "content": "Earlier reply"}
+    tool_response = {"role": "tool", "content": "result", "tool_call_id": "call-1"}
+    span = _orchestration_span(span_type, input=json.dumps({"messages": [ai, tool_response]}))
+
+    # When: the orchestration content is converted
+    attrs = build_span_attributes(span)
+
+    # Then: nothing is dropped from the input
+    assert [message["role"] for message in json.loads(attrs["gen_ai.input.messages"])] == ["assistant", "tool"]
+
+
+@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
+def test_orchestration_input_trim_keeps_output_history_dedup(span_type: type[WorkflowSpan] | type[AgentSpan]) -> None:
+    # Given: a multi-turn input and an output that echoes the full input history before the new reply
+    user1 = {"role": "user", "content": "Hello"}
+    ai1 = {"role": "assistant", "content": "Hi, how can I help?"}
+    user2 = {"role": "user", "content": "What is Lisinopril?"}
+    final = {"role": "assistant", "content": "A blood pressure medication."}
+    span = _orchestration_span(
+        span_type,
+        input=json.dumps({"messages": [user1, ai1, user2]}),
+        output=json.dumps({"messages": [user1, ai1, user2, final]}),
+    )
+
+    # When: the orchestration content is converted
+    attrs = build_span_attributes(span)
+
+    # Then: the input is trimmed to the current turn and the output still removes the full echoed history
+    assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "What is Lisinopril?")]
+    assert json.loads(attrs["gen_ai.output.messages"]) == [
+        _text_message("assistant", "A blood pressure medication.", finish_reason="unknown")
+    ]
+
+
+def test_orchestration_input_partial_update_container_is_unchanged() -> None:
+    # Given: a LangGraph partial update (not the full state), which only carries newly produced messages
+    user1 = {"role": "user", "content": "Hello"}
+    user2 = {"role": "user", "content": "Second question"}
+    span = AgentSpan(
+        name="Agent", agent_type=AgentType.default, input=json.dumps({"update": {"messages": [user1, user2]}})
+    )
+
+    # When: the orchestration content is converted
+    attrs = build_span_attributes(span)
+
+    # Then: the update is exported as is
+    assert json.loads(attrs["gen_ai.input.messages"]) == [
+        _text_message("user", "Hello"),
+        _text_message("user", "Second question"),
+    ]
+
+
+@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
+def test_orchestration_bare_message_list_input_keeps_only_current_turn(
+    span_type: type[WorkflowSpan] | type[AgentSpan],
+) -> None:
+    # Given: an agent whose input is the conversation as a bare message list (the LangChain agent middleware shape)
+    user1 = {"role": "user", "content": "What is the weather in Paris?"}
+    ai1 = {"role": "assistant", "content": "It is 72 degrees."}
+    user2 = {"role": "user", "content": "And the AAPL stock price?"}
+    ai2 = {"role": "assistant", "content": "It is 178.50 USD."}
+    user3 = {"role": "user", "content": "Remind me of the Paris weather"}
+    span = _orchestration_span(span_type, input=json.dumps([user1, ai1, user2, ai2, user3]))
+
+    # When: the orchestration content is converted
+    attrs = build_span_attributes(span)
+
+    # Then: only the current user turn is exported as input
+    assert json.loads(attrs["gen_ai.input.messages"]) == [_text_message("user", "Remind me of the Paris weather")]
+
+
+@pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
+def test_orchestration_bare_message_list_input_keeps_tool_loop_of_current_turn(
+    span_type: type[WorkflowSpan] | type[AgentSpan],
+) -> None:
+    # Given: a bare message list holding an earlier turn plus the current turn's tool call and tool response
+    user1 = {"role": "user", "content": "Hello"}
+    ai1 = {"role": "assistant", "content": "Hi, how can I help?"}
+    user2 = {"role": "user", "content": "Look up patient P001"}
+    tool_call = {
+        "role": "assistant",
+        "content": "",
+        "tool_calls": [{"id": "call-1", "function": {"name": "get_patient_info", "arguments": '{"id":"P001"}'}}],
+    }
+    tool_response = {"role": "tool", "content": "George Rivera", "tool_call_id": "call-1"}
+    span = _orchestration_span(span_type, input=json.dumps([user1, ai1, user2, tool_call, tool_response]))
+
+    # When: the orchestration content is converted
+    attrs = build_span_attributes(span)
+
+    # Then: the input starts at the current user turn and keeps its tool call and tool response
+    input_messages = json.loads(attrs["gen_ai.input.messages"])
+    assert [message["role"] for message in input_messages] == ["user", "assistant", "tool"]
+    assert input_messages[0] == _text_message("user", "Look up patient P001")
+
+
+@pytest.mark.parametrize(
+    "messages",
+    [
+        pytest.param([{"role": "user", "content": "Only question"}], id="single_user_turn"),
+        pytest.param(
+            [{"role": "assistant", "content": "Earlier reply"}, {"role": "tool", "content": "r", "tool_call_id": "c"}],
+            id="no_user_message",
+        ),
+    ],
+)
+def test_orchestration_bare_message_list_input_without_earlier_turns_is_unchanged(
+    messages: list[dict[str, Any]],
+) -> None:
+    # Given: a bare message list with a single user turn, or with no user message to anchor the current turn on
+    span = AgentSpan(name="Agent", agent_type=AgentType.default, input=json.dumps(messages))
+
+    # When: the orchestration content is converted
+    attrs = build_span_attributes(span)
+
+    # Then: nothing is dropped from the input
+    assert [message["role"] for message in json.loads(attrs["gen_ai.input.messages"])] == [
+        message["role"] for message in messages
+    ]
+
+
 @pytest.mark.parametrize("span_type", [WorkflowSpan, AgentSpan])
 def test_orchestration_full_history_keeps_last_terminal_message(
     span_type: type[WorkflowSpan] | type[AgentSpan],

tests/test_crewai_handler.py

@@ -1,6 +1,7 @@
 import asyncio
 import hashlib
 import inspect
+import json
 import sys
 import threading
 import uuid
@@ -2152,3 +2153,65 @@ def test_tool_lifecycle_previous_event_id_with_error(crewai_callback) -> None:
         start_run_id = mock_start.call_args[1]["run_id"]
         end_run_id = mock_end.call_args[1]["run_id"]
         assert start_run_id == end_run_id
+
+
+def _exported_message_texts(span: ReadableSpan) -> list[tuple[str, str]]:
+    messages = json.loads(span.attributes["gen_ai.input.messages"])
+    return [
+        (m["role"], " ".join(p.get("content", "") for p in m["parts"] if p.get("type") == "text")) for m in messages
+    ]
+
+
+def test_crewai_root_input_stays_current_turn_when_kickoff_inputs_resend_history(crewai_bus) -> None:
+    # Given: a chat-style crew kicked off 3 times, each time with the whole conversation as its inputs
+    _listener, _logger, sink, events, deliver = crewai_bus
+    crew = MockCrew()
+    agent = MockAgent(crew=crew)
+    task = MockTask(agent=agent)
+    questions = ["Name one AI benefit in healthcare", "Name a second benefit", "Summarize the two benefits"]
+    history: list[dict[str, str]] = []
+
+    async def run() -> None:
+        for turn, question in enumerate(questions):
+            history.append({"role": "user", "content": question})
+            call_id = f"call-{turn}"
+            # When: the turn runs with the full conversation as the kickoff inputs and the model prompt
+            await deliver(
+                [
+                    (crew, events["CrewKickoffStartedEvent"](crew_name="Crew", inputs={"messages": list(history)})),
+                    (task, events["TaskStartedEvent"](task=task)),
+                    (agent, events["AgentExecutionStartedEvent"](agent=agent, task=task, task_prompt=question)),
+                    (
+                        MockSource(),
+                        events["LLMCallStartedEvent"](
+                            call_id=call_id,
+                            model="local-model",
+                            agent_id=str(agent.id),
+                            task_id=str(task.id),
+                            messages=list(history),
+                        ),
+                    ),
+                    (
+                        MockSource(),
+                        events["LLMCallCompletedEvent"](
+                            call_id=call_id, agent_id=str(agent.id), task_id=str(task.id), response=f"Reply {turn}"
+                        ),
+                    ),
+                    (agent, events["AgentExecutionCompletedEvent"](agent=agent, task=task, output=f"Reply {turn}")),
+                    (task, events["TaskCompletedEvent"](task=task, output=MockOutput(f"Reply {turn}"))),
+                    (crew, events["CrewKickoffCompletedEvent"](output=MockOutput(f"Reply {turn}"))),
+                ]
+            )
+            history.append({"role": "assistant", "content": f"Reply {turn}"})
+
+    asyncio.run(run())
+
+    # Then: each crew root exports only the current question, whatever the turn
+    for turn, question in enumerate(questions):
+        spans = sink.spans[turn * 4 : turn * 4 + 4]
+        chat = next(s for s in spans if s.attributes["gen_ai.operation.name"] == "chat")
+        root = [s for s in spans if s.attributes["gen_ai.operation.name"] == "invoke_workflow"][-1]
+        assert _exported_message_texts(root) == [("user", question)]
+        # And: the chat span keeps the full prompt the model received, including every earlier question
+        prompt = " ".join(text for _role, text in _exported_message_texts(chat))
+        assert all(earlier in prompt for earlier in questions[: turn + 1])

tests/test_langchain.py

@@ -1,3 +1,4 @@
+import json
 import time
 import uuid
 from collections.abc import Generator
@@ -1810,3 +1811,58 @@ def test_async_callback_with_ingestion_hook_no_credentials(self):
 
         # Then: the callback is created successfully and the hook is attached
         assert callback._handler._splunk_ao_logger._ingestion_hook is mock_hook
+
+
+def _exported_message_texts(span: ReadableSpan, attribute: str = "gen_ai.input.messages") -> list[tuple[str, str]]:
+    messages = json.loads((span.attributes or {})[attribute])
+    return [
+        (m["role"], " ".join(p.get("content", "") for p in m["parts"] if p.get("type") == "text")) for m in messages
+    ]
+
+
+def test_langgraph_root_input_stays_current_turn_when_state_resends_history() -> None:
+    # Given: a LangGraph-style root, a node, and a chat model call, over 3 turns whose state re-sends the whole history
+    sink = RecordingSink()
+    logger = SplunkAOLogger(project_id="project-id", agent_stream_id="stream-id", _sink=sink)
+    callback = SplunkAOCallback(splunk_ao_logger=logger, flush_on_chain_end=False)
+    questions = ["Look up patient P001", "What is the dosage of Lisinopril?", "Look up patient P001 again"]
+    history: list[Any] = []
+    try:
+        for turn, question in enumerate(questions, start=1):
+            history.append(HumanMessage(content=question))
+            root_id, node_id, llm_id = uuid.uuid4(), uuid.uuid4(), uuid.uuid4()
+            reply = AIMessage(content=f"Reply {turn}")
+
+            # When: the turn runs with the full conversation as the graph state
+            callback.on_chain_start(
+                serialized={"name": "LangGraph"}, inputs={"messages": list(history)}, run_id=root_id
+            )
+            callback.on_chain_start(
+                serialized={"name": "chatbot"},
+                inputs={"messages": list(history)},
+                run_id=node_id,
+                parent_run_id=root_id,
+            )
+            callback.on_chat_model_start(
+                serialized={"name": "gpt"}, messages=[list(history)], run_id=llm_id, parent_run_id=node_id
+            )
+            callback.on_llm_end(
+                LLMResult(generations=[[ChatGeneration(message=reply)]], llm_output=None),
+                run_id=llm_id,
+                parent_run_id=node_id,
+            )
+            callback.on_chain_end(outputs={"messages": [reply]}, run_id=node_id, parent_run_id=root_id)
+            callback.on_chain_end(outputs={"messages": [*history, reply]}, run_id=root_id)
+            history.append(reply)
+
+        # Then: the root agent and the node export only the current question, whatever the turn
+        operations = [(span.attributes or {})["gen_ai.operation.name"] for span in sink.spans]
+        assert operations == ["chat", "invoke_workflow", "invoke_agent"] * 3
+        for turn, question in enumerate(questions):
+            chat, node, root = sink.spans[turn * 3 : turn * 3 + 3]
+            assert _exported_message_texts(root) == [("user", question)]
+            assert _exported_message_texts(node) == [("user", question)]
+            # And: the chat span keeps the full prompt the model received (1, 3 and 5 messages)
+            assert len(_exported_message_texts(chat)) == 2 * turn + 1
+    finally:
+        logger.terminate()