Python: fix(python): prevent local tool approvals from serializing as MCP responses

docs/specs/004-python-function-calling-loop.md

@@ -427,6 +427,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
   `ResponseStream.get_final_response()`.
 - The function invocation layer normalizes a private copy of caller messages. It must not mutate the caller's
   approval `Message`, approval `Content`, or an earlier returned response.
+
 - Approval-time `UserInputRequiredException` and `MiddlewareTermination` return immediately without another model
   call.
 
@@ -487,6 +488,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Function invocation disabled | The client bypasses the invocation loop without losing invocation kwargs. | `test_function_invocation_config_enabled_false`, `test_function_invocation_config_enabled_false_preserves_invocation_kwargs`, `test_streaming_function_invocation_config_enabled_false` |
 | Runtime tool changes | Added tools become available on the next iteration and retain approval behavior. | `test_add_tools_available_next_iteration`, `test_add_tools_with_approval_required_tool` |
 
+
 ### Approval pause and resume
 
 | Scenario | Required invariant | Primary regression test |
@@ -511,7 +513,8 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | Truthy non-boolean decision | Strings, integers, null, and other non-booleans do not authorize execution. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_treats_truthy_non_boolean_as_rejection`, `packages/core/tests/core/test_types.py::test_function_approval_response_deserialization_rejects_non_boolean_decisions`, `packages/ag-ui/tests/ag_ui/test_message_adapters.py::test_function_approval_requires_real_boolean`, `packages/ag-ui/tests/ag_ui/test_approval_result_event.py::test_resolve_approval_responses_treats_non_boolean_decision_as_rejection` |
 | Active batch replacement | A newly surfaced model batch replaces abandoned approval authority instead of growing session state. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_binding_replaces_abandoned_batch` |
 | Duplicate request id | Ambiguous request IDs within one active batch fail explicitly. | `packages/core/tests/core/test_function_invocation_logic.py::test_session_approval_batch_rejects_duplicate_request_ids` |
-| Tool registry changes | Same-name upgrades may execute the recorded operation; removing the recorded name executes nothing. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_allows_same_name_tool_upgrade`, `test_approval_resume_does_not_execute_when_recorded_tool_disappears` |
+
+| Tool registry changes | Same-name upgrades may execute the recorded operation; removing the recorded name executes nothing. | `packages/core/tests/core/test_harness_tool_approval.py::test_approval_resume_allows_same_name_tool_upgrade`, `test_approval_resume_does_not_execute_when_recorded_tool_disappears` | 
 
 ### Approval correlation and replay
 

python/packages/core/agent_framework/_sessions.py

@@ -29,7 +29,6 @@
 import weakref
 from abc import abstractmethod
 from base64 import urlsafe_b64encode
-from collections import deque
 from collections.abc import AsyncIterable, Awaitable, Callable, Generator, Iterable, Mapping, Sequence
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
@@ -876,46 +875,71 @@ def _is_approval_placeholder_result(content: Content) -> bool:
 
 
 def _approval_controls_to_keep(messages: Sequence[Message]) -> set[int]:
-    unresolved_requests_by_id: dict[str, Content] = {}
-    unresolved_local_responses_by_id: dict[str, Content] = {}
-    local_response_ids_by_call_id: dict[str, deque[str]] = {}
+    request_positions: dict[str, tuple[int, int]] = {}
+    response_ids: set[str] = set()
+    resolving_events: list[tuple[int, int, str]] = []
 
-    for message in messages:
-        for content in message.contents:
+    for msg_idx, message in enumerate(messages):
+        for content_idx, content in enumerate(message.contents):
             if content.type == "function_approval_request":
-                function_call = content.function_call
-                if content.id is not None and function_call is not None and function_call.call_id is not None:
-                    unresolved_requests_by_id.setdefault(content.id, content)
-                continue
-            if content.type == "function_approval_response":
-                function_call = content.function_call
                 if content.id is not None:
-                    unresolved_requests_by_id.pop(content.id, None)
-                if (
-                    content.id is not None
-                    and function_call is not None
-                    and function_call.call_id is not None
-                    and not function_call.additional_properties.get("server_label")
-                    and content.id not in unresolved_local_responses_by_id
-                ):
-                    unresolved_local_responses_by_id[content.id] = content
-                    local_response_ids_by_call_id.setdefault(function_call.call_id, deque()).append(content.id)
-                continue
-            if content.call_id is None:
-                continue
-            is_terminal_result = content.type == "function_result" and not _is_approval_placeholder_result(content)
-            is_follow_up_request = content.user_input_request and content.type not in {
-                "function_approval_request",
-                "function_approval_response",
-            }
-            if not (is_terminal_result or is_follow_up_request):
-                continue
-            if response_ids := local_response_ids_by_call_id.get(content.call_id):
-                unresolved_local_responses_by_id.pop(response_ids.popleft(), None)
+                    request_positions[content.id] = (msg_idx, content_idx)
+            elif content.type == "function_approval_response":
+                if content.id is not None:
+                    response_ids.add(content.id)
+            elif content.call_id is not None:
+                is_terminal_result = content.type == "function_result" and not _is_approval_placeholder_result(content)
+                is_follow_up_request = content.user_input_request and content.type not in {
+                    "function_approval_request",
+                    "function_approval_response",
+                }
+                if is_terminal_result or is_follow_up_request:
+                    resolving_events.append((msg_idx, content_idx, content.call_id))
+
+    keep_ids: set[int] = set()
+    seen_request_ids: set[str] = set()
+
+    for msg_idx, message in enumerate(messages):
+        for content_idx, content in enumerate(message.contents):
+            if content.type == "function_approval_request":
+                if content.id is None or content.function_call is None or content.function_call.call_id is None:
+                    continue
+                if content.id in seen_request_ids:
+                    continue
+
+                req_pos = (msg_idx, content_idx)
+                call_id = content.function_call.call_id
+
+                is_resolved = content.id in response_ids
+                if not is_resolved:
+                    for res_msg_idx, res_content_idx, res_call_id in resolving_events:
+                        if res_call_id == call_id and (res_msg_idx, res_content_idx) >= req_pos:
+                            is_resolved = True
+                            break
+                if not is_resolved:
+                    keep_ids.add(id(content))
+                    seen_request_ids.add(content.id)
+
+            elif content.type == "function_approval_response":
+                function_call = content.function_call
+                if content.id is None or function_call is None or function_call.call_id is None:
+                    continue
+                if function_call.additional_properties.get("server_label"):
+                    continue
+
+                call_id = function_call.call_id
+                resp_pos = (msg_idx, content_idx)
+                ref_pos = request_positions.get(content.id, resp_pos)
+
+                is_resolved = False
+                for res_msg_idx, res_content_idx, res_call_id in resolving_events:
+                    if res_call_id == call_id and (res_msg_idx, res_content_idx) >= ref_pos:
+                        is_resolved = True
+                        break
+                if not is_resolved:
+                    keep_ids.add(id(content))
 
-    return {
-        id(content) for content in (*unresolved_requests_by_id.values(), *unresolved_local_responses_by_id.values())
-    }
+    return keep_ids
 
 
 def _filter_approval_control_messages(messages: Sequence[Message]) -> list[Message]: