Abandon a parked approval only on its own timeout

src/exec/runner.ts

@@ -976,6 +976,8 @@ export async function runExec(config: Config): Promise<ExecResult> {
       const sendResult = await activeAgent.send(operatorTaskMessage(task));
       // A suspension must not park silently in exec: the approval resume owns
       // the terminal prompt flow and delivers the decision to the reactor.
+      // No resolveParkedCallId: the vendored reactor exposes no
+      // correlationId-to-call lookup, so the history heuristic is the path.
       await createApprovalResume({
         getAgent: () => activeAgent,
         gate: permissionGate,

src/session/approval-resume.test.ts

@@ -0,0 +1,212 @@
+import { describe, expect, test } from "bun:test";
+import type { Agent, SendResult } from "@intx/agent";
+import type {
+  ApprovalSnapshot,
+  ConversationTurn,
+  InboundMessage,
+} from "@intx/types/runtime";
+
+import { APPROVAL_TIMEOUT_RESULT_TEXT } from "../permission/decline-markers.js";
+import type { PermissionGate } from "../permission/gate.js";
+import { createApprovalResume } from "./approval-resume.js";
+
+function assistantTurn(
+  calls: { id: string; name: string; command: string }[],
+): ConversationTurn {
+  return {
+    role: "assistant",
+    content: calls.map((call) => ({
+      type: "tool_call" as const,
+      id: call.id,
+      name: call.name,
+      arguments: { command: call.command },
+    })),
+    timestamp: 1,
+  };
+}
+
+function timeoutTurn(callId: string): ConversationTurn {
+  return {
+    role: "user",
+    content: [
+      {
+        type: "tool_result" as const,
+        callId,
+        content: [
+          { type: "text" as const, text: APPROVAL_TIMEOUT_RESULT_TEXT },
+        ],
+        isError: true,
+      },
+    ],
+    timestamp: 2,
+  };
+}
+
+function suspension(
+  correlationId: string,
+  command: string,
+): Extract<SendResult, { type: "suspended" }> {
+  const snapshot: ApprovalSnapshot = {
+    name: "run_shell",
+    description: "run a shell command",
+    inputSchema: {},
+    arguments: { command },
+  };
+  return { type: "suspended", correlationId, approvalSnapshot: snapshot };
+}
+
+function setup(args: {
+  preTurns: ConversationTurn[];
+  onGate: (turns: ConversationTurn[]) => void;
+  resolveParkedCallId?: (correlationId: string) => string | undefined;
+}) {
+  const turns: ConversationTurn[] = [...args.preTurns];
+  const delivered: InboundMessage[] = [];
+  const agent = {
+    history: async () => turns,
+    deliver: (message: InboundMessage) => {
+      delivered.push(message);
+    },
+  };
+  const gate = {
+    resolveSuspended: async () => {
+      args.onGate(turns);
+      return { allow: true };
+    },
+  } as unknown as PermissionGate;
+  const resume = createApprovalResume({
+    getAgent: () => agent as Pick<Agent, "deliver" | "history">,
+    gate,
+    ...(args.resolveParkedCallId !== undefined
+      ? { resolveParkedCallId: args.resolveParkedCallId }
+      : {}),
+  });
+  return { resume, delivered };
+}
+
+function decisionBody(message: InboundMessage): { outcome: string } {
+  if (message.content === undefined)
+    throw new Error("expected a decision body");
+  return JSON.parse(message.content) as { outcome: string };
+}
+
+function deliveredCorrelationId(message: InboundMessage): string {
+  const correlationId = message.headers.interchangeCorrelationId;
+  if (correlationId === undefined)
+    throw new Error("expected an interchange correlation id");
+  return correlationId;
+}
+
+describe("approval-resume parallel-parked approvals", () => {
+  test("delivers A's decision when a different parked call's approval times out", async () => {
+    const { resume, delivered } = setup({
+      preTurns: [
+        assistantTurn([
+          { id: "call-A", name: "run_shell", command: "echo alpha" },
+          { id: "call-B", name: "run_shell", command: "echo bravo" },
+        ]),
+      ],
+      onGate: (turns) => {
+        turns.push(timeoutTurn("call-B"));
+      },
+    });
+
+    const handled = await resume.handle(suspension("corr-A", "echo alpha"));
+
+    expect(handled).toBe(true);
+    expect(delivered).toHaveLength(1);
+    const message = delivered[0];
+    if (message === undefined) throw new Error("expected a delivered decision");
+    expect(deliveredCorrelationId(message)).toBe("corr-A");
+    expect(decisionBody(message).outcome).toBe("approved");
+  });
+
+  test("still drops a genuinely late decision for the same parked call", async () => {
+    const { resume, delivered } = setup({
+      preTurns: [
+        assistantTurn([
+          { id: "call-A", name: "run_shell", command: "echo alpha" },
+          { id: "call-B", name: "run_shell", command: "echo bravo" },
+        ]),
+      ],
+      onGate: (turns) => {
+        turns.push(timeoutTurn("call-A"));
+      },
+    });
+
+    const handled = await resume.handle(suspension("corr-A", "echo alpha"));
+
+    expect(handled).toBe(true);
+    expect(delivered).toHaveLength(0);
+  });
+
+  test("pending-operation lookup identifies the parked call without history tool calls", async () => {
+    const { resume, delivered } = setup({
+      preTurns: [
+        {
+          role: "user",
+          content: [{ type: "text" as const, text: "run two shell commands" }],
+          timestamp: 1,
+        },
+      ],
+      onGate: (turns) => {
+        turns.push(timeoutTurn("call-B"));
+      },
+      resolveParkedCallId: (correlationId) =>
+        correlationId === "corr-A" ? "call-A" : undefined,
+    });
+
+    const handled = await resume.handle(suspension("corr-A", "echo alpha"));
+
+    expect(handled).toBe(true);
+    expect(delivered).toHaveLength(1);
+    const message = delivered[0];
+    if (message === undefined) throw new Error("expected a delivered decision");
+    expect(deliveredCorrelationId(message)).toBe("corr-A");
+    expect(decisionBody(message).outcome).toBe("approved");
+  });
+
+  test("identical name+args twin: sibling timeout still delivers", async () => {
+    const { resume, delivered } = setup({
+      preTurns: [
+        assistantTurn([
+          { id: "call-A", name: "run_shell", command: "echo same" },
+          { id: "call-B", name: "run_shell", command: "echo same" },
+        ]),
+      ],
+      onGate: (turns) => {
+        turns.push(timeoutTurn("call-B"));
+      },
+    });
+
+    const handled = await resume.handle(suspension("corr-A", "echo same"));
+
+    expect(handled).toBe(true);
+    expect(delivered).toHaveLength(1);
+  });
+
+  // Known limitation of the exact-one history derivation: with identical
+  // name+args twins, the answered own call drops out of the candidates and
+  // the derivation resolves to the unanswered twin, so the settled check
+  // misses and the genuinely-late decision is delivered. Telling the twins
+  // apart needs the resolveParkedCallId lookup; this test locks the current
+  // shape so a future fix can flip it to a drop.
+  test("identical name+args twin: own timeout with unanswered twin delivers", async () => {
+    const { resume, delivered } = setup({
+      preTurns: [
+        assistantTurn([
+          { id: "call-A", name: "run_shell", command: "echo same" },
+          { id: "call-B", name: "run_shell", command: "echo same" },
+        ]),
+      ],
+      onGate: (turns) => {
+        turns.push(timeoutTurn("call-A"));
+      },
+    });
+
+    const handled = await resume.handle(suspension("corr-A", "echo same"));
+
+    expect(handled).toBe(true);
+    expect(delivered).toHaveLength(1);
+  });
+});

src/session/approval-resume.ts

@@ -71,23 +71,85 @@ export function requestFromApprovalSnapshot(
 // The reactor's approval timeout answers the parked call with this exact
 // upstream text (see permission/decline-markers.ts) before removing the
 // correlation, so its presence after the suspension watermark marks the
-// correlation as settled.
+// correlation as settled — but only when the result answers this very call.
+// Parallel-parked calls each time out into their own tool result (callId is
+// the original tool-call id, not the minted correlationId), so matching text
+// alone abandons a still-valid sibling decision. The parked call id must come
+// along and match block.callId; without it the text-only scan stays as the
+// fallback so a genuinely late decision is still dropped.
 
-function settledAfterSuspend(
+function timeoutResultAfterSuspend(
   turns: Awaited<ReturnType<Agent["history"]>>,
   fromIndex: number,
-): boolean {
-  return turns
-    .slice(fromIndex)
-    .flatMap((turn) => turn.content)
-    .some(
-      (block) =>
-        block.type === "tool_result" &&
-        block.content.some(
-          (part) =>
-            part.type === "text" && part.text === APPROVAL_TIMEOUT_RESULT_TEXT,
-        ),
-    );
+  parkedCallId: string | undefined,
+): string | undefined {
+  for (const block of turns.slice(fromIndex).flatMap((turn) => turn.content)) {
+    if (
+      block.type === "tool_result" &&
+      (parkedCallId === undefined || block.callId === parkedCallId) &&
+      block.content.some(
+        (part) =>
+          part.type === "text" && part.text === APPROVAL_TIMEOUT_RESULT_TEXT,
+      )
+    ) {
+      return block.callId;
+    }
+  }
+  return undefined;
+}
+
+function stableArgs(value: unknown): string {
+  if (Array.isArray(value)) return `[${value.map(stableArgs).join(",")}]`;
+  if (typeof value === "object" && value !== null) {
+    const record = value as Record<string, unknown>;
+    return `{${Object.keys(record)
+      .sort()
+      .map((key) => `${JSON.stringify(key)}:${stableArgs(record[key])}`)
+      .join(",")}}`;
+  }
+  return JSON.stringify(value) ?? "null";
+}
+
+// Derive this suspension's parked call id from history: the pre-watermark
+// tool_call matching the snapshot's name and arguments that has no tool
+// result anywhere yet. A parked call is neither run nor answered, so an
+// unanswered match is normally this suspension's own call; a timed-out
+// sibling is answered by its timeout result and drops out of the candidates.
+// Returns undefined unless exactly one candidate matches.
+//
+// Known limitation: identical name+args twins are ambiguous. When this
+// suspension's own timeout fires while an identical twin sits unanswered,
+// the twin is the exact-one survivor, so the settled check misses and the
+// late decision is delivered instead of dropped. The sibling-timeout mirror
+// (identical args, sibling answered) still resolves to the live call and
+// delivers. Telling identical twins apart needs the resolveParkedCallId
+// lookup below; the history heuristic cannot do it.
+function parkedCallIdFromHistory(
+  turns: Awaited<ReturnType<Agent["history"]>>,
+  fromIndex: number,
+  snapshot: ApprovalSnapshot,
+): string | undefined {
+  const answered = new Set<string>();
+  for (const turn of turns) {
+    for (const block of turn.content) {
+      if (block.type === "tool_result") answered.add(block.callId);
+    }
+  }
+  const wanted = stableArgs(snapshot.arguments ?? {});
+  const candidates = new Set<string>();
+  for (const turn of turns.slice(0, fromIndex)) {
+    for (const block of turn.content) {
+      if (
+        block.type === "tool_call" &&
+        block.name === snapshot.name &&
+        stableArgs(block.arguments) === wanted &&
+        !answered.has(block.id)
+      ) {
+        candidates.add(block.id);
+      }
+    }
+  }
+  return candidates.size === 1 ? [...candidates][0] : undefined;
 }
 
 function decisionMessage(
@@ -134,6 +196,21 @@ export function createApprovalResume(args: {
   // TUI: interrupt/clear bump this to reject the parked call on the old
   // agent before close/rebuild. Cleared when handle returns.
   registerParkedCancel?: (cancel: (() => void) | undefined) => void;
+  // Pending-operation lookup: map this suspension's correlationId to the
+  // parked tool-call id (PendingOperation.suspendedCall.id). The timeout
+  // result carries the original call id while the suspension carries only
+  // the minted correlationId, so this is what ties them together. Takes
+  // precedence over the history derivation below; absent callers fall back
+  // to it.
+  //
+  // Heuristic-only in production: neither the exec nor the TUI caller wires
+  // this, because the vendored reactor surface ({ start, deliver, abort })
+  // exposes no correlationId-to-call lookup (see
+  // permission/decline-markers.ts), and the suspension snapshot carries
+  // name+arguments without the call id. Wire this if upstream ever exports
+  // the pending-operation lookup; until then the history derivation is the
+  // live path.
+  resolveParkedCallId?: (correlationId: string) => string | undefined;
   gate: PermissionGate;
 }): ApprovalResume {
   const { getAgent, gate } = args;
@@ -226,13 +303,22 @@ export function createApprovalResume(args: {
           return true;
         }
         args.registerParkedCancel?.(undefined);
-        if (
-          settledAfterSuspend(await requireAgent().history(), turnsAtSuspend)
-        ) {
+        const history = await requireAgent().history();
+        const parkedCallId =
+          args.resolveParkedCallId?.(correlationId) ??
+          parkedCallIdFromHistory(history, turnsAtSuspend, approvalSnapshot);
+        const matchedTimeoutCallId = timeoutResultAfterSuspend(
+          history,
+          turnsAtSuspend,
+          parkedCallId,
+        );
+        if (matchedTimeoutCallId !== undefined) {
           // The reactor already answered the parked call (its approval timeout
           // fired while the surface was still up). Delivering now would append
           // the raw decision JSON as an uncorrelated user turn — drop and log.
-          logger.warn`late approval decision dropped correlation=${correlationId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`;
+          // The matched call id rides along so a fallback drop (parkedCallId
+          // undefined) can still be attributed to the timeout that caused it.
+          logger.warn`late approval decision dropped correlation=${correlationId} timeoutCall=${matchedTimeoutCallId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`;
           return true;
         }
         if (outcome === undefined || !outcome.allow) {

src/tui/runner/session.ts

@@ -495,6 +495,8 @@ export async function assembleTUISession(
   // Reload, interrupt, compaction continuation, and proxy deliver share one queue
   // so a rebuild never races an in-flight deliver.
   const sessionOps = createSessionOperationQueue();
+  // No resolveParkedCallId: the vendored reactor exposes no
+  // correlationId-to-call lookup, so the history heuristic is the path.
   const approvalResume = createApprovalResume({
     getAgent: () => state.currentAgent,
     captureGeneration: deliveryGeneration.capture,