From d62efc0cb8faaafbcda3e063b02cc61e3cd27323 Mon Sep 17 00:00:00 2001 From: "L. Elaine Dazzio" <170764058+LEDazzio01@users.noreply.github.com> Date: Mon, 23 Feb 2026 18:08:30 -0500 Subject: [PATCH 1/2] fix: prevent doubled tool_call arguments in MESSAGES_SNAPSHOT MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit When streaming with client-side tools, some providers send a full- arguments replay after the streaming deltas complete. The `_emit_tool_call` function unconditionally appends every arguments delta to the internal `flow.tool_calls_by_id` tracking dictionary via `+=`. When the replay contains the exact same complete arguments string that was already accumulated from prior deltas, the arguments get doubled (e.g., `{"todoText":"buy groceries"}{"todoText":"buy groceries"}`). This causes `MESSAGES_SNAPSHOT` events to contain invalid doubled JSON in `tool_calls[].function.arguments`, breaking any client or middleware that relies on snapshots for state reconstruction. The fix adds a guard (mirroring the existing duplicate guard in `_emit_text`) that detects when the incoming delta exactly equals the already-accumulated arguments string, indicating a full-arguments replay rather than an incremental delta. In this case the append is skipped, preventing the doubling. The `ToolCallArgsEvent` deltas are still emitted correctly for real-time streaming — only the internal snapshot accumulator is guarded. Fixes #4194 --- .../ag-ui/agent_framework_ag_ui/_run_common.py | 14 +++++++++++++- 1 file changed, 13 insertions(+), 1 deletion(-) diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py b/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py index cec86bdcf3..8701b18af6 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py @@ -198,7 +198,19 @@ def _emit_tool_call( events.append(ToolCallArgsEvent(tool_call_id=tool_call_id, delta=delta)) if tool_call_id in flow.tool_calls_by_id: - flow.tool_calls_by_id[tool_call_id]["function"]["arguments"] += delta + accumulated = flow.tool_calls_by_id[tool_call_id]["function"]["arguments"] + # Guard against full-argument replay: if the accumulated arguments + # already equal the incoming delta, this is a non-delta replay of + # the complete arguments string (some providers send the full + # arguments again after streaming deltas). Skip the append to + # prevent doubling in MESSAGES_SNAPSHOT. (Fixes #4194) + if accumulated and delta == accumulated: + logger.debug( + "Skipping duplicate full-arguments replay for tool_call_id=%s", + tool_call_id, + ) + else: + flow.tool_calls_by_id[tool_call_id]["function"]["arguments"] += delta if predictive_handler and flow.tool_call_name: delta_events = predictive_handler.emit_streaming_deltas(flow.tool_call_name, delta) From 6f2f9fa970be7581165890ae7b09dce15f69e18e Mon Sep 17 00:00:00 2001 From: "L. Elaine Dazzio" <170764058+LEDazzio01@users.noreply.github.com> Date: Mon, 23 Feb 2026 18:26:16 -0500 Subject: [PATCH 2/2] fix: move duplicate check before event emission + add test Address Copilot review feedback: 1. Move duplicate full-arguments replay detection BEFORE emitting ToolCallArgsEvent, for consistency with _emit_text() which returns early without emitting any events on replay detection. 2. Add test_emit_tool_call_skips_duplicate_full_arguments_replay() to verify the duplicate detection behavior for tool call arguments, matching the existing test pattern for text content. --- .../agent_framework_ag_ui/_run_common.py | 15 +++++--- python/packages/ag-ui/tests/ag_ui/test_run.py | 37 +++++++++++++++++++ 2 files changed, 47 insertions(+), 5 deletions(-) diff --git a/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py b/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py index 8701b18af6..997c375ed1 100644 --- a/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py +++ b/python/packages/ag-ui/agent_framework_ag_ui/_run_common.py @@ -195,22 +195,27 @@ def _emit_tool_call( delta = ( content.arguments if isinstance(content.arguments, str) else json.dumps(make_json_safe(content.arguments)) ) - events.append(ToolCallArgsEvent(tool_call_id=tool_call_id, delta=delta)) if tool_call_id in flow.tool_calls_by_id: accumulated = flow.tool_calls_by_id[tool_call_id]["function"]["arguments"] # Guard against full-argument replay: if the accumulated arguments # already equal the incoming delta, this is a non-delta replay of # the complete arguments string (some providers send the full - # arguments again after streaming deltas). Skip the append to - # prevent doubling in MESSAGES_SNAPSHOT. (Fixes #4194) + # arguments again after streaming deltas). Skip the event emission + # and accumulation to prevent doubling in MESSAGES_SNAPSHOT. + # This mirrors the early-return behaviour of _emit_text(). + # (Fixes #4194) if accumulated and delta == accumulated: logger.debug( "Skipping duplicate full-arguments replay for tool_call_id=%s", tool_call_id, ) - else: - flow.tool_calls_by_id[tool_call_id]["function"]["arguments"] += delta + return events + + events.append(ToolCallArgsEvent(tool_call_id=tool_call_id, delta=delta)) + + if tool_call_id in flow.tool_calls_by_id: + flow.tool_calls_by_id[tool_call_id]["function"]["arguments"] += delta if predictive_handler and flow.tool_call_name: delta_events = predictive_handler.emit_streaming_deltas(flow.tool_call_name, delta) diff --git a/python/packages/ag-ui/tests/ag_ui/test_run.py b/python/packages/ag-ui/tests/ag_ui/test_run.py index 4abd63c799..73c9648c02 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_run.py +++ b/python/packages/ag-ui/tests/ag_ui/test_run.py @@ -6,6 +6,7 @@ from ag_ui.core import ( TextMessageEndEvent, TextMessageStartEvent, + ToolCallArgsEvent, ) from agent_framework import AgentResponseUpdate, Content, Message, ResponseStream from agent_framework.exceptions import AgentInvalidResponseException @@ -416,6 +417,42 @@ def test_emit_tool_call_generates_id(): assert flow.tool_call_id is not None # ID should be generated +def test_emit_tool_call_skips_duplicate_full_arguments_replay(): + """Test _emit_tool_call skips replayed full-arguments on an existing tool call. + + This is a regression test for issue #4194 where some streaming providers + send the full arguments string again after streaming deltas, causing the + arguments to be doubled in MESSAGES_SNAPSHOT events. + + Mirrors test_emit_text_skips_duplicate_full_message_delta for consistency. + """ + flow = FlowState() + full_args = '{"city": "Seattle"}' + + # Step 1: Initial tool call with name + arguments (normal start) + content_start = Content.from_function_call( + call_id="call_dup", + name="get_weather", + arguments=full_args, + ) + events_start = _emit_tool_call(content_start, flow) + + # Should emit ToolCallStartEvent + ToolCallArgsEvent + assert any(isinstance(e, ToolCallArgsEvent) for e in events_start) + assert flow.tool_calls_by_id["call_dup"]["function"]["arguments"] == full_args + + # Step 2: Provider replays the full arguments (duplicate) + content_replay = Content(type="function_call", call_id="call_dup", arguments=full_args) + events_replay = _emit_tool_call(content_replay, flow) + + # Should NOT emit any ToolCallArgsEvent (early return on replay) + args_events = [e for e in events_replay if isinstance(e, ToolCallArgsEvent)] + assert args_events == [], "Duplicate full-arguments replay should not emit ToolCallArgsEvent" + + # Accumulated arguments should remain unchanged + assert flow.tool_calls_by_id["call_dup"]["function"]["arguments"] == full_args + + def test_emit_tool_result_closes_open_message(): """Test _emit_tool_result emits TextMessageEndEvent for open text message.