Python: Improve function approval resume and 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, coverage gaps, 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

@@ -42,7 +42,7 @@
     _collect_approval_responses,  # 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,7 +1306,7 @@ 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:
@@ -1319,36 +1319,30 @@ 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,
                 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 approval and collect only terminal results for TOOL_CALL_RESULT events.
+    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.")
-        )
-
-    _replace_approval_contents_with_results(messages, fcc_todo, approved_results)
+        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, 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

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

python/packages/core/AGENTS.md

@@ -145,6 +145,27 @@ agent_framework/
   available, approval requests for known non-approval-required tools are treated as already approved, hidden, stored
   in session state keyed to the visible approval request ids from that batch, and reinjected only when that visible
   approval flow resumes.
+- Approval resume is an immutable response boundary: the function invocation layer normalizes a private copy of
+  caller messages, returns approved and rejected terminal results in the resumed response (and stream) before any
+  final assistant message, and does not mutate the caller's approval `Message` or the earlier approval-request
+  response.
+- Approval/result correlation is occurrence-aware. A `call_id` may be reused after a completed round, so approval
+  normalization matches ordered call occurrences and consumes approved results per occurrence rather than using one
+  global result per `call_id`. All contents produced by one execution remain one result group and are consumed
+  together, including multiple user-input requests.
+- Approval resume keeps terminal `function_result` contents in tool-role messages and follow-up user-input requests
+  in assistant-role messages, including mixed sibling batches.
+- Function-call budget accounting counts one unit per executed result group, not per emitted `function_result`, so
+  executions that pause for user input still consume `max_function_calls`.
+- `function_approval_request` and `function_approval_response` are control-plane contents. History providers may
+  retain them in their backing store for audit. The base `HistoryProvider.before_run` filters resolved wrappers from
+  later model replay, but preserves unresolved requests/responses until a terminal result or follow-up request closes
+  the occurrence. On unrelated turns the function layer omits the complete pending call batch from model input while
+  retaining it for a later approval response. Providers configured with `load_messages=False` do not replay history,
+  so this filter is intentionally not invoked.
+- Reasoning content or opaque reasoning metadata bound to a function call is part of the same logical group as the
+  call and terminal result. Function-loop replay and compaction must preserve that group atomically; adapters should
+  fail before a stateless request when required reasoning cannot be reconstructed.
 ### Agent Loop (`_harness/_loop.py`)
 
 - **`AgentLoopMiddleware`** - `AgentMiddleware` that re-runs an agent in a loop by calling `call_next()` repeatedly (the pipeline re-reads `context.messages` each time). One configurable class covers two patterns: a required user `should_continue` predicate (sync or async, the first positional/keyword arg), and a chat-client judge built via the `.with_judge(...)` factory (a second chat client decides whether the original request was answered; loops while it is *not*, using a `JudgeVerdict` structured-output response — internally just an async `should_continue` predicate). The constructor covers the predicate pattern directly; only the judge has a convenience classmethod factory (`.with_judge(judge_client, ...)`) that forwards to `__init__`. Supports both streaming and non-streaming runs. By default a non-streaming run returns an aggregated `AgentResponse` containing every iteration's messages plus the injected `next_message` "nudge" messages (as `user` messages); set `return_final_only=True` to return only the last iteration's response. Streaming runs always yield each iteration's updates and emit the injected nudge messages as `user` updates between iterations (the `return_final_only` flag has no effect on streaming, and the final response reflects the last iteration; `MiddlewareTermination` is handled cleanly). `should_continue` is required; other constructor args are optional: `max_iterations` (safety cap; defaults to `DEFAULT_MAX_ITERATIONS`=10, explicit `None`→unbounded, positive int caps; `.with_judge` uses `DEFAULT_JUDGE_MAX_ITERATIONS`=5 as its default), `next_message` (defaults to a short "continue" nudge), `return_final_only`, and `additional_instructions` (an extra `system` message injected ahead of the input before the agent runs — becomes part of the original messages so it survives `fresh_context` resets and persists via a session). The judge is configured only through `.with_judge` (`judge_client`/`instructions`/`criteria`), not the constructor, and its `reasoning` is fed back to the agent as the next iteration's input; the judge forwards the original request messages and the agent's latest response messages verbatim so multi-modal content is preserved. `criteria` (a `list[str]`) is both injected as the agent's `additional_instructions` and rendered into the judge instructions wherever the `{{criteria}}` placeholder (`CRITERIA_PLACEHOLDER`) appears (`DEFAULT_JUDGE_INSTRUCTIONS` ends with it; custom `instructions` may include it, and it is stripped when no criteria are given). The `should_continue`/`next_message` callables are invoked with keyword args (`iteration`, `last_result`, `messages`, `original_messages`, `session`, `agent`, `progress`, `feedback`) and may be sync or async; declare only what you need plus `**kwargs`. `should_continue` may return a plain `bool` or a `(bool, str | None)` tuple whose second item is feedback surfaced to `next_message`/`record_feedback` via the `feedback` kwarg (the judge uses this to relay its `reasoning`). Stop precedence per iteration is `max_iterations` → `should_continue`, evaluated before `record_feedback` so the feedback is available to it.