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/__init__.pyi

@@ -601,9 +601,9 @@ __all__ = [
     "WorkflowEventType",
     "WorkflowException",
     "WorkflowExecutor",
+    "WorkflowInvocationKwargs",
     "WorkflowMessage",
     "WorkflowRunResult",
-    "WorkflowInvocationKwargs",
     "WorkflowRunState",
     "WorkflowRunnerException",
     "WorkflowValidationError",

python/packages/core/agent_framework/_sessions.py

@@ -27,7 +27,6 @@
 import warnings
 import weakref
 from abc import abstractmethod
-from collections import deque
 from collections.abc import AsyncIterable, Awaitable, Callable, Generator, Iterable, Mapping, Sequence
 from contextvars import ContextVar, Token
 from dataclasses import dataclass
@@ -44,12 +43,12 @@
 )
 from ._middleware import ChatContext, ChatMiddleware
 from ._telemetry import FeatureIndex, mark_feature_used
+from ._tools import _index_approval_occurrences  # pyright: ignore[reportPrivateUsage]
 from ._types import (
     AgentResponse,
     AgentRunInputs,
     ChatResponse,
     ChatResponseUpdate,
-    Content,
     Message,
     ResponseStream,
     _build_agent_response_from_chat_response,  # pyright: ignore[reportPrivateUsage]
@@ -829,53 +828,45 @@ async def after_run(
         """
 
 
-def _is_approval_placeholder_result(content: Content) -> bool:
-    result = getattr(content, "result", None)
-    return isinstance(result, str) and "[APPROVAL_PENDING]" in result
-
-
 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]] = {}
+    idx = _index_approval_occurrences(messages)
+    keep_ids: set[int] = set()
 
-    for message in messages:
-        for content in 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)
+    response_ids = {r.content.id for r in idx.responses if r.content.id is not None}
 
-    return {
-        id(content) for content in (*unresolved_requests_by_id.values(), *unresolved_local_responses_by_id.values())
+    request_pos_by_id: dict[str, tuple[int, int]] = {
+        req.content.id: (req.msg_idx, req.content_idx) for req in idx.requests if req.content.id is not None
     }
 
+    seen_request_ids: set[str] = set()
+    for req in idx.requests:
+        if req.content.id is None or req.content.id in seen_request_ids:
+            continue
+        req_pos = (req.msg_idx, req.content_idx)
+        is_resolved = req.content.id in response_ids
+        if not is_resolved:
+            is_resolved = any(
+                (r.msg_idx, r.content_idx) >= req_pos and r.call_id == req.call_id
+                for r in (*idx.terminal_results, *idx.follow_ups)
+            )
+        if not is_resolved:
+            keep_ids.add(id(req.content))
+            seen_request_ids.add(req.content.id)
+
+    for resp in idx.responses:
+        if resp.content.id is None:
+            continue
+        resp_pos = (resp.msg_idx, resp.content_idx)
+        ref_pos = request_pos_by_id.get(resp.content.id, resp_pos)
+        is_resolved = any(
+            (r.msg_idx, r.content_idx) >= ref_pos and r.call_id == resp.call_id
+            for r in (*idx.terminal_results, *idx.follow_ups)
+        )
+        if not is_resolved:
+            keep_ids.add(id(resp.content))
+
+    return keep_ids
+
 
 def _filter_approval_control_messages(messages: Sequence[Message]) -> list[Message]:
     """Remove resolved approval controls while preserving pending occurrences."""