fix(session): record a decision an interrupt skips as a dropped approval

The parked cancel is unregistered before an operator's decision is
delivered, so an interrupt that bumped the generation while the
decision waited on the session queue skipped it silently. Nothing
reached either agent, the call was marked handed over so a later
recovery ignored it, and the rebuilt agent answered it with "approval
timed out" at the old deadline.

Author: bwachman

e2e/reactor-approval-reload.test.ts

@@ -25,6 +25,7 @@ import {
   DROPPED_APPROVAL_SETTLED_EVENT,
   type DroppedApprovalCalls,
 } from "../src/agent/director.js";
+import { createQueuedApprovalDeliver } from "../src/tui/approval-delivery.js";
 import {
   createDeliveryGeneration,
   createSessionOperationQueue,
@@ -427,13 +428,25 @@ const ADAPTERS: Adapter[] = [
 
 interface Dropped extends Reloaded {
   resume: ApprovalResume;
+  correlationId: string;
+  /** Events from the parked agent, up to its close. */
+  parkedEvents: ReactorEmittedEvent[];
+  /** The calls the closed agent's store still showed parked at the rebuild. */
+  parkedAtRebuild: (string | undefined)[];
 }
 
+/** Where the interrupt lands relative to the operator's decision. */
+type InterruptPoint =
+  /** Before the prompt opens, while the approval resolves its parked call. */
+  | "resolving-call"
+  /** After an approval, while it waits behind a busy session queue. */
+  | "decision-queued";
+
 /**
- * Parks `call` on an ask, then interrupts while the approval is still
- * resolving its parked call, so the drop sends the old agent nothing. The
- * rebuild settles the drop on the rebuilt agent, as the TUI's interrupt does,
- * and no reply is scripted for it.
+ * Parks `call` on an ask, then interrupts at `opts.interruptWhile` (while
+ * the approval resolves its parked call, by default), so no decision reaches
+ * a reactor that runs it. The rebuild settles the drop on the rebuilt agent,
+ * as the TUI's interrupt does, and no reply is scripted for it.
  */
 async function dropThenInterrupt(
   call: { name: string; args: Record<string, unknown> },
@@ -445,21 +458,36 @@ async function dropThenInterrupt(
      * slower than that reactor's own.
      */
     deadlinePassesDuringRebuild?: boolean;
+    interruptWhile?: InterruptPoint;
   } = {},
 ): Promise<Dropped> {
+  const interruptWhile = opts.interruptWhile ?? "resolving-call";
   // Every build's director claims from the resume, which needs the session.
   let claimFrom: ApprovalResume | undefined;
   const droppedCalls: DroppedApprovalCalls = {
     claim: (callId) => claimFrom?.droppedCalls.claim(callId) === true,
   };
-  const { live, sessionOpts, ctx, clock, suspended, callId, deadline } =
-    await parkOnAsk(call, { adapter, droppedApprovalCalls: droppedCalls });
+  const {
+    live,
+    sessionOpts,
+    ctx,
+    clock,
+    suspended,
+    events: parkedEvents,
+    callId,
+    deadline,
+  } = await parkOnAsk(call, { adapter, droppedApprovalCalls: droppedCalls });
   const parkedCancel = { fn: undefined as (() => void) | undefined };
   const generation = createDeliveryGeneration(() => parkedCancel.fn?.());
   let releaseCallId: (() => void) | undefined;
   const callIdResolved = new Promise<void>((resolve) => {
     releaseCallId = resolve;
   });
+  const { enqueue, awaitTail } = createSessionOperationQueue();
+  let decisionQueued: (() => void) | undefined;
+  const queued = new Promise<void>((resolve) => {
+    decisionQueued = resolve;
+  });
   const resume = createApprovalResume({
     getAgent: () => live.session.agent,
     gate: ctx.gate,
@@ -468,12 +496,40 @@ async function dropThenInterrupt(
       parkedCancel.fn = cancel;
     },
     resolveParkedCallId: () => callIdResolved.then(() => callId),
+    // As the TUI does: behind the session's other operations.
+    deliver: createQueuedApprovalDeliver({
+      enqueue: (op) => {
+        const queuedOp = enqueue(op);
+        decisionQueued?.();
+        return queuedOp;
+      },
+      fatalBuildError: () => null,
+      deliverer: {
+        deliver: async (message) => live.session.agent.deliver(message),
+      },
+    }),
   });
   claimFrom = resume;
   const handling = resume.handle(suspended);
 
-  const { enqueue, awaitTail } = createSessionOperationQueue();
+  let releaseBusy: (() => void) | undefined;
+  if (interruptWhile !== "resolving-call") {
+    releaseCallId?.();
+    await waitFor(ctx.asked, "approval prompt");
+  }
+  if (interruptWhile === "decision-queued") {
+    // Any earlier operation holds the queue, as a send or another deliver would.
+    void enqueue(
+      () =>
+        new Promise<void>((resolve) => {
+          releaseBusy = resolve;
+        }),
+    );
+    ctx.approve();
+    await queued;
+  }
   let events: ReactorEmittedEvent[] = [];
+  let parkedAtRebuild: (string | undefined)[] = [];
   startInterruptRebuild({
     deliveryGeneration: generation,
     markSendAborted: () => undefined,
@@ -482,6 +538,9 @@ async function dropThenInterrupt(
       if (opts.deadlinePassesDuringRebuild === true) {
         clock.shift(deadline + 1_000);
       }
+      parkedAtRebuild = (
+        await live.session.storage.load()
+      ).pendingOperations.map((op) => op.suspendedCall?.id);
       await resume.prepareRebuild(live.session.storage);
       live.session = await rebuildIntegrationSession(live.session, sessionOpts);
       events = collect(live.session);
@@ -500,6 +559,7 @@ async function dropThenInterrupt(
     },
   });
   releaseCallId?.();
+  releaseBusy?.();
   expect(await handling).toBe(true);
   await awaitTail();
 
@@ -524,6 +584,9 @@ async function dropThenInterrupt(
     deadline,
     clock,
     resume,
+    correlationId: suspended.correlationId,
+    parkedEvents,
+    parkedAtRebuild,
   };
 }
 
@@ -997,4 +1060,40 @@ describe("integration: an ask dropped by an interrupt, then settled on the rebui
     },
     20_000,
   );
+
+  /** The approved call never ran, and the rebuilt agent settled it once. */
+  async function expectSettledWithoutRunning(reloaded: Dropped): Promise<void> {
+    const { session, callId, events, parkedEvents } = reloaded;
+    expect(reloaded.parkedAtRebuild).toEqual([callId]);
+    expect(parkedEvents.filter((e) => e.type === "tool.start")).toEqual([]);
+    expect((await session.storage.load()).pendingOperations).toEqual([]);
+    const settled = toolResultsFor(await session.agent.history(), callId);
+    expect(settled).toHaveLength(1);
+    expect(settled.join("")).toContain(APPROVAL_DROPPED_NOTICE);
+    expect(session.harness.scenario.matchedRequests()).toHaveLength(0);
+    expect(events.filter((e) => e.type === "inference.start")).toEqual([]);
+
+    await passDeadline(reloaded, reloaded.deadline + FRESH_WINDOW_PASSED_MS);
+    expect(toolResultsFor(await session.agent.history(), callId)).toEqual(
+      settled,
+    );
+    expect(events.filter((e) => e.type === "reactor.error")).toEqual([]);
+  }
+
+  test.serial(
+    "an approval the interrupt skips while it waits for delivery settles with no model turn",
+    async () => {
+      const reloaded = await dropThenInterrupt(
+        { name: "run_shell", args: { command: CLOSED_PORT_PROBE } },
+        NATIVE_ANTHROPIC,
+        { interruptWhile: "decision-queued" },
+      );
+
+      await expectSettledWithoutRunning(reloaded);
+      expect(reloaded.resume.status(reloaded.correlationId)).not.toBe(
+        "handed-over",
+      );
+    },
+    20_000,
+  );
 });

src/exec/runner.ts

@@ -1214,8 +1214,9 @@ export async function runExec(config: Config): Promise<ExecResult> {
         cwd: config.cwd,
         extraDeniedPaths: [config.globalSettingsPath],
         deliver: async (message, stillCurrent) => {
-          if (!stillCurrent()) return;
+          if (!stillCurrent()) return "skipped";
           await approvalDeliverer.deliver(message);
+          return "delivered";
         },
       }).handle(sendResult);
       sendCompleted = true;

src/session/approval-resume-interrupt.test.ts

@@ -27,6 +27,7 @@ import {
   createSessionOperationQueue,
   SESSION_IDENTITY_ABORT_REASON,
 } from "../tui/delivery-queue.js";
+import { createQueuedApprovalDeliver } from "../tui/approval-delivery.js";
 import { createGateRequestApproval } from "../tui/request-approval.js";
 import { startInterruptRebuild } from "../tui/runner/exit.js";
 import { runWhileAgentBusy } from "../tui/runner/state.js";
@@ -135,8 +136,9 @@ describe("approval resume generation capture", () => {
       getAgent: () => agent,
       captureGeneration: generation.capture,
       deliver: (message, stillCurrent) => {
-        if (!stillCurrent()) return;
+        if (!stillCurrent()) return "skipped";
         delivered.push(message);
+        return "delivered";
       },
       gate: {
         resolveSuspended: async () => {
@@ -166,8 +168,9 @@ describe("approval resume generation capture", () => {
       getAgent: () => agent,
       captureGeneration: generation.capture,
       deliver: (message, stillCurrent) => {
-        if (!stillCurrent()) return;
+        if (!stillCurrent()) return "skipped";
         delivered.push(message);
+        return "delivered";
       },
       gate: {
         resolveSuspended: async () => ({ allow: true }),
@@ -202,8 +205,9 @@ describe("approval resume generation capture", () => {
       getAgent: () => current,
       captureGeneration: generation.capture,
       deliver: (message, stillCurrent) => {
-        if (!stillCurrent()) return;
+        if (!stillCurrent()) return "skipped";
         current.deliver(message);
+        return "delivered";
       },
       gate: {
         resolveSuspended: async () => {
@@ -460,6 +464,81 @@ describe("approval resume dropped approvals", () => {
   });
 });
 
+describe("interrupt while the operator's decision waits for delivery", () => {
+  test("the skipped decision is recorded as a drop and settled on the rebuilt agent", async () => {
+    const parkedCancel = { fn: undefined as (() => void) | undefined };
+    const generation = createDeliveryGeneration(() => parkedCancel.fn?.());
+    const sessionOps = createSessionOperationQueue();
+    const oldAgentDelivered: InboundMessage[] = [];
+
+    // Holds the session queue busy, as any earlier deliver or op would.
+    let releaseBusy: (() => void) | undefined;
+    void sessionOps.enqueue(
+      () =>
+        new Promise<void>((resolve) => {
+          releaseBusy = resolve;
+        }),
+    );
+
+    let decisionQueued: (() => void) | undefined;
+    const queued = new Promise<void>((resolve) => {
+      decisionQueued = resolve;
+    });
+    const resume = createApprovalResume({
+      resolveParkedCallId: () => "call-ask",
+      getAgent: () => ({
+        deliver: (message: InboundMessage) => oldAgentDelivered.push(message),
+        history: async () => [userTurn()],
+      }),
+      captureGeneration: generation.capture,
+      registerParkedCancel: (cancel) => {
+        parkedCancel.fn = cancel;
+      },
+      deliver: createQueuedApprovalDeliver({
+        enqueue: (op) => {
+          const queuedOp = sessionOps.enqueue(op);
+          decisionQueued?.();
+          return queuedOp;
+        },
+        fatalBuildError: () => null,
+        deliverer: {
+          deliver: async (message) => {
+            oldAgentDelivered.push(message);
+          },
+        },
+      }),
+      gate: {
+        resolveSuspended: async () => ({ allow: false, message: "no" }),
+      } as unknown as PermissionGate,
+    });
+
+    const handling = resume.handle(SUSPENDED);
+    await queued;
+
+    const rebuilt: InboundMessage[] = [];
+    startInterruptRebuild({
+      deliveryGeneration: generation,
+      markSendAborted: () => undefined,
+      enqueue: sessionOps.enqueue,
+      rebuild: async () => {
+        await resume.settleOn(
+          { deliver: (message) => rebuilt.push(message) },
+          parkedStore().storage,
+        );
+      },
+    });
+    releaseBusy?.();
+    expect(await handling).toBe(true);
+    await sessionOps.awaitTail();
+
+    expect(oldAgentDelivered).toEqual([]);
+    expect(
+      rebuilt.map((message) => message.headers.interchangeCorrelationId),
+    ).toEqual(["corr-1"]);
+    expect(resume.status("corr-1")).not.toBe("handed-over");
+  });
+});
+
 const persistAllow: ApprovalScope = {
   id: "project-curl",
   label: "Allow curl *",
@@ -659,8 +738,8 @@ describe("approval resume occupancy until correlation", () => {
     const resume = createApprovalResume({
       resolveParkedCallId: () => "call-ask",
       getAgent: () => agent,
-      deliver: (message) =>
-        enqueue(async () => {
+      deliver: async (message) => {
+        await enqueue(async () => {
           const correlationId = message.headers.interchangeCorrelationId;
           const accepted =
             correlationId === undefined
@@ -669,7 +748,9 @@ describe("approval resume occupancy until correlation", () => {
           agent.deliver(message);
           delivered?.();
           await accepted;
-        }),
+        });
+        return "delivered" as const;
+      },
       gate: {
         resolveSuspended: async () => {
           state.pendingReload = true;
@@ -778,12 +859,13 @@ describe("approval resume occupancy until correlation", () => {
         history: async () => [userTurn()],
       }),
       captureGeneration: generation.capture,
-      deliver: (message) => {
+      deliver: async (message) => {
         const correlationId = message.headers.interchangeCorrelationId;
-        if (correlationId === undefined) return;
+        if (correlationId === undefined) return "skipped";
         const accepted = correlationAcceptance.wait(correlationId);
         waiting?.();
-        return accepted;
+        await accepted;
+        return "delivered";
       },
       gate: {
         resolveSuspended: async () => ({ allow: true }),

src/session/approval-resume.test.ts

@@ -257,10 +257,12 @@ describe("approval decision intent headers", () => {
         getAgent: () => agent,
         gate,
         resolveParkedCallId: () => "typed-call",
-        deliver: (message) =>
-          queue.enqueue(async () => {
+        deliver: async (message) => {
+          await queue.enqueue(async () => {
             agent.deliver(message);
-          }),
+          });
+          return "delivered" as const;
+        },
       });
       const handling = resume.handle(
         suspension("typed-correlation", "echo alpha"),
@@ -1413,6 +1415,7 @@ describe("parked overlay registration after the operator answers", () => {
       deliver: async (message) => {
         promptSignal = binding.identitySignal(identity.signal);
         agent.deliver(message);
+        return "delivered" as const;
       },
     });
     expect(