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

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

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."""

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"]