Python: Execute every approved tool in a parallel AG-UI HITL batch

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

@@ -491,6 +491,7 @@ that manually replay messages own the equivalent rule: do not resend an approval
 | AG-UI provider boundary | Completed local approval controls from AG-UI request and snapshot replay are absent from raw chat-client input while deferred and hosted approvals keep their respective in-run/provider paths. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_does_not_forward_resolved_local_approval_control_to_chat_client`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_deferred_provider_tool_executes`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_canonical_resume_preserves_hosted_approval_for_provider`, `packages/ag-ui/tests/ag_ui/test_run.py::test_filter_local_approval_responses_for_provider_removes_duplicate_completed_controls`, `packages/ag-ui/tests/ag_ui/test_run.py::test_filter_local_approval_responses_for_provider_pairs_reused_call_ids_by_occurrence`, `packages/ag-ui/tests/ag_ui/test_run.py::test_canonical_hosted_approval_resume_rejects_edited_arguments_without_mutating_pending` |
 | AG-UI standard approval payload | Agent and workflow tool approvals emit canonical `tool_call` interrupts. `approved` plus full-replacement `editedArgs` executes once and replays idempotently, while legacy `accepted` plus direct partial edits remains supported. Hosted approvals remain decision-only. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_entry_applies_standard_full_replacement_edited_args`, `test_endpoint_agent_approval_replayed_standard_edited_resume_is_idempotent`, `test_endpoint_agent_approval_resume_entry_applies_edited_arguments`, `test_workflow_endpoint_emits_canonical_tool_approval_interrupt`, `test_workflow_endpoint_accepts_canonical_tool_approval_resume`, `test_workflow_endpoint_applies_canonical_approval_edited_args`, `test_workflow_endpoint_accepts_legacy_partial_approval_edits`, `test_workflow_endpoint_hosted_approval_rejects_argument_edits` |
 | AG-UI cancellation | A cancelled interrupt executes zero times and completes normally, including an identical retry during retained cancellation state; resolved siblings in the same complete resume still execute once. Workflow cancellation clears both runner correlation and the owning agent executor's pending request so later approvals remain resumable. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_cancelled_resume_entry_completes_without_execution`, `test_endpoint_agent_approval_replayed_cancellation_completes_idempotently`, `test_endpoint_agent_approval_mixed_cancelled_and_resolved_resume_executes_resolved_tool`, `test_endpoint_workflow_request_info_cancelled_resume_completes_normally`, `test_workflow_endpoint_cancelled_agent_approval_does_not_block_next_approval` |
+| AG-UI parallel gated batch resume | Consecutive approval-control user messages for one assistant batch keep leftover gated call ids open. Every accepted call executes once, mixed reject/accept reaches the resolver without skip injection, and ID-less approval-control messages for distinct call ids are not collapsed. A later reject or edit that reuses the same approval and call ids is kept until occurrence matching. Real follow-up user text still skips leftover siblings. | `packages/ag-ui/tests/ag_ui/test_message_adapters.py::test_sanitize_consecutive_approval_controls_keep_gated_siblings`, `test_sanitize_consecutive_approval_controls_keep_mixed_decisions`, `test_sanitize_approval_then_followup_still_skips_leftover_sibling`, `test_deduplicate_idless_approval_controls_for_distinct_calls`, `test_deduplicate_idless_approval_controls_keep_later_reject_for_reused_ids`, `test_deduplicate_idless_approval_controls_keep_later_edit_for_reused_ids`, `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_executes_every_gated_call_in_parallel_batch` |
 | AG-UI approval retention and capacity | Pending authority expires automatically, indeterminate outcomes remain non-retryable until their safety window permits reclamation, and one trusted scope cannot consume another scope's occurrence quota. | `packages/ag-ui/tests/ag_ui/test_approval_lifecycle.py::test_abandoned_pending_occurrence_expires_and_releases_capacity`, `test_indeterminate_occurrence_is_reclaimed_after_its_safety_window`, `test_capacity_is_enforced_per_trusted_scope` |
 | AG-UI local executor unavailable on resume | A claimed local occurrence whose executor disappeared releases its unstarted claim, reports temporary unavailability, and remains safely retryable. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_agent_approval_resume_remains_retryable_when_local_tool_is_temporarily_unavailable` |
 | AG-UI forwarded execution interruption | A provider failure, cancellation, or stream close after forwarding an approval recovers the open occurrence as indeterminate when no idempotency key proves retry safety. | `packages/ag-ui/tests/ag_ui/test_endpoint.py::test_endpoint_hosted_approval_becomes_indeterminate_when_provider_stream_fails` |
@@ -596,6 +597,7 @@ Before accepting an update, reviewers must confirm:
 - #7267 / #7271 and #7304 — replayed calls and reused ids
 - #7043 — provider-injected approval execution
 - #6828 — AG-UI `confirm_changes` snapshot correlation
+- #7569 — AG-UI parallel gated approval resume executes every approved call
 - #7212 — non-adjacent and reused-id compaction integrity
 - #7125 — service-side approval response serialization
 - #7045 — post-limit tool-content transcript integrity

python/packages/ag-ui/AGENTS.md

@@ -31,8 +31,12 @@ AG-UI protocol integration for building agent UIs with the AG-UI standard.
 - `Interrupt` and `ResumeEntry` come from the `ag-ui-protocol` package (`ag_ui.core`), not from an Agent Framework-specific interrupt model.
 - Tool approval interrupts, including approvals surfaced through workflow `request_info`, advertise standard
   `approved` and full-replacement `editedArgs` responses while retaining the existing `accepted` alias and direct
-  partial edits for MAF client compatibility. A `cancelled` resume completes normally without executing that call;
-  resolved siblings in the same complete resume still proceed.
+  partial edits for MAF client compatibility. A `cancelled` resume completes normally without executing that
+  call; resolved siblings in the same complete resume still proceed. Consecutive approval-control user messages
+  for a parallel `always_require` batch keep leftover gated call ids open; they are not treated as follow-up that
+  skip remaining siblings. ID-less approval-control messages are keyed by approval
+  id, call id, decision, and function-call payload so a later reject or edit that
+  reuses those ids is not collapsed before occurrence matching.
 - 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.
 - Approval responses for tools injected during `before_run` are deferred to the in-run approval middleware rather

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

@@ -8,6 +8,7 @@
 import binascii
 import json
 import logging
+from collections.abc import Mapping
 from typing import Any, cast
 
 from agent_framework import (
@@ -75,6 +76,69 @@ def _function_result_call_ids(messages: list[Message]) -> set[str]:
     return result_ids
 
 
+def _is_approval_control_user_message(msg: Message) -> bool:
+    """Whether a user message is only tool-approval controls, not a new turn.
+
+    Resume synthesis emits one user message per interrupt. A later sibling
+    approval in the same batch must still consume its own call id; it is not a
+    follow-up that abandons remaining gated calls.
+    """
+    has_approval = False
+    for content in msg.contents or []:
+        if content.type == "function_approval_response":
+            has_approval = True
+            continue
+        if content.type == "text" and (content.text or "").strip():
+            return False
+        if content.type not in {"function_approval_response", "text"}:
+            return False
+    return has_approval
+
+
+def _function_call_payload_identity(function_call: Content | None) -> tuple[str | None, str]:
+    """Canonical name plus arguments for one approval-bound function call."""
+    if function_call is None:
+        return None, ""
+    name = function_call.name
+    arguments = function_call.arguments
+    if isinstance(arguments, Mapping):
+        payload = json.dumps(dict(arguments), sort_keys=True, default=str)
+    elif arguments is None:
+        payload = ""
+    else:
+        payload = str(arguments)
+    return (str(name) if name else None, payload)
+
+
+def _approval_control_identity(
+    msg: Message,
+) -> tuple[tuple[str | None, str | None, bool | None, str | None, str], ...] | None:
+    """Stable identity for an approval-control user message.
+
+    ID-less approval messages hashing only ``str(Content)`` can collapse two
+    distinct call occurrences from the same assistant batch. Dedup must also
+    keep a later reject or edit that reuses the same approval and call ids;
+    ``(content.id, call_id)`` alone would drop it before occurrence matching.
+    """
+    contents = msg.contents or []
+    if not contents or any(content.type != "function_approval_response" for content in contents):
+        return None
+    identities: list[tuple[str | None, str | None, bool | None, str | None, str]] = []
+    for content in contents:
+        call_id = content.function_call.call_id if content.function_call is not None else None
+        name, arguments = _function_call_payload_identity(content.function_call)
+        identities.append(
+            (
+                content.id,
+                str(call_id) if call_id else None,
+                content.approved,
+                name,
+                arguments,
+            )
+        )
+    return tuple(identities)
+
+
 def _sanitize_tool_history(
     messages: list[Message],
     *,
@@ -217,7 +281,7 @@ def _sanitize_tool_history(
                 except (json.JSONDecodeError, KeyError) as exc:
                     logger.debug(f"Could not parse user message as confirm_changes response: {type(exc).__name__}")
 
-            if pending_tool_call_ids:
+            if pending_tool_call_ids and not _is_approval_control_user_message(msg):
                 logger.info(
                     f"User message arrived with {len(pending_tool_call_ids)} pending tool calls - "
                     "injecting synthetic results"
@@ -342,8 +406,12 @@ def _deduplicate_messages(messages: list[Message]) -> list[Message]:
             if msg.message_id:
                 key = ("id", msg.message_id)
             else:
-                content_str = str([str(c) for c in msg.contents]) if msg.contents else ""
-                key = ("content", role_value, hash(content_str))
+                approval_identity = _approval_control_identity(msg)
+                if approval_identity is not None:
+                    key = ("approval", approval_identity)
+                else:
+                    content_str = str([str(c) for c in msg.contents]) if msg.contents else ""
+                    key = ("content", role_value, hash(content_str))
 
             if key in seen_keys:
                 logger.info(f"Skipping duplicate message at index {idx}: role={role_value}")

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

@@ -2049,6 +2049,63 @@ async def stream_fn(
     return TestClient(app), executed, messages_received, state
 
 
+def _build_parallel_gated_approval_batch_endpoint(
+    streaming_chat_client_stub: Any,
+    *,
+    snapshot_store: InMemoryAGUIThreadSnapshotStore | None = None,
+) -> tuple[TestClient, list[str], list[Message], dict[str, str]]:
+    executed: list[str] = []
+    messages_received: list[Message] = []
+    state = {"phase": "pause"}
+
+    def first_tool() -> str:
+        executed.append("first")
+        return "first result"
+
+    def second_tool() -> str:
+        executed.append("second")
+        return "second result"
+
+    async def stream_fn(
+        messages: list[Message],
+        options: dict[str, Any],
+        **kwargs: Any,
+    ) -> AsyncIterator[ChatResponseUpdate]:
+        del options, kwargs
+        if state["phase"] == "pause":
+            yield ChatResponseUpdate(
+                contents=[
+                    Content.from_function_call(call_id="call_first", name="first_tool", arguments="{}"),
+                    Content.from_function_call(call_id="call_second", name="second_tool", arguments="{}"),
+                ],
+                role="assistant",
+            )
+            return
+        messages_received[:] = list(messages)
+        yield ChatResponseUpdate(contents=[Content.from_text(text="Done.")], role="assistant")
+
+    agent = Agent(
+        name="test_agent",
+        instructions="Test",
+        client=streaming_chat_client_stub(stream_fn),
+        tools=[
+            FunctionTool(name="first_tool", description="First tool", func=first_tool, approval_mode="always_require"),
+            FunctionTool(
+                name="second_tool", description="Second tool", func=second_tool, approval_mode="always_require"
+            ),
+        ],
+    )
+    app = FastAPI()
+    add_agent_framework_fastapi_endpoint(
+        app,
+        AgentFrameworkAgent(agent=agent, require_confirmation=False),
+        path="/approval",
+        snapshot_store=snapshot_store,
+        snapshot_scope_resolver=(lambda _request: "tenant-a") if snapshot_store is not None else None,
+    )
+    return TestClient(app), executed, messages_received, state
+
+
 def _build_tool_approval_queue_endpoint(
     streaming_chat_client_stub: Any,
 ) -> tuple[TestClient, list[str], list[Message], dict[str, str], AgentFrameworkAgent]:
@@ -2286,6 +2343,75 @@ async def test_endpoint_agent_approval_resume_releases_already_approved_sibling(
     assert sorted(replayed_call_ids) == ["call_sensitive", "call_weather"]
 
 
+async def test_endpoint_agent_approval_resume_executes_every_gated_call_in_parallel_batch(streaming_chat_client_stub):
+    """A complete resume for a parallel gated batch must execute every approved call."""
+    client, executed, messages_received, state = _build_parallel_gated_approval_batch_endpoint(
+        streaming_chat_client_stub,
+        snapshot_store=InMemoryAGUIThreadSnapshotStore(),
+    )
+
+    pause_response = client.post(
+        "/approval",
+        json={
+            "runId": "run-pause",
+            "threadId": "thread-parallel-gated",
+            "messages": [{"id": "user-1", "role": "user", "content": "Run both tools"}],
+        },
+    )
+
+    assert pause_response.status_code == 200
+    pause_events = _decode_sse_events(pause_response)
+    pause_finished = [event for event in pause_events if event.get("type") == "RUN_FINISHED"]
+    interrupts = _run_finished_interrupts(pause_finished[-1])
+    assert [interrupt["id"] for interrupt in interrupts] == ["call_first", "call_second"]
+    assert not [event for event in pause_events if event.get("type") == "TOOL_CALL_RESULT"]
+
+    state["phase"] = "resume"
+    resume_response = client.post(
+        "/approval",
+        json={
+            "runId": "run-resume",
+            "threadId": "thread-parallel-gated",
+            "messages": [],
+            "resume": [
+                {"interruptId": "call_first", "status": "resolved", "payload": {"accepted": True}},
+                {"interruptId": "call_second", "status": "resolved", "payload": {"accepted": True}},
+            ],
+        },
+    )
+
+    assert resume_response.status_code == 200
+    resume_events = _decode_sse_events(resume_response)
+    tool_results = [
+        (event["toolCallId"], event["content"]) for event in resume_events if event.get("type") == "TOOL_CALL_RESULT"
+    ]
+    assert tool_results == [("call_first", "first result"), ("call_second", "second result")]
+    assert executed == ["first", "second"]
+    assert not any("Tool execution skipped" in str(event.get("content")) for event in resume_events)
+
+    hydrate_response = client.post(
+        "/approval",
+        json={"runId": "run-hydrate", "threadId": "thread-parallel-gated", "messages": []},
+    )
+    assert hydrate_response.status_code == 200
+    hydrated_messages = _latest_messages_snapshot(hydrate_response)
+    replayed_results = [
+        (message.get("toolCallId"), message.get("content"))
+        for message in hydrated_messages
+        if message.get("role") == "tool" and message.get("toolCallId") in {"call_first", "call_second"}
+    ]
+    assert replayed_results == [("call_first", "first result"), ("call_second", "second result")]
+    assert not any("Tool execution skipped" in str(message.get("content")) for message in hydrated_messages)
+
+    replayed_provider_results = [
+        content for message in messages_received for content in message.contents if content.type == "function_result"
+    ]
+    assert [(content.call_id, content.result) for content in replayed_provider_results] == [
+        ("call_first", "first result"),
+        ("call_second", "second result"),
+    ]
+
+
 async def test_endpoint_agent_approval_resume_persists_replayable_tool_results(streaming_chat_client_stub):
     """Approved batches should hydrate with real results under original tool call ids."""
     client, executed, messages_received, state = _build_mixed_approval_batch_endpoint(