From 695f511f415bab196f4d95adbcc3a856eca4b790 Mon Sep 17 00:00:00 2001 From: "yu.xia" Date: Wed, 28 Oct 2026 00:58:01 +0800 Subject: [PATCH 1/3] fix(ag-ui): start each replayed declaration-only tool call once Guard TOOL_CALL_START with the started-call set instead of the last-id cursor: with two declaration-only calls in one response, the cursor sits on the second when the function loop replays the pauses, so both calls re-emitted START and duplicated in the thread snapshot. Fixes #9638 --- .../agent_framework_ag_ui/_run_common.py | 5 ++++- .../ag-ui/tests/ag_ui/test_run_common.py | 22 +++++++++++++++++++ 2 files changed, 26 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 8afb2e8002e..b07ed4a4de5 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 @@ -732,7 +732,10 @@ def _emit_tool_call( tool_call_id = content.call_id or flow.tool_call_id or generate_event_id() - if content.name and tool_call_id != flow.tool_call_id: + # Why a started-set, not the last-id cursor: declaration-only calls replay after the + # function loop pauses; with two calls the cursor sits on the second, so the replay of + # the first (and then the second) re-emitted TOOL_CALL_START (#9638). + if content.name and tool_call_id not in flow.tool_calls_by_id: flow.tool_call_id = tool_call_id flow.tool_call_name = content.name if predictive_handler: diff --git a/python/packages/ag-ui/tests/ag_ui/test_run_common.py b/python/packages/ag-ui/tests/ag_ui/test_run_common.py index 69822014420..de7c83206a5 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_run_common.py +++ b/python/packages/ag-ui/tests/ag_ui/test_run_common.py @@ -27,6 +27,7 @@ from agent_framework_ag_ui._run_common import ( FlowState, _build_run_finished_event, + _emit_content, _close_reasoning_block, _emit_mcp_tool_result, _emit_text_reasoning, @@ -1108,3 +1109,24 @@ def test_new_turn_gets_new_reasoning_id(self): assert id1 is not None assert id2 is not None assert id1 != id2, "New turn should get a new reasoning message_id" + + +def test_replayed_declaration_only_calls_start_each_tool_call_once() -> None: + """Two declaration-only calls replayed after streaming must not re-emit START (#9638).""" + flow = FlowState() + streamed = [ + Content.from_function_call(call_id="call_a", name="get_weather", arguments='{"city": "NYC"}'), + Content.from_function_call(call_id="call_b", name="get_time", arguments='{"zone": "UTC"}'), + ] + replayed = [ + Content.from_function_call(call_id="call_a", name="get_weather"), + Content.from_function_call(call_id="call_b", name="get_time"), + ] + events = [] + for content in [*streamed, *replayed]: + events.extend(_emit_content(content, flow)) + + starts = [e for e in events if e.type == EventType.TOOL_CALL_START] + assert sorted(e.tool_call_id for e in starts) == ["call_a", "call_b"] + ids = [entry["id"] for entry in flow.pending_tool_calls] + assert sorted(ids) == ["call_a", "call_b"] From 803db775b58b4f71b2ad32480b98e1ce550d41e6 Mon Sep 17 00:00:00 2001 From: "yu.xia" Date: Wed, 28 Oct 2026 01:26:00 +0800 Subject: [PATCH 2/3] fix(ag-ui): let a reused provider call id start a new occurrence Guard TOOL_CALL_START on whether the call is still open instead of started-id membership: a completed id reused by a later distinct call STARTs again with a fresh snapshot entry and becomes endable again, while declaration-only replay of an open call stays deduplicated. --- .../agent_framework_ag_ui/_run_common.py | 13 +++++++---- .../ag-ui/tests/ag_ui/test_run_common.py | 22 +++++++++++++++++++ 2 files changed, 31 insertions(+), 4 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 b07ed4a4de5..c349945879e 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 @@ -732,10 +732,13 @@ def _emit_tool_call( tool_call_id = content.call_id or flow.tool_call_id or generate_event_id() - # Why a started-set, not the last-id cursor: declaration-only calls replay after the - # function loop pauses; with two calls the cursor sits on the second, so the replay of - # the first (and then the second) re-emitted TOOL_CALL_START (#9638). - if content.name and tool_call_id not in flow.tool_calls_by_id: + # Why an open-call check, not the last-id cursor: declaration-only calls replay after + # the function loop pauses, and with two calls the cursor sits on the second, so the + # replay re-emitted TOOL_CALL_START for both (#9638). An id that already ENDED is not a + # replay: provider call_ids may be legally reused by a later distinct call (spec 004), + # which must START again with a fresh snapshot entry. + call_is_open = tool_call_id in flow.tool_calls_by_id and tool_call_id not in flow.tool_calls_ended + if content.name and not call_is_open: flow.tool_call_id = tool_call_id flow.tool_call_name = content.name if predictive_handler: @@ -757,6 +760,8 @@ def _emit_tool_call( } flow.pending_tool_calls.append(tool_entry) flow.tool_calls_by_id[tool_call_id] = tool_entry + # A reused provider id starting a fresh occurrence must be endable again. + flow.tool_calls_ended.discard(tool_call_id) elif tool_call_id: flow.tool_call_id = tool_call_id diff --git a/python/packages/ag-ui/tests/ag_ui/test_run_common.py b/python/packages/ag-ui/tests/ag_ui/test_run_common.py index de7c83206a5..43c4f8cca05 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_run_common.py +++ b/python/packages/ag-ui/tests/ag_ui/test_run_common.py @@ -1130,3 +1130,25 @@ def test_replayed_declaration_only_calls_start_each_tool_call_once() -> None: assert sorted(e.tool_call_id for e in starts) == ["call_a", "call_b"] ids = [entry["id"] for entry in flow.pending_tool_calls] assert sorted(ids) == ["call_a", "call_b"] + + +def test_reused_provider_call_id_starts_a_new_occurrence() -> None: + """A later distinct call reusing a completed provider call_id must START again (#9638).""" + flow = FlowState() + streamed = [ + Content.from_function_call(call_id="call_x", name="get_weather", arguments='{"city": "NYC"}'), + ] + for content in streamed: + _emit_content(content, flow) + _emit_content(Content.from_function_result(call_id="call_x", result="72F"), flow) + second = [ + Content.from_function_call(call_id="call_x", name="get_weather", arguments='{"city": "LA"}'), + ] + events = [] + for content in second: + events.extend(_emit_content(content, flow)) + + starts = [e for e in events if e.type == EventType.TOOL_CALL_START] + assert [e.tool_call_id for e in starts] == ["call_x"] + assert flow.tool_calls_by_id["call_x"]["function"]["arguments"] == '{"city": "LA"}' + assert "call_x" not in flow.tool_calls_ended From 17d6a6e646080675ca48a034d7a8fa74b494fef0 Mon Sep 17 00:00:00 2001 From: "yu.xia" Date: Wed, 4 Nov 2026 21:49:50 +0800 Subject: [PATCH 3/3] fix(ag-ui): end a reused call id exactly once, per the review MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The end-state check keyed only by call id: discarding the id when a provider reuses it reopens every snapshot entry with that id, including the occurrence that already ended, so the end-of-run loop emitted two TOOL_CALL_ENDs for one open call. get_pending_without_end() now reads from tool_calls_by_id — the latest occurrence per id — so a reused id ends exactly once. - _run_common.py: get_pending_without_end reads tool_calls_by_id - test_run.py: the hand-filled fixture populates tool_calls_by_id; a reused-id case counts the pending END entries (none while ended, one after the id is discarded) - test_run_common.py: the reused-id scenario asserts the end-of-run pending list holds exactly one entry; ToolCallStartEvent filters use isinstance so type checkers narrow the attribute; import order fixed --- .../ag-ui/agent_framework_ag_ui/_run_common.py | 10 ++++++++-- python/packages/ag-ui/tests/ag_ui/test_run.py | 18 ++++++++++++++++++ .../ag-ui/tests/ag_ui/test_run_common.py | 10 +++++++--- 3 files changed, 33 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 c349945879e..d2e43d442d9 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 @@ -562,8 +562,14 @@ def get_tool_name(self, call_id: str | None) -> str | None: return str(name) if name else None def get_pending_without_end(self) -> list[dict[str, Any]]: - """Get tool calls that started but never received an end event (declaration-only).""" - return [tc for tc in self.pending_tool_calls if tc.get("id") not in self.tool_calls_ended] + """Get tool calls that started but never received an end event (declaration-only). + + Read from ``tool_calls_by_id`` so a reused provider call id counts once: + an earlier occurrence that already ended must not be re-endable just + because a later occurrence reused the id, and the latest occurrence is + the entry end-of-run needs to close. + """ + return [tc for call_id, tc in self.tool_calls_by_id.items() if call_id not in self.tool_calls_ended] def _open_text_segment(flow: FlowState, message_id: str) -> dict[str, Any]: 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 e462e1dbbb3..2f552acb8ee 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_run.py +++ b/python/packages/ag-ui/tests/ag_ui/test_run.py @@ -451,12 +451,30 @@ def test_get_pending_without_end(self): {"id": "call_2", "function": {"name": "tool2"}}, {"id": "call_3", "function": {"name": "tool3"}}, ] + flow.tool_calls_by_id = {tc["id"]: tc for tc in flow.pending_tool_calls} flow.tool_calls_ended = {"call_1", "call_3"} result = flow.get_pending_without_end() assert len(result) == 1 assert result[0]["id"] == "call_2" + def test_get_pending_without_end_counts_a_reused_id_once(self): + """An earlier, ended occurrence must not become endable again when its id is reused.""" + flow = FlowState() + flow.pending_tool_calls = [ + {"id": "call_x", "function": {"name": "tool"}}, + {"id": "call_x", "function": {"name": "tool"}}, + ] + flow.tool_calls_by_id = {"call_x": flow.pending_tool_calls[-1]} + flow.tool_calls_ended = {"call_x"} + + assert flow.get_pending_without_end() == [] + + flow.tool_calls_ended.discard("call_x") + result = flow.get_pending_without_end() + assert len(result) == 1 + assert result[0]["id"] == "call_x" + class TestNormalizeResponseStream: """Tests for _normalize_response_stream helper.""" diff --git a/python/packages/ag-ui/tests/ag_ui/test_run_common.py b/python/packages/ag-ui/tests/ag_ui/test_run_common.py index 43c4f8cca05..cbf880912d2 100644 --- a/python/packages/ag-ui/tests/ag_ui/test_run_common.py +++ b/python/packages/ag-ui/tests/ag_ui/test_run_common.py @@ -13,6 +13,7 @@ ReasoningMessageContentEvent, ReasoningMessageStartEvent, ReasoningStartEvent, + ToolCallStartEvent, ) from agent_framework import Content, Message @@ -27,8 +28,8 @@ from agent_framework_ag_ui._run_common import ( FlowState, _build_run_finished_event, - _emit_content, _close_reasoning_block, + _emit_content, _emit_mcp_tool_result, _emit_text_reasoning, _emit_tool_result, @@ -1126,7 +1127,7 @@ def test_replayed_declaration_only_calls_start_each_tool_call_once() -> None: for content in [*streamed, *replayed]: events.extend(_emit_content(content, flow)) - starts = [e for e in events if e.type == EventType.TOOL_CALL_START] + starts = [e for e in events if isinstance(e, ToolCallStartEvent)] assert sorted(e.tool_call_id for e in starts) == ["call_a", "call_b"] ids = [entry["id"] for entry in flow.pending_tool_calls] assert sorted(ids) == ["call_a", "call_b"] @@ -1148,7 +1149,10 @@ def test_reused_provider_call_id_starts_a_new_occurrence() -> None: for content in second: events.extend(_emit_content(content, flow)) - starts = [e for e in events if e.type == EventType.TOOL_CALL_START] + starts = [e for e in events if isinstance(e, ToolCallStartEvent)] assert [e.tool_call_id for e in starts] == ["call_x"] assert flow.tool_calls_by_id["call_x"]["function"]["arguments"] == '{"city": "LA"}' assert "call_x" not in flow.tool_calls_ended + # The first occurrence already ended; only the reused, still-open one may + # receive an END from the end-of-run loop. + assert [tc["id"] for tc in flow.get_pending_without_end()] == ["call_x"]