Python: Fix duplicate function call on approval round-trip (#7267)

python/packages/core/agent_framework/_tools.py

@@ -2096,6 +2096,12 @@ def _collect_approval_responses(
     return fcc_todo
 
 
+def _is_approval_placeholder_result(content: Content) -> bool:
+    """Whether a function_result is the stand-in emitted while approval is pending."""
+    result = getattr(content, "result", None)
+    return isinstance(result, str) and "[APPROVAL_PENDING]" in result
+
+
 def _replace_approval_contents_with_results(
     messages: list[Message],
     fcc_todo: dict[str, Content],
@@ -2119,12 +2125,30 @@ def _replace_approval_contents_with_results(
     # Track which call_ids had their placeholders replaced
     placeholders_replaced: set[str] = set()
 
-    for msg in messages:
-        # First pass - collect existing function call IDs to avoid duplicates
-        existing_call_ids = {
-            content.call_id for content in msg.contents if content.type == "function_call" and content.call_id
-        }
+    # 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
+    }
 
+    for msg in messages:
         # Track approval requests that should be removed (duplicates)
         contents_to_remove: list[int] = []
 
@@ -2140,6 +2164,8 @@ def _replace_approval_contents_with_results(
                 elif content.function_call is not None:
                     # Put back the function call content only if it doesn't exist
                     msg.contents[content_idx] = content.function_call
+                    if content.function_call.call_id:
+                        existing_call_ids.add(content.function_call.call_id)
             elif content.type == "function_approval_response":
                 # Skip hosted tool approvals — they must pass through to the API unchanged
                 if _is_hosted_tool_approval(content):
@@ -2169,12 +2195,7 @@ def _replace_approval_contents_with_results(
                     msg.role = "tool"
             elif content.type == "function_result":
                 # Check if this is a placeholder result that should be replaced
-                if (
-                    hasattr(content, "result")
-                    and isinstance(content.result, str)
-                    and "[APPROVAL_PENDING]" in content.result
-                    and content.call_id in result_by_call_id
-                ):
+                if _is_approval_placeholder_result(content) and content.call_id in result_by_call_id:
                     # Replace placeholder with actual result
                     msg.contents[content_idx] = result_by_call_id[content.call_id]
                     placeholders_replaced.add(content.call_id)

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

@@ -2338,6 +2338,42 @@ def test_replace_approval_contents_with_results_uses_result_call_ids_without_pla
     ]
 
 
+def test_replace_approval_contents_with_results_allows_reused_call_id_after_completion() -> None:
+    """A completed call must not suppress a later approval request that reuses its id.
+
+    Re-approving the same ``(call_id, function)`` is supported behaviour. If the dedupe
+    matched every occurrence of the id, the fresh request would be dropped and its result
+    attached to the already-answered call, leaving one call with two results.
+    """
+    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")
+    _, 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=[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