feat(runner): allow multiple simultaneous approval requests in one turn

Remove the one-approval-per-turn latch (PendingApprovalLatch) so each gated tool
call in a turn emits its own approval card; pluralize the parked-approval record
(a Map keyed by tool-call id) and the warm keep-alive resume input (a list of
decisions), settling each parked gate by its own tool-call id on a live resume;
defer orphan-sibling settling to the post-drain sweep so a concurrent gate is not
force-settled before its own permission request arrives. Add runner unit tests
(two cards emit, neither force-settled; multi-gate warm park + resume;
deny-one-approve-one; mixed/unresumable sets stay cold), SDK vercel adapter plural
egress/ingress tests, and a playground all-settled resume test.

Closes #5373.

Claude-Session: https://claude.ai/code/session_01DnWRxU3dCJ11hgDidm26vW

Author: mmabrouk

sdks/python/oss/tests/pytest/unit/agents/adapters/test_vercel_stream_park.py

@@ -84,6 +84,87 @@ async def test_parked_run_emits_approval_then_finish() -> None:
     assert finish["finishReason"] == "other"
 
 
+def _two_concurrent_gates_run() -> AgentStream:
+    """One turn that raises TWO approval gates at once (concurrent approvals): two distinct tool
+    calls, each followed by its own ``user_approval`` request, then the terminal ``paused`` result.
+    The egress must surface one ``tool-approval-request`` per gate, each keyed to its own tool-call
+    id, and still drain to a single clean finish."""
+    return AgentStream(
+        _records(
+            [
+                {
+                    "kind": "event",
+                    "event": {
+                        "type": "tool_call",
+                        "id": "tool-1",
+                        "name": "github__get_user",
+                        "input": {},
+                    },
+                },
+                {
+                    "kind": "event",
+                    "event": {
+                        "type": "tool_call",
+                        "id": "tool-2",
+                        "name": "github__list_repos",
+                        "input": {},
+                    },
+                },
+                {
+                    "kind": "event",
+                    "event": {
+                        "type": "interaction_request",
+                        "id": "perm-1",
+                        "kind": "user_approval",
+                        "payload": {"toolCallId": "tool-1"},
+                    },
+                },
+                {
+                    "kind": "event",
+                    "event": {
+                        "type": "interaction_request",
+                        "id": "perm-2",
+                        "kind": "user_approval",
+                        "payload": {"toolCallId": "tool-2"},
+                    },
+                },
+                {"kind": "event", "event": {"type": "done", "stopReason": "paused"}},
+                {
+                    "kind": "result",
+                    "result": {
+                        "ok": True,
+                        "output": "",
+                        "stopReason": "paused",
+                        "sessionId": "conv-1",
+                        "traceId": "trace-1",
+                    },
+                },
+            ]
+        )
+    )
+
+
+@pytest.mark.asyncio
+async def test_concurrent_gates_emit_one_approval_request_each() -> None:
+    """Two ``user_approval`` events in one turn yield TWO ``tool-approval-request`` frames — one
+    per event — each carrying its own approvalId and tool-call id, so the FE can prompt for both
+    gates independently. Pins the plural side of the single-gate contract above."""
+    parts = [
+        part async for part in agent_run_to_vercel_parts(_two_concurrent_gates_run())
+    ]
+
+    approvals = [p for p in parts if p["type"] == "tool-approval-request"]
+    assert len(approvals) == 2
+    # One frame per event, each keyed to its own gate (perm/tool pair), order preserved.
+    assert [(a["approvalId"], a["toolCallId"]) for a in approvals] == [
+        ("perm-1", "tool-1"),
+        ("perm-2", "tool-2"),
+    ]
+    # Both gates share the turn but the stream still drains to a single clean finish (F-040).
+    assert parts[-1]["type"] == "finish"
+    assert [p.get("type") for p in parts].count("finish") == 1
+
+
 def _parked_run_with_real_args() -> AgentStream:
     """A parked turn where the runner first surfaced the tool call with EMPTY input, then the
     approval request carries the REAL args on ``payload.toolCall.rawInput`` (the cold-replay

sdks/python/oss/tests/pytest/unit/agents/test_ui_messages.py

@@ -240,6 +240,47 @@ def test_inline_approval_responded_approve_emits_approved_envelope(self):
         assert result.tool_call_id == "call_2"
         assert result.output == {"approved": True}
 
+    def test_concurrent_inline_approvals_each_emit_their_own_result(self):
+        # Concurrent approvals: the browser round-trips TWO tool parts in one turn, each with its
+        # own inline decision (`state: approval-responded`, `approval.approved`) — one approve, one
+        # deny. The ingress must emit one `{approved}` tool_result per part, keyed by its own
+        # toolCallId, preserving each decision independently (neither gate clobbers the other).
+        msgs = vercel_ui_messages_to_messages(
+            [
+                {
+                    "id": "m10",
+                    "role": "assistant",
+                    "parts": [
+                        {
+                            "type": "tool-deleteFile",
+                            "toolCallId": "call_1",
+                            "state": "approval-responded",
+                            "input": {"path": "/a"},
+                            "approval": {"id": "perm_1", "approved": True},
+                        },
+                        {
+                            "type": "tool-writeFile",
+                            "toolCallId": "call_2",
+                            "state": "approval-responded",
+                            "input": {"path": "/b"},
+                            "approval": {"id": "perm_2", "approved": False},
+                        },
+                    ],
+                }
+            ]
+        )
+        results = [b for b in msgs[0].content if b.type == "tool_result"]
+        assert len(results) == 2
+        # Each decision lands on its own toolCallId, in order, approve and deny preserved.
+        assert (results[0].tool_call_id, results[0].output) == (
+            "call_1",
+            {"approved": True},
+        )
+        assert (results[1].tool_call_id, results[1].output) == (
+            "call_2",
+            {"approved": False},
+        )
+
     def test_inline_output_denied_state_emits_denied_envelope(self):
         # The AI SDK terminal deny state has no `approval.approved` flag; `output-denied`
         # itself means denied, so the ingress still emits `{approved: False}`.

services/runner/src/engines/sandbox_agent/acp-interactions.ts

@@ -11,7 +11,6 @@ import {
 } from "../../responder.ts";
 import {
   piBuiltinIdentity,
-  PendingApprovalLatch,
   type GateDescriptor,
 } from "../../permission-plan.ts";
 import {
@@ -44,7 +43,6 @@ export interface AttachPermissionResponderInput {
     markToolCallDenied?: (toolCallId: string | undefined) => void;
   };
   responder: Responder;
-  latch: PendingApprovalLatch;
   serverPermissions?: ReadonlyMap<string, ToolPermission>;
   /**
    * Called when a gate pauses the turn. The orchestration loop uses this to end the turn
@@ -54,6 +52,13 @@ export interface AttachPermissionResponderInput {
   log?: (msg: string) => void;
   /** Called with the ACP tool-call id when a gate pauses the turn. */
   onPausedToolCall?: (id: string) => void;
+  /**
+   * Called when a NON-parkable pause happens this turn (a client-tool ACP gate, which cannot be
+   * answered across a turn boundary on the live session). The keep-alive dispatch reads this to
+   * keep a turn that mixes an approval gate with a client-tool pause on the cold path, since only
+   * the cold path can multiplex that mixed set. An approval gate does NOT fire it.
+   */
+  onNonParkablePause?: () => void;
   /** Called on pause to record the pending gate as an interaction (fire-and-forget). */
   onCreateInteraction?: (
     token: string,
@@ -65,10 +70,10 @@ export interface AttachPermissionResponderInput {
   onResolveInteraction?: (token: string) => void;
   /**
    * Fires for EVERY parkable permission gate (a Claude ACP gate or a Pi ACP gate) that
-   * resolves to pendingApproval, BEFORE the single-pause latch. Keep-alive uses it to record
-   * the parked permission id / tool-call id (for a live resume via `respondPermission`) and to
-   * count how many gates are pending this turn (a multi-gate pause does not park). It never
-   * fires for a client-tool gate (`pauseClientTool`), which stays on the cold path.
+   * resolves to pendingApproval. Keep-alive uses it to record every parked permission id /
+   * tool-call id (for a live resume via `respondPermission`, keyed by tool-call id) and to count
+   * how many gates are pending this turn. It never fires for a client-tool gate
+   * (`pauseClientTool`), which fires `onNonParkablePause` instead and stays on the cold path.
    */
   onUserApprovalGate?: (info: {
     permissionId: string;
@@ -109,11 +114,11 @@ export function attachPermissionResponder({
   session,
   run,
   responder,
-  latch,
   serverPermissions = new Map(),
   onPause,
   log,
   onPausedToolCall,
+  onNonParkablePause,
   onCreateInteraction,
   onResolveInteraction,
   onUserApprovalGate,
@@ -155,19 +160,18 @@ export function attachPermissionResponder({
     gate: GateDescriptor,
     gateType: ParkedApprovalGateType,
   ): void => {
-    // Signal the parkable gate BEFORE the latch so a keep-alive resume can record the pending
-    // permission id and the multi-gate detector counts every pending gate (not just the first).
-    const gateToolCallId = stringValue(req?.toolCall?.toolCallId);
+    // Signal the parkable gate so a keep-alive resume can record the pending permission id and
+    // count every pending gate. Each gate emits its own card: there is no per-turn cap, so N
+    // gated calls in one turn all render and all park (the plural approval path).
+    const toolCallId = stringValue(req?.toolCall?.toolCallId);
     onUserApprovalGate?.({
       permissionId: id,
-      toolCallId: gateToolCallId ?? "",
+      toolCallId: toolCallId ?? "",
       toolName: gate.toolName,
       args: gate.args,
-      interactionToken: interactionEventId(id, gateToolCallId),
+      interactionToken: interactionEventId(id, toolCallId),
       gateType,
     });
-    if (!latch.tryAcquire()) return;
-    const toolCallId = stringValue(req?.toolCall?.toolCallId);
     const eventId = interactionEventId(id, toolCallId);
     if (toolCallId) onPausedToolCall?.(toolCallId);
     run.emitEvent({
@@ -192,7 +196,9 @@ export function attachPermissionResponder({
     gate: GateDescriptor,
     spec: ResolvedToolSpec,
   ): void => {
-    if (!latch.tryAcquire()) return;
+    // A client-tool ACP pause cannot be answered on the live session across a turn boundary, so
+    // flag the turn non-parkable: a turn that mixes this with an approval gate stays cold.
+    onNonParkablePause?.();
     const toolCallId = stringValue(req?.toolCall?.toolCallId);
     const eventId = interactionEventId(id, toolCallId);
     if (toolCallId) onPausedToolCall?.(toolCallId);

services/runner/src/engines/sandbox_agent/client-tools.ts

@@ -11,7 +11,7 @@
  *     `tools/call` handler (`tools/tool-mcp-http.ts`).
  *
  * Both consume the `ClientToolRelay` built here, so the pause decision (the responder's verdict
- * ladder), the single `client_tool` payload shape, the pause-latch bookkeeping, and the
+ * ladder), the single `client_tool` payload shape, the pause bookkeeping, and the
  * ACP-tool-call correlation live in ONE place instead of being re-derived per path.
  */
 import type { AgentEvent, RenderHint } from "../../protocol.ts";
@@ -200,11 +200,6 @@ export function emitClientToolInteraction(
   });
 }
 
-/** The one-pause-per-turn latch surface the seam needs (see `PendingApprovalLatch`). */
-interface LatchLike {
-  tryAcquire(): boolean;
-}
-
 /** The pause-controller surface the seam needs (see `PendingApprovalPauseController`). */
 interface PauseLike {
   markPausedToolCall(toolCallId: string): void;
@@ -214,12 +209,14 @@ interface PauseLike {
 export interface BuildClientToolRelayInput {
   responder: Responder;
   run: EmitRun;
-  /** The approval-dialog latch. Client tools do NOT gate on it — each pending client tool parks
-   *  its own widget, and only the approval path (`acp-interactions.ts`) enforces one dialog per
-   *  turn. Kept in the shared input so the wiring stays symmetric; ignored here. */
-  latch?: LatchLike;
   /** The turn-ender: `pause()` cancels the prompt; `markPausedToolCall` suppresses late frames. */
   pause: PauseLike;
+  /**
+   * Flag the turn non-parkable when a browser-fulfilled client tool pauses: it cannot be answered
+   * on the live session across a turn boundary, so a turn that mixes it with an approval gate must
+   * stay on the cold path. Fires once per pending client tool (idempotent counter on the caller).
+   */
+  onNonParkablePause?: () => void;
   /** Seeds the durable interactions plane for the pending call (fire-and-forget). */
   recordPendingInteraction: (
     token: string,
@@ -247,6 +244,7 @@ export function buildClientToolRelay({
   pause,
   recordPendingInteraction,
   toolCallIndex,
+  onNonParkablePause,
   log = () => {},
 }: BuildClientToolRelayInput): ClientToolRelay {
   return {
@@ -281,12 +279,15 @@ export function buildClientToolRelay({
       const correlatedId =
         toolCallIndex?.lookup(request.toolName, request.input) ??
         request.toolCallId;
-      // Every pending client tool parks its OWN widget. Unlike an approval DIALOG (one at a time,
-      // gated by the latch), browser-fulfilled client tools are independent surfaces — an agent may
-      // request several connections in one turn, and each must render. The turn still ends exactly
-      // once: `pause.pause()` (via `onPause`) is idempotent, so N emits pause the turn once, and
-      // `markPausedToolCall` keeps each parked call from being force-settled as a deferred sibling.
+      // Every pending client tool parks its OWN widget: browser-fulfilled client tools are
+      // independent surfaces — an agent may request several connections in one turn, and each must
+      // render. Approval gates now behave the same way (no per-turn cap). The turn still ends
+      // exactly once: `pause.pause()` (via `onPause`) is idempotent, so N emits pause the turn
+      // once, and `markPausedToolCall` keeps each parked call from being force-settled as a sibling.
       pause.markPausedToolCall(correlatedId);
+      // A client-tool pause keeps the whole turn on the cold path: the warm resume cannot answer
+      // a browser-fulfilled call, so a turn that also holds an approval gate must not park.
+      onNonParkablePause?.();
       emitClientToolInteraction(run, {
         id: request.id,
         toolCallId: correlatedId,

services/runner/src/engines/sandbox_agent/environment-setup.ts

@@ -347,8 +347,10 @@ export async function prepareEnvironmentSetup(
     closeToolMcp: undefined,
     currentTurn: undefined,
     lastTurnToolCallIds: [],
+    parkedApprovals: new Map(),
     parkedApproval: undefined,
     approvalGateCount: 0,
+    nonParkablePauseCount: 0,
     destroyed: false,
     destroy: async () => {},
     clearTurn: () => {},

services/runner/src/engines/sandbox_agent/runtime-contracts.ts

@@ -156,8 +156,14 @@ export interface RunTurnOptions {
    * keep-alive turn.
    */
   approvalParkMode?: boolean;
-  /** A live approval resume: answer the parked gate and stream the continued prompt's events. */
-  resume?: ResumeApprovalInput;
+  /**
+   * A live approval resume: answer every parked gate of the turn (keyed by tool-call id) and
+   * stream the continued prompt's events. A turn can hold more than one parked gate, so the
+   * resume carries a LIST — the server builds it from the inbound approval replies, one decision
+   * per parked gate. All decisions share the one held prompt promise (there is one prompt per
+   * turn); `runTurn` answers each gate then awaits that single continuation.
+   */
+  resume?: { decisions: ResumeApprovalInput[] };
 }
 
 /**
@@ -222,17 +228,31 @@ export interface SessionEnvironment {
    */
   lastTurnToolCallIds: string[];
   /**
-   * The Claude ACP permission gate the LAST turn paused on, or undefined. Set only for a harness
-   * ACP permission gate, reset at each turn start; the dispatch reads it after a paused turn to
-   * decide whether to park in `awaiting_approval` and, on the next request, how to resume.
+   * Every parkable ACP permission gate the LAST turn paused on, keyed by the gated tool-call id
+   * (reset at each turn start). This is the source of truth the warm resume iterates: a turn can
+   * hold more than one gate (parallel gated tool calls), and each is answered by its own
+   * `permissionId` on the live session. Empty when no parkable gate paused the turn.
+   */
+  parkedApprovals: Map<string, ParkedApproval>;
+  /**
+   * The FIRST parked gate this turn, a convenience for per-turn-uniform reads (logging, the
+   * gate-type check, the shared history/credential validation). Undefined when the map is empty.
+   * The multi-answer resume and the all-parkable park check read `parkedApprovals`, not this.
    */
   parkedApproval?: ParkedApproval;
   /**
-   * How many Claude ACP permission gates resolved to pendingApproval THIS turn (reset at turn
-   * start). More than one means parallel gates the single-gate resume cannot answer, so the
-   * dispatch does not park (tears down cold as today).
+   * How many ACP permission gates resolved to pendingApproval THIS turn (reset at turn start).
+   * Equals `parkedApprovals.size` when every gate carried a resumable tool-call id; a larger
+   * count means a gate lacked an id and cannot be resumed live, so the dispatch stays cold.
    */
   approvalGateCount: number;
+  /**
+   * How many NON-parkable pauses happened this turn (a client-tool ACP gate or a browser-fulfilled
+   * relay/MCP client tool), reset at turn start. Non-zero means the turn mixes an unanswerable
+   * client-tool pause into the set, so the whole turn stays on the cold path (only cold can
+   * multiplex a mixed set today).
+   */
+  nonParkablePauseCount: number;
   destroyed: boolean;
   /** Complete, idempotent teardown selected from the typed teardown reason. */
   destroy: (opts?: { reason?: TeardownReason }) => Promise<void>;

services/runner/src/engines/sandbox_agent/runtime-policy.ts

@@ -39,7 +39,24 @@ export function shouldSuppressPausedToolCallUpdate(
   if (kind !== "tool_call" && kind !== "tool_call_update") return false;
   const toolCallId =
     typeof frame?.toolCallId === "string" ? frame.toolCallId : undefined;
-  return pause.isPausedToolCall(toolCallId);
+  // F-024: a paused (gated) tool call's later frames are teardown artifacts and never reach the
+  // stream.
+  if (pause.isPausedToolCall(toolCallId)) return true;
+  // Once the turn is pausing, a FAILED update for any other open call is a managed-cancel
+  // artifact (a sibling that will not run this turn), so suppress it and let the deterministic
+  // post-drain sweep settle that call as TOOL_NOT_EXECUTED_PAUSED — the human sees "paused", not a
+  // spurious cancellation error. A `completed` update is NOT suppressed: an auto-allowed sibling
+  // that legitimately finishes while another gate holds the (kept-alive) warm session must show
+  // its real result, not be overwritten as paused. Announcements (`tool_call`) are never
+  // suppressed: the call must be recorded so the sweep knows which calls to settle. (Removing the
+  // eager first-pause settle is what lets concurrent gates each mark their own call paused before
+  // the sweep runs; this narrow suppression only masks the cancel artifact.) Residual: a genuine
+  // mid-pause tool error also arrives as `failed` and reads as paused here — acceptable, since the
+  // paused result invites a retry and the runner has no adapter provenance to tell the two apart.
+  if (kind === "tool_call_update" && pause.active && frame?.status === "failed") {
+    return true;
+  }
+  return false;
 }
 
 const CLAUDE_STRICT_DEPLOYMENTS = new Set([