diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 13993cf94..e2d659eda 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -976,6 +976,8 @@ export async function runExec(config: Config): Promise { 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, diff --git a/src/session/approval-resume.test.ts b/src/session/approval-resume.test.ts new file mode 100644 index 000000000..489483d24 --- /dev/null +++ b/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 { + 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, + 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); + }); +}); diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index 56db49391..73c090135 100644 --- a/src/session/approval-resume.ts +++ b/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>, 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; + 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>, + fromIndex: number, + snapshot: ApprovalSnapshot, +): string | undefined { + const answered = new Set(); + 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(); + 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) { diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index abca12066..dae36d013 100644 --- a/src/tui/runner/session.ts +++ b/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,