Python: fix(python): preserve replayed approval calls

python/packages/core/agent_framework/_tools.py

@@ -2125,28 +2125,16 @@ def _replace_approval_contents_with_results(
     # Track which call_ids had their placeholders replaced
     placeholders_replaced: set[str] = set()
 
-    # Collect *pending* function call IDs across all messages to avoid duplicates. The
-    # function call and its approval request are frequently carried in separate messages
-    # (e.g. when a hosting layer replays them as separate items on an approval round trip),
-    # so scoping this per-message would let the same call_id be restored twice and leave
-    # the copy without a result unanswered.
-    #
-    # Calls that already carry a real result are excluded: reusing a call_id for a later
-    # invocation is supported, and a completed pair must not suppress the fresh request —
-    # that would drop the new call and attach its result to the old one. Placeholder
-    # results still count as pending, since the call they answer is the one being restored.
-    answered_call_ids = {
-        content.call_id
-        for msg in messages
-        for content in msg.contents
-        if content.type == "function_result" and content.call_id and not _is_approval_placeholder_result(content)
-    }
-    existing_call_ids = {
-        content.call_id
-        for msg in messages
-        for content in msg.contents
-        if content.type == "function_call" and content.call_id and content.call_id not in answered_call_ids
-    }
+    # Collect pending function calls in message order. A call_id can be reused after a
+    # result, so independently collecting all calls and all results loses whether a later
+    # call is still pending. Approval requests may be replayed beside that later call.
+    existing_call_ids: set[str] = set()
+    for msg in messages:
+        for content in msg.contents:
+            if content.type == "function_call" and content.call_id:
+                existing_call_ids.add(content.call_id)
+            elif content.type == "function_result" and content.call_id and not _is_approval_placeholder_result(content):
+                existing_call_ids.discard(content.call_id)
 
     for msg in messages:
         # Track approval requests that should be removed (duplicates)

python/packages/core/tests/core/test_function_invocation_logic.py

@@ -2374,6 +2374,40 @@ def test_replace_approval_contents_with_results_allows_reused_call_id_after_comp
         ("call_reused", "second output"),
     ]
 
+
+def test_replace_approval_contents_with_results_deduplicates_replayed_call_id_after_completion() -> None:
+    """Keep a replayed pending call when an earlier call with its id has completed."""
+    from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results
+
+    completed_call = Content.from_function_call(call_id="call_reused", name="run_skill_script", arguments="{}")
+    completed_result = Content.from_function_result(call_id="call_reused", result="first output")
+    replayed_call, request, response = _build_approved_tool_roundtrip(
+        call_id="call_reused", approval_id="approval_2", tool_name="run_skill_script"
+    )
+
+    messages = [
+        Message(role="assistant", contents=[completed_call]),
+        Message(role="tool", contents=[completed_result]),
+        Message(role="assistant", contents=[replayed_call]),
+        Message(role="assistant", contents=[request]),
+        Message(role="user", contents=[response]),
+    ]
+
+    _replace_approval_contents_with_results(
+        messages,
+        _collect_approval_responses(messages),
+        [Content.from_function_result(call_id="call_reused", result="second output")],
+    )
+
+    function_calls = [c for m in messages for c in m.contents if c.type == "function_call"]
+    assert [c.call_id for c in function_calls] == ["call_reused", "call_reused"]
+    results = [c for m in messages for c in m.contents if c.type == "function_result"]
+    assert [(c.call_id, c.result) for c in results] == [
+        ("call_reused", "first output"),
+        ("call_reused", "second output"),
+    ]
+
+
 def test_replace_approval_contents_with_results_uses_result_call_ids_for_placeholders() -> None:
     from agent_framework._tools import _collect_approval_responses, _replace_approval_contents_with_results