Python: Harden function calling loop and approval replay

python/AGENTS.md

@@ -6,6 +6,8 @@ Instructions for AI coding agents working in the Python codebase.
 - [DEV_SETUP.md](DEV_SETUP.md) - Development environment setup and available poe tasks
 - [CODING_STANDARD.md](CODING_STANDARD.md) - Coding standards, docstring format, and performance guidelines
 - [samples/SAMPLE_GUIDELINES.md](samples/SAMPLE_GUIDELINES.md) - Sample structure and guidelines
+- [Python function-calling loop specification](../docs/specs/004-python-function-calling-loop.md) - Required
+  behavior, scenario-to-test mapping, remaining coverage gap, and extra validation for function-loop changes
 
 **Agent Skills** (`.github/skills/`) — detailed, task-specific instructions loaded on demand:
 - `python-development` — coding standards, type annotations, docstrings, logging, performance
@@ -48,6 +50,16 @@ When preparing a PR description:
 
 Run `uv run poe` from the `python/` directory to see available commands. See [DEV_SETUP.md](DEV_SETUP.md) for detailed usage.
 
+## Function-Calling Loop Changes
+
+Changes to the Python function-calling loop, approval resume behavior, function-call history, provider
+serialization, or transport result handling must follow
+[the function-calling loop specification](../docs/specs/004-python-function-calling-loop.md). This area requires
+extra validation because small changes can duplicate side effects, orphan call/result pairs, replay stale approval
+authority, or make streaming and non-streaming behavior diverge. Update the specification and its scenario-to-test
+mapping whenever coverage or behavior changes. External contributors must check with the Agent Framework core team
+before picking up issues in this area.
+
 ## Project Structure
 
 ```

python/packages/ag-ui/AGENTS.md

@@ -29,6 +29,8 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
 - Multimodal user inputs support both legacy (`text`, `binary`) and draft-style (`image`, `audio`, `video`, `document`) shapes.
 - Interrupted runs complete with `RUN_FINISHED.outcome.type == "interrupt"` and canonical `outcome.interrupts`; do not document or add new flows that depend on the legacy top-level `RUN_FINISHED.interrupt` field.
 - `Interrupt` and `ResumeEntry` come from the `ag-ui-protocol` package (`ag_ui.core`), not from an Agent Framework-specific interrupt model.
+- Approval-time execution preserves each call's complete result group. Follow-up user-input requests remain in the
+  resumed messages, while `TOOL_CALL_RESULT` events are emitted only for terminal `function_result` contents.
 - SSE keepalive is endpoint-owned transport behavior configured through
   `add_agent_framework_fastapi_endpoint(keepalive_seconds=...)`. It emits SSE comments only; do not add `PING`,
   `HEARTBEAT`, or `KEEPALIVE` AG-UI events, and do not add runner-level keepalive settings.

python/packages/ag-ui/agent_framework_ag_ui/_agent_run.py

@@ -40,9 +40,10 @@
 from agent_framework._tools import (
     _ALREADY_APPROVED_APPROVAL_REQUEST_GROUPS_KEY,  # type: ignore
     _collect_approval_responses,  # type: ignore
+    _get_tool_map,  # type: ignore
     _replace_approval_contents_with_results,  # type: ignore
     _TOOL_APPROVAL_STATE_KEY,  # type: ignore
-    _try_execute_function_calls,  # type: ignore
+    _try_execute_function_call_groups,  # type: ignore
     normalize_function_invocation_configuration,
 )
 from agent_framework._types import ResponseStream
@@ -1306,10 +1307,19 @@ async def _resolve_approval_responses(
         approved_responses = validated
         rejected_responses = validated_rejected
 
-    approved_function_results: list[Any] = []
+    approved_function_result_groups: list[list[Content]] = []
 
-    # Execute approved tool calls
-    if approved_responses and tools:
+    # Partition approved responses into static (execute now) and deferred (execute during run)
+    tool_map = _get_tool_map(tools) if tools else {}
+    static_approved: list[Content] = []
+
+    for approval in approved_responses:
+        tool_name = approval.function_call.name if approval.function_call else None
+        if tool_name in tool_map:
+            static_approved.append(approval)
+
+    # Execute only statically-available approved tool calls
+    if static_approved and tools:
         client = getattr(agent, "client", None)
         config = normalize_function_invocation_configuration(getattr(client, "function_invocation_configuration", None))
         middleware_pipeline = FunctionMiddlewarePipeline(
@@ -1319,36 +1329,31 @@ async def _resolve_approval_responses(
         # Filter out AG-UI-specific kwargs that should not be passed to tool execution
         tool_kwargs = {k: v for k, v in run_kwargs.items() if k != "options"}
         try:
-            results, _ = await _try_execute_function_calls(
+            approved_function_result_groups, _ = await _try_execute_function_call_groups(
                 custom_args=tool_kwargs,
-                attempt_idx=0,
-                function_calls=approved_responses,
+                function_calls=static_approved,
                 tools=tools,
                 middleware_pipeline=middleware_pipeline,
                 config=config,
             )
-            approved_function_results = list(results)
         except Exception as e:
             logger.exception("Failed to execute approved tool calls; injecting error results: %s", e)
-            approved_function_results = []
+            approved_function_result_groups = []
 
-    # Build results for approved responses (used for TOOL_CALL_RESULT event emission)
+    # Normalize one group per static approval and collect only terminal results for TOOL_CALL_RESULT events.
+    # Deferred provider-injected approvals are left in messages for ToolApprovalMiddleware to process.
+    replacement_groups: list[list[Content]] = []
     approved_results: list[Content] = []
-    for idx, approval in enumerate(approved_responses):
-        if (
-            idx < len(approved_function_results)
-            and getattr(approved_function_results[idx], "type", None) == "function_result"
-        ):
-            approved_results.append(approved_function_results[idx])
-            continue
-        # Get call_id from function_call if present, otherwise use approval.id
-        func_call = approval.function_call
-        call_id = (func_call.call_id if func_call else None) or approval.id or ""
-        approved_results.append(
-            Content.from_function_result(call_id=call_id, result="Error: Tool call invocation failed.")
-        )
+    for idx, approval in enumerate(static_approved):
+        result_group = approved_function_result_groups[idx] if idx < len(approved_function_result_groups) else []
+        if not result_group:
+            func_call = approval.function_call
+            call_id = (func_call.call_id if func_call else None) or approval.id or ""
+            result_group = [Content.from_function_result(call_id=call_id, result="Error: Tool call invocation failed.")]
+        replacement_groups.append(result_group)
+        approved_results.extend(content for content in result_group if content.type == "function_result")
 
-    _replace_approval_contents_with_results(messages, fcc_todo, approved_results)
+    _replace_approval_contents_with_results(messages, fcc_todo, replacement_groups)
 
     # Post-process: Convert user messages with function_result content to proper tool messages.
     # After _replace_approval_contents_with_results, approved tool calls have their results
@@ -1401,6 +1406,39 @@ def _convert_approval_results_to_tool_messages(messages: list[Message]) -> None:
     messages[:] = result
 
 
+def _confirm_changes_target_call_id(
+    snapshot_messages: list[dict[str, Any]],
+    confirm_call_id: str,
+    approval_payload: Mapping[str, Any],
+) -> str | None:
+    explicit_call_id = approval_payload.get("function_call_id")
+    if explicit_call_id:
+        return str(explicit_call_id)
+
+    for snapshot_message in snapshot_messages:
+        if normalize_agui_role(snapshot_message.get("role", "")) != "assistant":
+            continue
+        tool_calls = snapshot_message.get("tool_calls") or snapshot_message.get("toolCalls")
+        if not isinstance(tool_calls, list):
+            continue
+        for tool_call in tool_calls:
+            if not isinstance(tool_call, Mapping) or str(tool_call.get("id") or "") != confirm_call_id:
+                continue
+            function = tool_call.get("function")
+            if not isinstance(function, Mapping) or function.get("name") != "confirm_changes":
+                return None
+            arguments = function.get("arguments")
+            if isinstance(arguments, str):
+                try:
+                    arguments = json.loads(arguments)
+                except json.JSONDecodeError:
+                    return None
+            if isinstance(arguments, Mapping) and arguments.get("function_call_id"):
+                return str(arguments["function_call_id"])
+            return None
+    return None
+
+
 def _clean_resolved_approvals_from_snapshot(
     snapshot_messages: list[dict[str, Any]],
     resolved_messages: list[Message],
@@ -1434,9 +1472,6 @@ def _clean_resolved_approvals_from_snapshot(
                 )
                 result_by_call_id[str(content.call_id)] = result_text
 
-    if not result_by_call_id:
-        return
-
     for snap_msg in snapshot_messages:
         if normalize_agui_role(snap_msg.get("role", "")) != "tool":
             continue
@@ -1455,12 +1490,23 @@ def _clean_resolved_approvals_from_snapshot(
         # Find matching tool result by toolCallId
         tool_call_id = snap_msg.get("toolCallId") or snap_msg.get("tool_call_id") or ""
         replacement = result_by_call_id.get(str(tool_call_id))
-        if replacement is not None:
-            snap_msg["content"] = replacement
-            logger.info(
-                "Replaced approval payload in snapshot for tool_call_id=%s with actual result",
-                tool_call_id,
+        if replacement is None:
+            target_call_id = _confirm_changes_target_call_id(
+                snapshot_messages,
+                str(tool_call_id),
+                parsed,
             )
+            if target_call_id is None:
+                continue
+            if parsed.get("accepted"):
+                replacement = result_by_call_id.get(target_call_id, "Changes confirmed and applied successfully.")
+            else:
+                replacement = "Changes declined."
+        snap_msg["content"] = replacement
+        logger.info(
+            "Replaced approval payload in snapshot for tool_call_id=%s with resolved content",
+            tool_call_id,
+        )
 
 
 def _snapshot_tool_call_ids(message: Mapping[str, Any]) -> list[str]:

python/packages/ag-ui/tests/ag_ui/test_approval_result_event.py

@@ -524,6 +524,90 @@ async def test_resolve_approval_responses_returns_only_approved() -> None:
     assert "rejected" in str(rejection_results[0].result).lower()
 
 
+async def test_resolve_approval_responses_preserves_follow_up_user_input_group() -> None:
+    """Approval-time follow-up requests stay grouped and do not emit a synthetic tool result."""
+    from agent_framework import Message
+    from agent_framework.exceptions import UserInputRequiredException
+
+    from agent_framework_ag_ui._agent_run import _resolve_approval_responses
+
+    def request_consent() -> str:
+        raise UserInputRequiredException(
+            contents=[
+                Content.from_oauth_consent_request(consent_link="https://example.com/consent-1"),
+                Content.from_oauth_consent_request(consent_link="https://example.com/consent-2"),
+            ]
+        )
+
+    consent_tool = FunctionTool(
+        name="request_consent",
+        description="Request two consent steps",
+        func=request_consent,
+        approval_mode="always_require",
+    )
+    function_call = Content.from_function_call(call_id="call_consent", name="request_consent", arguments="{}")
+    approval_request = Content.from_function_approval_request(id="approval_consent", function_call=function_call)
+    messages: list[Any] = [
+        Message(role="assistant", contents=[approval_request]),
+        Message(role="user", contents=[approval_request.to_function_approval_response(approved=True)]),
+    ]
+    agent = StubAgent(updates=[], default_options={"tools": [consent_tool]})
+
+    results = await _resolve_approval_responses(messages, [consent_tool], agent, {})
+
+    follow_up_requests = [content for message in messages for content in message.contents if content.user_input_request]
+    assert results == []
+    assert [request.consent_link for request in follow_up_requests] == [
+        "https://example.com/consent-1",
+        "https://example.com/consent-2",
+    ]
+    assert not [content for message in messages for content in message.contents if content.type == "function_result"]
+    assert any(message.role == "assistant" and message.contents == follow_up_requests for message in messages)
+
+
+async def test_resolve_approval_responses_returns_failure_when_grouped_execution_raises(
+    monkeypatch: Any,
+) -> None:
+    """A grouped-execution failure produces one deterministic result for the approved call."""
+    from agent_framework import Message
+
+    from agent_framework_ag_ui._agent_run import _resolve_approval_responses
+
+    async def fail_grouped_execution(**kwargs: Any) -> tuple[list[list[Content]], bool]:
+        del kwargs
+        raise RuntimeError("execution failed")
+
+    monkeypatch.setattr(
+        "agent_framework_ag_ui._agent_run._try_execute_function_call_groups",
+        fail_grouped_execution,
+    )
+    weather_tool = _make_weather_tool()
+    function_call = Content.from_function_call(
+        call_id="call_execution_failure",
+        name="get_weather",
+        arguments='{"city": "Seattle"}',
+    )
+    approval_request = Content.from_function_approval_request(
+        id="approval_execution_failure",
+        function_call=function_call,
+    )
+    messages: list[Any] = [
+        Message(role="assistant", contents=[approval_request]),
+        Message(role="user", contents=[approval_request.to_function_approval_response(approved=True)]),
+    ]
+    agent = StubAgent(updates=[], default_options={"tools": [weather_tool]})
+
+    results = await _resolve_approval_responses(messages, [weather_tool], agent, {})
+
+    assert len(results) == 1
+    assert results[0].type == "function_result"
+    assert results[0].call_id == "call_execution_failure"
+    assert results[0].result == "Error: Tool call invocation failed."
+    assert [
+        content.result for message in messages for content in message.contents if content.type == "function_result"
+    ] == ["Error: Tool call invocation failed."]
+
+
 class TestApprovalToolResultDisplayChannel:
     """Approved tools using ``state_update(..., tool_result=...)`` must route the
     display payload to the UI event while ``flow.tool_results`` still receives