Python: fix(ag-ui): start each replayed declaration-only tool call once

python/packages/ag-ui/agent_framework_ag_ui/_run_common.py

@@ -732,7 +732,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 +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

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,46 @@ 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"]
+
+
+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