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..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]: @@ -732,7 +738,13 @@ 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 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: @@ -754,6 +766,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.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 69822014420..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 @@ -28,6 +29,7 @@ FlowState, _build_run_finished_event, _close_reasoning_block, + _emit_content, _emit_mcp_tool_result, _emit_text_reasoning, _emit_tool_result, @@ -1108,3 +1110,49 @@ 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 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"] + + +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 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"]