Merge pull request #1519 from corbitsdev/cl-10852-deliver-approvals-for-parked-calls-in-queue-drained-turns

fix(tui): deliver approvals for parked calls in queue-drained turns

Author: TheGreatAxios

src/permission/reactor-authorize.ts

@@ -31,6 +31,7 @@ import {
   type WorkerGrantStore,
 } from "./worker-grant.js";
 import { canonicalToolName } from "../agent/canonical-tool-name.js";
+import { prepareDispatchedToolCall } from "../agent/tool-aliases.js";
 import { getSubAgentIdentity } from "../subagent/identity-context.js";
 import {
   FLEET_VERBS,
@@ -100,6 +101,23 @@ function resolveWorkerCwd(cwd: string | undefined): string {
   return cwd ?? getSubAgentIdentity()?.cwd ?? process.cwd();
 }
 
+/**
+ * The call as DynamicToolRunner dispatches it: canonical engine name plus
+ * hidden-alias argument coercion (gpt `shell` -> run_shell via
+ * coerceShellArgs). Authorize sees the raw wire call; the middleware that
+ * consumes an authorized retry sees the dispatched call. Keying both on the
+ * dispatched shape keeps one fingerprint per executed call. Invalid hidden
+ * arguments never dispatch, so they keep their raw shape.
+ */
+function dispatchedCall(call: ToolCallType): ToolCallType {
+  const engine = canonicalToolName(call.name);
+  try {
+    return prepareDispatchedToolCall(call, engine);
+  } catch {
+    return { ...call, name: engine };
+  }
+}
+
 function workerCallIdentity(
   sessionId: string,
   call: ToolCallType,
@@ -110,10 +128,11 @@ function workerCallIdentity(
   args: Record<string, unknown>;
   cwd: string;
 } {
+  const dispatched = dispatchedCall(call);
   return {
     sessionId,
-    canonicalTool: canonicalToolName(call.name),
-    args: (call.arguments ?? {}) as Record<string, unknown>,
+    canonicalTool: dispatched.name,
+    args: (dispatched.arguments ?? {}) as Record<string, unknown>,
     cwd,
   };
 }

src/permission/worker-grant-dispatch-identity.test.ts

@@ -0,0 +1,147 @@
+// gpt-family workers call run_shell as `shell`. Reactor authz sees that raw
+// wire call; the permission middleware sees DynamicToolRunner's
+// prepareDispatchedToolCall output, whose coerceShellArgs rebuilds the
+// arguments. The authorized retry must match the same envelope at both
+// stages.
+import { describe, expect, test } from "bun:test";
+import type { ToolCall } from "@intx/types/runtime";
+
+import { prepareDispatchedToolCall } from "../agent/tool-aliases.js";
+import { gateToolCall } from "../plugins/permission-plugin.js";
+import { createPermissionGate } from "./gate.js";
+import { workerPermissionGate } from "./reactor-authorize.js";
+import { WorkerGrantStore } from "./worker-grant.js";
+
+const CWD = process.cwd();
+
+// The call shape a gpt-family worker sends for run_shell.
+const wireCall = (id: string): ToolCall => ({
+  id,
+  name: "shell",
+  arguments: {
+    command: "git status --short",
+    timeout: 120000,
+    cwd: CWD,
+    background: false,
+    stop: "",
+  },
+});
+
+function extractRequestId(reason: string): string {
+  const match = /grant request ([0-9a-f-]{36})/.exec(reason);
+  if (match?.[1] === undefined) throw new Error(`no grant id: ${reason}`);
+  return match[1];
+}
+
+describe("gpt-family worker authorized retry identity", () => {
+  test("approved retry of a `shell` call executes once instead of minting a new grant request", async () => {
+    // Parent: reactor-gated interactive gate, nothing pre-approved, no auto.
+    const parentGate = createPermissionGate({
+      approvals: [],
+      cwd: CWD,
+      skipPermissions: false,
+      interactive: true,
+      reactorGated: true,
+      auto: false,
+      requestApproval: async () => ({ allow: false }),
+    });
+    const store = new WorkerGrantStore();
+    const workerGate = workerPermissionGate(parentGate, {
+      sessionId: "worker-gpt",
+      store,
+    });
+
+    // 1. First attempt: reactor authz denies and records envelope A.
+    const denied = await workerGate.authorizeCall(wireCall("call-1"));
+    if (denied.effect !== "deny") throw new Error("expected worker deny");
+    const requestId = extractRequestId(denied.reason);
+    const envelope = store.peek(requestId);
+    expect(envelope?.canonicalTool).toBe("run_shell");
+
+    // 2. approve_worker_call: parent approved the frozen call (allow-once,
+    // no session grant) and flipped the envelope to authorized.
+    expect(store.authorizeRetry(requestId)).toBe(true);
+
+    // 3. Retry, stage 1: reactor authz on the raw wire call allows via
+    // hasAuthorizedRetry (does not consume).
+    const retryAuthz = await workerGate.authorizeCall(wireCall("call-2"));
+    expect(retryAuthz).toEqual({ effect: "allow" });
+    expect(store.peek(requestId)?.status).toBe("authorized");
+
+    // 4. Retry, stage 2: DynamicToolRunner coerces the `shell` call onto
+    // run_shell, then the posix middleware runs gateToolCall.
+    const dispatched = prepareDispatchedToolCall(
+      wireCall("call-2"),
+      "run_shell",
+    );
+    expect(dispatched.arguments).not.toEqual(wireCall("x").arguments);
+
+    let executed = 0;
+    const result = await gateToolCall(
+      workerGate,
+      dispatched,
+      new AbortController().signal,
+      async (call) => {
+        executed++;
+        return { callId: call.id, content: "ran" };
+      },
+    );
+
+    expect(result.isError).not.toBe(true);
+    expect(executed).toBe(1);
+    expect(store.peek(requestId)?.status).toBe("consumed");
+
+    // One grant, one execution: a replay of the same call fails closed at
+    // both stages, and no second envelope was minted.
+    const replay = await gateToolCall(
+      workerGate,
+      prepareDispatchedToolCall(wireCall("call-3"), "run_shell"),
+      new AbortController().signal,
+      async (call) => {
+        executed++;
+        return { callId: call.id, content: "ran" };
+      },
+    );
+    expect(replay.isError).toBe(true);
+    expect(String(replay.content)).toContain("already consumed");
+    expect((await workerGate.authorizeCall(wireCall("call-4"))).effect).toBe(
+      "deny",
+    );
+    expect(executed).toBe(1);
+  });
+
+  test("an altered retry (different command) still fails closed", async () => {
+    const parentGate = createPermissionGate({
+      approvals: [],
+      cwd: CWD,
+      skipPermissions: false,
+      interactive: true,
+      reactorGated: true,
+      auto: false,
+      requestApproval: async () => ({ allow: false }),
+    });
+    const store = new WorkerGrantStore();
+    const workerGate = workerPermissionGate(parentGate, {
+      sessionId: "worker-gpt-alt",
+      store,
+    });
+    const denied = await workerGate.authorizeCall(wireCall("call-1"));
+    if (denied.effect !== "deny") throw new Error("expected worker deny");
+    const requestId = extractRequestId(denied.reason);
+    expect(store.authorizeRetry(requestId)).toBe(true);
+
+    const altered: ToolCall = {
+      ...wireCall("call-2"),
+      arguments: { ...wireCall("x").arguments, command: "git push --force" },
+    };
+    expect((await workerGate.authorizeCall(altered)).effect).toBe("deny");
+    const result = await gateToolCall(
+      workerGate,
+      prepareDispatchedToolCall(altered, "run_shell"),
+      new AbortController().signal,
+      async (call) => ({ callId: call.id, content: "ran" }),
+    );
+    expect(result.isError).toBe(true);
+    expect(store.peek(requestId)?.status).toBe("authorized");
+  });
+});

src/session/approval-resume.test.ts

@@ -12,6 +12,7 @@ import type {
   InboundMessage,
 } from "@intx/types/runtime";
 
+import { APPROVAL_TIMEOUT_RESULT_TEXT } from "../permission/decline-markers.js";
 import type { PermissionGate } from "../permission/gate.js";
 import { createExtraDeniedPathMatcher } from "../plugins/secret-guard-plugin.js";
 import {
@@ -23,6 +24,7 @@ import {
   resolveSuspendedApprovalFromStore,
 } from "./approval-resume.js";
 import { createSessionOperationQueue } from "../tui/delivery-queue.js";
+import { createParkedOverlayAbortBinding } from "../tui/runner/parked-overlay-abort.js";
 import { defined } from "../../testkit/defined.js";
 import { redactedCorrelationId } from "./redacted-correlation.js";
 import {
@@ -936,6 +938,78 @@ describe("approval identity ordering", () => {
     expect(decisionBody(message).outcome).toBe("rejected");
   });
 
+  test("generation change during catch history drops the parked call once", async () => {
+    const ctx = deferredResume();
+    const failure = new Error("gate exploded");
+    ctx.resolveSuspended.mockImplementation(async () => {
+      throw failure;
+    });
+    ctx.history.mockResolvedValueOnce([]).mockImplementationOnce(async () => {
+      ctx.invalidate();
+      return [];
+    });
+    const handling = ctx.resume.handle(suspension("corr-A", "echo same"));
+    ctx.lookup.resolve("call-A");
+    await expect(handling).rejects.toBe(failure);
+    expect(ctx.onDropped).toHaveBeenCalledTimes(1);
+    expect(ctx.onDropped).toHaveBeenCalledWith(APPROVAL_DROPPED_NOTICE);
+    expect(ctx.deliver).toHaveBeenCalledTimes(1);
+    const message = ctx.deliver.mock.calls[0]?.[0];
+    if (message === undefined)
+      throw new Error("expected cancellation rejection");
+    expect(deliveredCorrelationId(message)).toBe("corr-A");
+    expect(decisionBody(message)).toEqual({
+      outcome: "rejected",
+      message: APPROVAL_DROPPED_NOTICE,
+    });
+  });
+
+  test("a gate throw with no generation change rejects once as approval surface failed", async () => {
+    const ctx = deferredResume();
+    const failure = new Error("gate exploded");
+    ctx.resolveSuspended.mockImplementation(async () => {
+      throw failure;
+    });
+    const handling = ctx.resume.handle(suspension("corr-A", "echo same"));
+    ctx.lookup.resolve("call-A");
+    await expect(handling).rejects.toBe(failure);
+    expect(ctx.onDropped).not.toHaveBeenCalled();
+    expect(ctx.deliver).toHaveBeenCalledTimes(1);
+    const message = ctx.deliver.mock.calls[0]?.[0];
+    if (message === undefined) throw new Error("expected surface rejection");
+    expect(deliveredCorrelationId(message)).toBe("corr-A");
+    expect(decisionBody(message)).toEqual({
+      outcome: "rejected",
+      message: "approval surface failed",
+    });
+  });
+
+  test("generation change during catch history after a timeout does not reject again", async () => {
+    const ctx = deferredResume();
+    const failure = new Error("gate exploded");
+    ctx.resolveSuspended.mockImplementation(async () => {
+      throw failure;
+    });
+    const reading = Promise.withResolvers<undefined>();
+    const history = Promise.withResolvers<ConversationTurn[]>();
+    ctx.history.mockResolvedValueOnce([]).mockImplementationOnce(() => {
+      reading.resolve(undefined);
+      return history.promise;
+    });
+    const handling = ctx.resume.handle(suspension("corr-A", "echo same"));
+    ctx.lookup.resolve("call-A");
+    await reading.promise;
+    expect(ctx.resolveSuspended).toHaveBeenCalledTimes(1);
+    expect(ctx.registered()).toBeUndefined();
+    ctx.invalidate();
+    history.resolve([timeoutTurn("call-A")]);
+    await expect(handling).rejects.toBe(failure);
+    expect(ctx.onDropped).toHaveBeenCalledTimes(1);
+    expect(ctx.onDropped).toHaveBeenCalledWith(APPROVAL_DROPPED_NOTICE);
+    expect(ctx.deliver).not.toHaveBeenCalled();
+    expect(ctx.registered()).toBeUndefined();
+  });
+
   for (const timedOutCallId of ["call-A", "call-B"]) {
     test(`generation change during post-gate history with ${timedOutCallId} timeout preserves exact-call cancellation`, async () => {
       const ctx = deferredResume();
@@ -1192,10 +1266,9 @@ describe("approval resume retry re-await", () => {
     expect(delivered).toHaveLength(1);
   });
 
-  // Concurrent duplicates share one in-flight rejection (one gate, zero
-  // deliveries), then a retry re-gates and delivers exactly once, and later
-  // retries memoize.
-  test("a shared rejection re-gates once, delivers once, then memoizes", async () => {
+  // Concurrent duplicates share one in-flight gate failure (one gate, one
+  // rejection delivered), and later retries memoize.
+  test("a shared gate failure rejects the parked call once, then memoizes", async () => {
     const gate = Promise.withResolvers<{ allow: boolean }>();
     const { resume, delivered, resolveSuspended } = retryHarness({
       allow: true,
@@ -1210,15 +1283,79 @@ describe("approval resume retry re-await", () => {
     await expect(first).rejects.toBe(failure);
     await expect(second).rejects.toBe(failure);
     expect(resolveSuspended).toHaveBeenCalledTimes(1);
-    expect(delivered).toHaveLength(0);
-
-    expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true);
-    expect(resolveSuspended).toHaveBeenCalledTimes(2);
+    // Nothing retries a parked call, so the failure answers it.
     expect(delivered).toHaveLength(1);
-    expect(decisionBody(onlyDecision(delivered)).outcome).toBe("approved");
+    expect(decisionBody(onlyDecision(delivered)).outcome).toBe("rejected");
 
     expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true);
-    expect(resolveSuspended).toHaveBeenCalledTimes(2);
+    expect(resolveSuspended).toHaveBeenCalledTimes(1);
     expect(delivered).toHaveLength(1);
   });
+  test("a gate failure after the reactor timed the call out delivers nothing", async () => {
+    const delivered: InboundMessage[] = [];
+    let timedOut = false;
+    const failure = new Error("gate exploded");
+    const resume = createApprovalResume({
+      getAgent: () => ({
+        deliver: (message: InboundMessage) => {
+          delivered.push(message);
+        },
+        history: async () =>
+          timedOut
+            ? ([
+                {
+                  role: "user",
+                  content: [
+                    {
+                      type: "tool_result",
+                      callId: "call-A",
+                      content: [
+                        { type: "text", text: APPROVAL_TIMEOUT_RESULT_TEXT },
+                      ],
+                    },
+                  ],
+                },
+              ] as unknown as Awaited<ReturnType<Agent["history"]>>)
+            : [],
+      }),
+      resolveParkedCallId: () => "call-A",
+      gate: {
+        resolveSuspended: async () => {
+          timedOut = true;
+          throw failure;
+        },
+      } as unknown as PermissionGate,
+    });
+    await expect(
+      resume.handle(suspension("corr-A", "echo alpha")),
+    ).rejects.toBe(failure);
+    expect(delivered).toEqual([]);
+  });
+});
+
+describe("parked overlay registration after the operator answers", () => {
+  // approve_worker_call raises a nested permission prompt as soon as its
+  // redispatched call starts, while the decision that resumed it is still being
+  // delivered. That prompt's abort signal merges the parked overlay controller.
+  test("a permission prompt raised while the decision is delivered is not pre-aborted", async () => {
+    const binding = createParkedOverlayAbortBinding();
+    const identity = new AbortController();
+    const { agent, gate } = createApprovalResumeHarness();
+    let promptSignal: AbortSignal | undefined;
+    const resume = createApprovalResume({
+      getAgent: () => agent,
+      gate,
+      resolveParkedCallId: () => "call-A",
+      registerOverlayAbort: binding.registerOverlayAbort,
+      deliver: async (message) => {
+        promptSignal = binding.identitySignal(identity.signal);
+        agent.deliver(message);
+      },
+    });
+    expect(
+      await resume.handle(suspension("corr-A", "git status --short")),
+    ).toBe(true);
+    expect(promptSignal).toBeDefined();
+    expect(promptSignal?.aborted).toBe(false);
+  });
 });