From dc29b9bf20f35bf03c2a8824cd0fdbdf5c76ed20 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 14 Sep 2026 21:16:00 -0700 Subject: [PATCH] fix(session): match approvals to the exact parked call --- docs/ARCHITECTURE.md | 1 + src/exec/runner.ts | 12 +- src/session/approval-resume.test.ts | 359 +++++++++++++++++- src/session/approval-resume.ts | 260 ++++--------- src/tui/runner/session.ts | 13 +- tests/integration/harness.ts | 3 +- .../reactor-approval-suspend.test.ts | 19 + tests/unit/approval-resume.test.ts | 40 +- .../approval-reload-during-suspend.test.ts | 1 + 9 files changed, 502 insertions(+), 206 deletions(-) diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3e4fdcb7e..8c4dbb52f 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -393,6 +393,7 @@ tool call - **auto-shell-policy** — Constrains `run_shell` even when auto mode would otherwise rubber-stamp it. Before matching, `expandShellSubjects` peels `bash`/`sh`/`zsh -c`, `xargs` utility tails, and transparent prefixes (`env`, `nice`, `timeout`, …) so rules see the real payload; an unparseable wrapper (variable expansion or command substitution) sets an opaque flag that forces `ask`. Effects: `deny` blocks outright (file mutations through ad-hoc tooling — output redirection, `tee`, `sed -i`/`perl -i`, interpreter inline programs or heredocs — which must instead go through `write_file`/`edit_file`); `ask` declines to auto-allow and falls through to the operator prompt (recursive `rm`, dependency installs and remote runners: npm/yarn/pnpm/bun, pip, cargo, go, brew, npx/bunx, …, force or uncontained `git worktree` ops, shell that references a sensitive path such as `.env` or a private key, and opaque wrappers). Contained non-force `git worktree add`/`remove`/`prune` and read-only `list` auto-allow (sibling destinations like `../corbits-dispatch-wts/…` included; absolute outside, `~`, globs, and credential basenames still ask). Deny beats ask when multiple subjects match. Quoted spans are stripped before pattern matching so a quoted `>` or install word in an argument is not flagged, and program names are matched only in command position. Adding a table category is a one-line rule append in `AUTO_SHELL_RULES`. - **gate** — Evaluates a call: `skipPermissions` allows everything; `allow`-tier passes; for `ask`-tier, checks persisted approvals, otherwise requests operator approval. Shell security classifies each chain segment (`||` / `&&` / `|` / `;` / newlines), but the operator is prompted once for the full command block — any unapproved segment fails the whole block, and execution always runs the unsplit original. Safe pipeline tails and pure shell no-ops (`true` / `false` / `:` and bare control-flow keywords stranded by chain-splitting) skip without a prompt. In a non-interactive run an unresolved `ask` becomes a denial. In auto mode: non-shell built-ins in `AUTO_ALLOWED_TOOLS` (writes/edits/deletes, `manage_tasks`, `spawn_agent`, `wait_agents`, …) auto-allow when not path-restricted; for `run_shell` the gate consults the auto-shell policy — a `deny` rule fails the call, an `ask` rule skips the auto-allow shortcut and proceeds to the normal approval flow, and anything unmatched is auto-allowed. Paths outside the workspace on path-arg tools are denied at authorize time (the same sandbox path-escape enforces at execution, so the gate does not show an Accept overlay that cannot succeed). Writes under the in-workspace session state root (legacy `.agent-state`) still ask under auto mode. Under `--dangerously-skip-permissions` (forces this process) or `/yolo` (persists as the user-global default via `setSkipPermissions`), the gate auto-allows those same cases, and pre-gate sandboxes (path-escape, shell session cwd retention, `list_dir` / `delete_file` workspace bounds) honor `getSkipPermissions()` live so outside-workspace access is not hard-denied after the gate already allowed it — without rebuilding the plugin stack. Secret-guard path denies and authorization hard blocks still apply. Mutating MCP and unknown built-ins are not blanket-allowed outside skip. Newly granted scopes are appended in memory and persisted. - **Reactor-gated sessions (main session; `reactorGated: true`).** The gate's decision logic lives in one `decide()` used by both consumers: `evaluate()` (the middleware path below, still used by sub-agents) and `authorizeCall()`, which expresses the decision as the vendored reactor's before-tool authz effect (`src/permission/reactor-authorize.ts` bridges it into `env.authorize`). An `ask` there suspends the call as a reactor `PendingOperation` keyed by a correlationId (persisted through the context store's existing `pendingOperations`); `send()` settles as `suspended` and `src/session/approval-resume.ts` rebuilds the operator request from the approval snapshot, resolves it through the same `requestApproval` seam the TUI overlay uses, and delivers the decision to the reactor on the correlationId signal channel — an approved decision grants a one-shot bypass and the exact parked call re-dispatches; a rejected one answers it with an error result. `inFlight` occupancy owns idle rebuild: the TUI stays busy across the overlay and waits until the correlated resume is accepted (`message.received` / `message.correlated`) or a generation bump `settleAll`s the waiter. Delivery generation owns session identity: interrupt, `/clear`, and `/new` abort the outstanding overlay, skip minting a grant, drop the decision, and surface an operator notice rather than delivering into a rebuilt agent. Under reactor gating the middleware/MCP `gateToolCall` is an execution backstop, not a second copy of `env.authorize`: it consumes the `authorizeCall` verdict only when id, name, and arguments match, and does not re-decide. Deny still blocks and does not call `next`; an `ask` or `allow` skips the middleware prompt so an approved re-dispatch never re-asks. A reused `codex-proxy` id cannot apply an outer `shell` allow to an inner `run_shell` deny. Inner posix runs whose outer tool is not `run_shell` (Codex `apply_patch` proxy) never pass `env.authorize`, so `gateToolCall` decides on that cache miss and still blocks a deny. The headless denial and the stricter chained-command hard-deny are preserved as deny effects (upstream `block`s) decided inside the same `decide()`. + - **Approval resume identity.** Before opening the operator gate, resume captures the session generation and agent/store pair and resolves the correlation exactly once through `ContextStore.load().pendingOperations` to one approval operation's `suspendedCall.id`. Missing or duplicate mappings produce no gate or delivery; store errors propagate with registration cleanup. Every decision path first checks captured history for an exact-call approval timeout, and operator decisions check again after the gate. Identical tool names and arguments never establish identity. Cancellation during lookup cannot deliver a rejection. This suppresses observed exact timeouts, not all expired correlations: the reactor removes correlation state before the queued timeout result publishes, and expiration can also race the final history check or TUI delivery queue. Atomic stale-decision admission remains reactor-owned work tracked separately in CL-8000. - **Worker reactor ownership.** `workerPermissionGate` is a reactor-gated view over the parent's live permission gate: grants and policy are shared, not copied or toggled. Worker posix plugins and inherited MCP tools are bound to that view at worker start, so they take the reactor-gated `gateToolCall` path because the view reports `isReactorGated()` — they do not close over the parent's middleware-gated `isReactorGated()`. Deny still blocks; ask/allow skip the middleware prompt. `authorizeCall` on the view never emits `ask` — unresolved approvals become denials that name the permission subject, without invoking an approval callback or suspending, even with an interactive parent; the parent can obtain a grant and retry. Worker control-plane tools (`submit_result`, `ask_director`, and nested fleet verbs other than `spawn_agent`) allow without a parent grant. Authorization and tool execution run under the same async-local worker identity and cwd. Fleet authority remains an independent restriction, not an alternative permission grant. - **matcher** — Approval pattern matching via `@intx/authz` `matchPattern` (`*` wildcards). Exact-command grants store a backslash before each metacharacter; those patterns match by equality after unescape (the package has no escape syntax). diff --git a/src/exec/runner.ts b/src/exec/runner.ts index e347a8fc0..07e607eb5 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -110,7 +110,10 @@ import { loadSessionChatPrompt, skillDirsFromEnabledPlugins, } from "../session/runtime-assembly.js"; -import { createApprovalResume } from "../session/approval-resume.js"; +import { + createApprovalResume, + resolveParkedCallIdFromStore, +} from "../session/approval-resume.js"; import { createReactorAuthorize } from "../permission/reactor-authorize.js"; import { assembleChatAgent, @@ -932,6 +935,9 @@ export async function runExec(config: Config): Promise { agent = currentAgent; // Local non-null handle after init (currentAgent stays mutable for compaction deliver). const activeAgent = currentAgent; + const activeStorage = currentStorage; + if (activeStorage === null) + throw new Error("approval resume: no context store"); if (agentToolset.connectMCP !== undefined) { await agentToolset @@ -1051,10 +1057,10 @@ 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, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(activeStorage, correlationId), gate: permissionGate, }).handle(sendResult); sendCompleted = true; diff --git a/src/session/approval-resume.test.ts b/src/session/approval-resume.test.ts index 489483d24..e86577bfb 100644 --- a/src/session/approval-resume.test.ts +++ b/src/session/approval-resume.test.ts @@ -1,14 +1,20 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, mock, test } from "bun:test"; import type { Agent, SendResult } from "@intx/agent"; import type { ApprovalSnapshot, + ContextStore, + PendingOperation, 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"; +import { + APPROVAL_DROPPED_NOTICE, + createApprovalResume, + resolveParkedCallIdFromStore, +} from "./approval-resume.js"; function assistantTurn( calls: { id: string; name: string; command: string }[], @@ -77,17 +83,20 @@ function setup(args: { const resume = createApprovalResume({ getAgent: () => agent as Pick, gate, - ...(args.resolveParkedCallId !== undefined - ? { resolveParkedCallId: args.resolveParkedCallId } - : {}), + resolveParkedCallId: + args.resolveParkedCallId ?? + ((correlationId) => (correlationId === "corr-A" ? "call-A" : undefined)), }); return { resume, delivered }; } -function decisionBody(message: InboundMessage): { outcome: string } { +function decisionBody(message: InboundMessage): { + outcome: string; + message?: string; +} { if (message.content === undefined) throw new Error("expected a decision body"); - return JSON.parse(message.content) as { outcome: string }; + return JSON.parse(message.content) as { outcome: string; message?: string }; } function deliveredCorrelationId(message: InboundMessage): string { @@ -185,14 +194,13 @@ describe("approval-resume parallel-parked approvals", () => { 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 () => { + test("identical name+args twin: own timeout with unanswered twin drops", async () => { + const pending = new Map([ + ["corr-A", "call-A"], + ["corr-B", "call-B"], + ]); const { resume, delivered } = setup({ + resolveParkedCallId: (correlationId) => pending.get(correlationId), preTurns: [ assistantTurn([ { id: "call-A", name: "run_shell", command: "echo same" }, @@ -200,6 +208,7 @@ describe("approval-resume parallel-parked approvals", () => { ]), ], onGate: (turns) => { + pending.delete("corr-A"); turns.push(timeoutTurn("call-A")); }, }); @@ -207,6 +216,328 @@ describe("approval-resume parallel-parked approvals", () => { const handled = await resume.handle(suspension("corr-A", "echo same")); expect(handled).toBe(true); + expect(delivered).toHaveLength(0); + expect(await resume.handle(suspension("corr-B", "echo same"))).toBe(true); expect(delivered).toHaveLength(1); + const message = delivered[0]; + if (message === undefined) throw new Error("expected B decision"); + expect(deliveredCorrelationId(message)).toBe("corr-B"); + }); +}); + +function storeWith( + pendingOperations: PendingOperation[], +): Pick { + return { + load: async () => ({ + turns: [], + pendingOperations, + tokenUsage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + thinking: 0, + }, + connectorState: null, + }), + }; +} + +const operation: PendingOperation = { + correlationId: "corr-A", + kind: "approval", + registeredAt: 1, + gateId: "gate-A", + suspendedCall: { + id: "call-A", + name: "run_shell", + arguments: { command: "echo same" }, + }, +}; + +describe("persisted approval identity", () => { + test("requires exactly one matching approval with a suspended call", async () => { + expect( + await resolveParkedCallIdFromStore(storeWith([operation]), "corr-A"), + ).toBe("call-A"); + expect( + await resolveParkedCallIdFromStore(storeWith([operation]), "corr-B"), + ).toBeUndefined(); + expect( + await resolveParkedCallIdFromStore(storeWith([]), "corr-A"), + ).toBeUndefined(); + expect( + await resolveParkedCallIdFromStore( + storeWith([operation, operation]), + "corr-A", + ), + ).toBeUndefined(); + const { suspendedCall: _call, ...withoutCall } = operation; + expect( + await resolveParkedCallIdFromStore(storeWith([withoutCall]), "corr-A"), + ).toBeUndefined(); + expect( + await resolveParkedCallIdFromStore( + storeWith([operation, withoutCall]), + "corr-A", + ), + ).toBeUndefined(); + }); +}); + +test("missing or ambiguous stored identity never falls back to an identical history call", async () => { + for (const pending of [[], [operation, operation]]) { + const deliver = mock((_message: InboundMessage): void => undefined); + const resolveSuspended = mock(async () => ({ allow: true })); + const resume = createApprovalResume({ + getAgent: () => ({ + deliver, + history: async () => [ + assistantTurn([ + { id: "call-B", name: "run_shell", command: "echo same" }, + ]), + ], + }), + resolveParkedCallId: (id) => + resolveParkedCallIdFromStore(storeWith(pending), id), + gate: { resolveSuspended } as unknown as PermissionGate, + }); + expect(await resume.handle(suspension("corr-A", "echo same"))).toBe(true); + expect(resolveSuspended).not.toHaveBeenCalled(); + expect(deliver).not.toHaveBeenCalled(); + } +}); + +function deferredResume() { + const lookup = Promise.withResolvers(); + const deliver = mock((_message: InboundMessage): void => undefined); + const history = mock(async (): Promise => []); + const resolveSuspended = mock(async () => ({ allow: true })); + const resolveParkedCallId = mock(() => lookup.promise); + const onDropped = mock((_text: string): void => undefined); + let cancel: (() => void) | undefined; + let current = true; + const resume = createApprovalResume({ + getAgent: () => ({ deliver, history }), + resolveParkedCallId, + gate: { resolveSuspended } as unknown as PermissionGate, + captureGeneration: () => () => current, + onDropped, + registerParkedCancel: (registered) => { + cancel = registered; + }, + }); + return { + lookup, + deliver, + onDropped, + history, + resolveSuspended, + resolveParkedCallId, + resume, + cancel: () => cancel?.(), + registered: () => cancel, + invalidate: () => { + current = false; + }, + }; +} + +describe("approval identity ordering", () => { + test("resolves once before opening the gate", async () => { + const ctx = deferredResume(); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.history).not.toHaveBeenCalled(); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.resolveParkedCallId).toHaveBeenCalledTimes(1); + expect(ctx.resolveSuspended).toHaveBeenCalledTimes(1); + expect(ctx.deliver).toHaveBeenCalledTimes(1); + expect(ctx.registered()).toBeUndefined(); + }); + + for (const cancel of [true, false]) { + test(`${cancel ? "registered cancellation" : "generation change"} during lookup never delivers or opens gate`, async () => { + const ctx = deferredResume(); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + expect(ctx.registered()).toBeDefined(); + if (cancel) ctx.cancel(); + else ctx.invalidate(); + expect(ctx.deliver).not.toHaveBeenCalled(); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.deliver).not.toHaveBeenCalled(); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.registered()).toBeUndefined(); + }); + } + + test("generation change during initial history never delivers", async () => { + const ctx = deferredResume(); + ctx.history.mockImplementation(async () => { + ctx.invalidate(); + return []; + }); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.deliver).not.toHaveBeenCalled(); + }); + + test("generation change during post-gate history cannot deliver an approval", async () => { + const ctx = deferredResume(); + ctx.history.mockResolvedValueOnce([]).mockImplementationOnce(async () => { + ctx.invalidate(); + return []; + }); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.deliver).toHaveBeenCalledTimes(1); + const message = ctx.deliver.mock.calls[0]?.[0]; + if (message === undefined) + throw new Error("expected cancellation rejection"); + expect(decisionBody(message).outcome).toBe("rejected"); + }); + + 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(); + const reading = Promise.withResolvers(); + const history = Promise.withResolvers(); + 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(timedOutCallId)]); + expect(await handling).toBe(true); + expect(ctx.onDropped).toHaveBeenCalledTimes(1); + expect(ctx.onDropped).toHaveBeenCalledWith(APPROVAL_DROPPED_NOTICE); + expect(ctx.registered()).toBeUndefined(); + if (timedOutCallId === "call-A") { + expect(ctx.deliver).not.toHaveBeenCalled(); + } else { + expect(ctx.deliver).toHaveBeenCalledTimes(1); + const message = ctx.deliver.mock.calls[0]?.[0]; + if (message === undefined) + throw new Error("expected live-call cancellation rejection"); + expect(deliveredCorrelationId(message)).toBe("corr-A"); + expect(decisionBody(message)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); + } + }); + } + + test("uses the captured agent for both history reads and direct delivery", async () => { + const agent = { + deliver: mock((_message: InboundMessage): void => undefined), + history: mock(async () => []), + }; + const other = { + deliver: mock((_message: InboundMessage): void => undefined), + history: mock(async () => []), + }; + const getAgent = mock(() => agent) + .mockReturnValueOnce(agent) + .mockReturnValue(other); + const resume = createApprovalResume({ + getAgent, + resolveParkedCallId: () => "call-A", + gate: { + resolveSuspended: async () => ({ allow: true }), + } as unknown as PermissionGate, + }); + expect(await resume.handle(suspension("corr-A", "echo same"))).toBe(true); + expect(getAgent).toHaveBeenCalledTimes(1); + expect(agent.history).toHaveBeenCalledTimes(2); + expect(agent.deliver).toHaveBeenCalledTimes(1); + expect(other.history).not.toHaveBeenCalled(); + expect(other.deliver).not.toHaveBeenCalled(); + }); + + test("does not claim atomic expiry detection before a timeout result is observed", async () => { + const pending = new Map([["corr-A", "call-A"]]); + const { resume, delivered } = setup({ + preTurns: [], + resolveParkedCallId: (id) => pending.get(id), + onGate: () => { + pending.clear(); + }, + }); + // Reactor correlation removal precedes queued timeout publication. History + // alone cannot close this interval; atomic admission belongs to the reactor. + expect(await resume.handle(suspension("corr-A", "echo same"))).toBe(true); + expect(delivered).toHaveLength(1); + }); + + test("load errors propagate and clear registration without rejection delivery", async () => { + const ctx = deferredResume(); + const error = new Error("store unavailable"); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.lookup.reject(error); + await expect(handling).rejects.toBe(error); + expect(ctx.registered()).toBeUndefined(); + expect(ctx.deliver).not.toHaveBeenCalled(); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + await expect( + resolveParkedCallIdFromStore( + { + load: async () => { + throw error; + }, + }, + "corr-A", + ), + ).rejects.toBe(error); + }); + + for (const snapshot of [ + undefined, + { name: 42 }, + { name: "run_shell", arguments: { command: 42 } }, + ]) { + for (const identity of ["missing", "expired", "live"] as const) { + test(`invalid snapshot ${JSON.stringify(snapshot)} with ${identity} identity`, async () => { + const ctx = deferredResume(); + if (identity === "expired") + ctx.history.mockResolvedValue([timeoutTurn("call-A")]); + const result = { + type: "suspended", + correlationId: "corr-A", + approvalSnapshot: snapshot, + } as unknown as SendResult; + const handling = ctx.resume.handle(result); + ctx.lookup.resolve(identity === "missing" ? undefined : "call-A"); + expect(await handling).toBe(true); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.deliver).toHaveBeenCalledTimes(identity === "live" ? 1 : 0); + if (identity === "live") { + const message = ctx.deliver.mock.calls[0]?.[0]; + if (message === undefined) throw new Error("expected rejection"); + expect(decisionBody(message).outcome).toBe("rejected"); + } + }); + } + } + + test("observed exact timeout during lookup prevents the gate and delivery", async () => { + const ctx = deferredResume(); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.history.mockResolvedValue([timeoutTurn("call-A")]); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.deliver).not.toHaveBeenCalled(); }); }); diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index 73c090135..b412cf44a 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -12,7 +12,11 @@ // a rejected one answers it with an error result. import type { Agent, SendResult } from "@intx/agent"; -import type { ApprovalSnapshot, InboundMessage } from "@intx/types/runtime"; +import type { + ApprovalSnapshot, + ContextStore, + InboundMessage, +} from "@intx/types/runtime"; import { type } from "arktype"; import { getLogger } from "@intx/log"; @@ -68,88 +72,34 @@ export function requestFromApprovalSnapshot( return anySecret ? { ...request, scopes: [] } : request; } -// 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 — 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 timeoutResultAfterSuspend( - turns: Awaited>, - fromIndex: number, - 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"; +export async function resolveParkedCallIdFromStore( + storage: Pick, + correlationId: string, +): Promise { + const { pendingOperations } = await storage.load(); + const matches = pendingOperations.filter( + (operation) => + operation.kind === "approval" && + operation.correlationId === correlationId, + ); + return matches.length === 1 ? matches[0]?.suspendedCall?.id : undefined; } -// 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( +function timeoutResult( 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; + parkedCallId: string, +): boolean { + return turns.some((turn) => + turn.content.some( + (block) => + block.type === "tool_result" && + block.callId === parkedCallId && + block.content.some( + (part) => + part.type === "text" && part.text === APPROVAL_TIMEOUT_RESULT_TEXT, + ), + ), + ); } function decisionMessage( @@ -177,114 +127,74 @@ function decisionMessage( } export function createApprovalResume(args: { - // Live agent at history/deliver time. TUI occupancy holds this identity - // until the correlated resume is accepted; a generation bump aborts the - // gate rather than retargeting a rebuilt agent. getAgent: () => Pick | undefined; - // TUI session queue. When present, each decision is awaited through this - // seam; exec omits it and uses getAgent().deliver. deliver?: ( message: InboundMessage, stillCurrent: () => boolean, ) => void | Promise; - // TUI: capture at handle() start so interrupt, /clear, or /new during the - // overlay aborts the gate and drops the decision. Exec omits this. captureGeneration?: () => () => boolean; - // TUI: operator-visible notice when an overlay decision is dropped after - // a generation bump. onDropped?: (text: string) => void; - // 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; + resolveParkedCallId: ( + correlationId: string, + ) => string | undefined | Promise; gate: PermissionGate; }): ApprovalResume { - const { getAgent, gate } = args; - - const requireAgent = (): Pick => { - const agent = getAgent(); - if (agent === undefined) { - throw new Error("approval resume: no live agent"); - } - return agent; - }; - return { handle: async (result) => { if (result.type !== "suspended") return false; - const stillCurrent = args.captureGeneration?.() ?? (() => true); - const parkedAgent = requireAgent(); + const generationCurrent = args.captureGeneration?.() ?? (() => true); + const parkedAgent = args.getAgent(); + if (parkedAgent === undefined) + throw new Error("approval resume: no live agent"); const { correlationId, approvalSnapshot } = result; - let cancelled = false; + let canReject = false; + const stillCurrent = (): boolean => !cancelled && generationCurrent(); const cancelParked = (): void => { if (cancelled) return; cancelled = true; - parkedAgent.deliver( - decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE), - ); + if (canReject) { + parkedAgent.deliver( + decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE), + ); + } }; - args.registerParkedCancel?.(cancelParked); - const dropParked = (): void => { args.onDropped?.(APPROVAL_DROPPED_NOTICE); cancelParked(); }; - + args.registerParkedCancel?.(cancelParked); try { + // The resolver captures the paired store synchronously before its first await. + const parkedCallId = await args.resolveParkedCallId(correlationId); + if (!stillCurrent()) { + dropParked(); + return true; + } + if (parkedCallId === undefined) return true; + const initialHistory = await parkedAgent.history(); + if (!stillCurrent()) { + dropParked(); + return true; + } + if (timeoutResult(initialHistory, parkedCallId)) return true; + canReject = true; + const deliverDecision = async ( message: InboundMessage, ): Promise => { if (!stillCurrent()) return; if (args.deliver !== undefined) { await args.deliver(message, stillCurrent); - return; + } else { + parkedAgent.deliver(message); } - requireAgent().deliver(message); }; - - // Turn-count watermark for the settled guard below: a "approval timed - // out" tool result appended after this point means the reactor settled - // this very correlation before our decision lands. - const turnsAtSuspend = (await parkedAgent.history()).length; - if (!stillCurrent()) { - dropParked(); - return true; - } - - if (approvalSnapshot === undefined) { - // A suspension without a snapshot cannot be surfaced; fail closed by - // rejecting the parked call so the run does not hang on an invisible - // gate. - args.registerParkedCancel?.(undefined); - await deliverDecision( - decisionMessage( - correlationId, - "rejected", - "approval surface unavailable", - ), - ); - return true; - } - - const request = requestFromApprovalSnapshot( - approvalSnapshot, - correlationId, - ); + const request = + approvalSnapshot === undefined + ? null + : requestFromApprovalSnapshot(approvalSnapshot, correlationId); if (request === null) { args.registerParkedCancel?.(undefined); await deliverDecision( @@ -296,38 +206,30 @@ export function createApprovalResume(args: { ); return true; } - - const outcome = await gate.resolveSuspended(request, stillCurrent); + const outcome = await args.gate.resolveSuspended(request, stillCurrent); if (!stillCurrent()) { dropParked(); return true; } args.registerParkedCancel?.(undefined); - 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. - // 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"}`; + const history = await parkedAgent.history(); + const timedOut = timeoutResult(history, parkedCallId); + if (timedOut) canReject = false; + if (!stillCurrent()) { + dropParked(); return true; } - if (outcome === undefined || !outcome.allow) { - await deliverDecision( - decisionMessage(correlationId, "rejected", outcome?.message), - ); + if (timedOut) { + logger.warn`late approval decision dropped correlation=${correlationId} timeoutCall=${parkedCallId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`; return true; } - await deliverDecision(decisionMessage(correlationId, "approved")); + await deliverDecision( + decisionMessage( + correlationId, + outcome?.allow === true ? "approved" : "rejected", + outcome?.allow === true ? undefined : outcome?.message, + ), + ); return true; } finally { args.registerParkedCancel?.(undefined); diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 083ad5558..08e83de06 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -57,7 +57,10 @@ import { type LiveSessionSources, } from "../../session/assemble-runtime.js"; import type { CompactionArchive } from "../../session/compaction-archive.js"; -import { createApprovalResume } from "../../session/approval-resume.js"; +import { + createApprovalResume, + resolveParkedCallIdFromStore, +} from "../../session/approval-resume.js"; import { createReactorAuthorize } from "../../permission/reactor-authorize.js"; import { buildShellBackgroundMessage, @@ -502,8 +505,6 @@ 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. // The deliverer bounds the acceptance wait so one stuck delivery fails fast // with diagnostics instead of wedging the sessionOps tail for every later // approval (send_input answers, interrupt_agent releases, ask_operator). @@ -513,6 +514,12 @@ export async function assembleTUISession( }); const approvalResume = createApprovalResume({ getAgent: () => state.currentAgent, + resolveParkedCallId: (correlationId) => { + const storage = state.currentStorage; + if (storage === null) + throw new Error("approval resume: no context store"); + return resolveParkedCallIdFromStore(storage, correlationId); + }, captureGeneration: deliveryGeneration.capture, onDropped: (text) => state.systemNotice?.(text), registerParkedCancel: (cancel) => { diff --git a/tests/integration/harness.ts b/tests/integration/harness.ts index 2701ee8d5..e3cc03970 100644 --- a/tests/integration/harness.ts +++ b/tests/integration/harness.ts @@ -74,6 +74,7 @@ export const INTEGRATION_SOURCE: InferenceSource = { const INTEGRATION_SECRET = "integration-test-key"; export interface IntegrationSession { + storage: ContextStore; harness: Harness; cwd: string; workdir: string; @@ -248,7 +249,7 @@ export async function openIntegrationSession( ? innerAgent : createPrimaryDeliveryAdmission(innerAgent, primaryArchive); - return { harness, cwd, workdir, agent, toolset }; + return { harness, cwd, workdir, agent, toolset, storage: storageForAgent }; } export async function closeIntegrationSession( diff --git a/tests/integration/reactor-approval-suspend.test.ts b/tests/integration/reactor-approval-suspend.test.ts index bfc04cf15..57d37a131 100644 --- a/tests/integration/reactor-approval-suspend.test.ts +++ b/tests/integration/reactor-approval-suspend.test.ts @@ -6,6 +6,7 @@ import { createPermissionGate } from "../../src/permission/gate.js"; import { createReactorAuthorize } from "../../src/permission/reactor-authorize.js"; import { createApprovalResume, + resolveParkedCallIdFromStore, requestFromApprovalSnapshot, } from "../../src/session/approval-resume.js"; import { @@ -84,6 +85,16 @@ describe("integration — reactor approval suspend/resume", () => { expect(result.type).toBe("suspended"); if (result.type !== "suspended") return; + const parkedCallId = await resolveParkedCallIdFromStore( + session.storage, + result.correlationId, + ); + const parkedCalls = (await session.agent.history()) + .flatMap((historyTurn) => historyTurn.content) + .filter((block) => block.type === "tool_call"); + expect(parkedCalls).toHaveLength(1); + expect(parkedCallId).toBe(defined(parkedCalls[0]).id); + // The parked call carried an approver-facing snapshot and parked on a // gate keyed by the correlationId (reactor.gate.blocked). expect(result.approvalSnapshot?.name).toBe("run_shell"); @@ -112,6 +123,8 @@ describe("integration — reactor approval suspend/resume", () => { expect(request?.subject).toBe("curl -sS https://example.com"); const resume = createApprovalResume({ getAgent: () => session.agent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(session.storage, correlationId), gate: ctx.gate, }); const handling = resume.handle(result); @@ -171,6 +184,8 @@ describe("integration — reactor approval suspend/resume", () => { const resume = createApprovalResume({ getAgent: () => session.agent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(session.storage, correlationId), gate: ctx.gate, }); const handling = resume.handle(result); @@ -221,6 +236,8 @@ describe("integration — reactor approval suspend/resume", () => { const resume = createApprovalResume({ getAgent: () => session.agent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(session.storage, correlationId), gate: ctx.gate, }); const handling = resume.handle(result); @@ -334,6 +351,8 @@ describe("integration — reactor approval suspend/resume", () => { const resume = createApprovalResume({ getAgent: () => session.agent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(session.storage, correlationId), gate: ctx.gate, }); const handling = resume.handle(result); diff --git a/tests/unit/approval-resume.test.ts b/tests/unit/approval-resume.test.ts index fc8ce1d92..d2ae50c9b 100644 --- a/tests/unit/approval-resume.test.ts +++ b/tests/unit/approval-resume.test.ts @@ -81,7 +81,11 @@ describe("approval resume late-decision guard", () => { test("a decision after the reactor settled the correlation is dropped", async () => { const turns = [userTurn()]; const { agent, gate, delivered } = harness(turns, true); - const resume = createApprovalResume({ getAgent: () => agent, gate }); + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => agent, + gate, + }); expect(await resume.handle(SUSPENDED)).toBe(true); expect(delivered).toEqual([]); @@ -90,7 +94,11 @@ describe("approval resume late-decision guard", () => { test("a live decision is still delivered", async () => { const turns = [userTurn()]; const { agent, gate, delivered } = harness(turns, false); - const resume = createApprovalResume({ getAgent: () => agent, gate }); + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => agent, + gate, + }); expect(await resume.handle(SUSPENDED)).toBe(true); expect(delivered).toHaveLength(1); @@ -105,13 +113,17 @@ describe("approval resume late-decision guard", () => { expect(correlationHeaders(delivered[0]).messageId).toBe("approval-corr-1"); }); - test("an approval timeout from before the suspension does not suppress delivery", async () => { + test("an exact approval timeout already present before lookup suppresses delivery", async () => { const turns = [userTurn(), approvalTimedOutTurn()]; const { agent, gate, delivered } = harness(turns, false); - const resume = createApprovalResume({ getAgent: () => agent, gate }); + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => agent, + gate, + }); expect(await resume.handle(SUSPENDED)).toBe(true); - expect(delivered).toHaveLength(1); + expect(delivered).toHaveLength(0); }); }); @@ -125,6 +137,7 @@ describe("approval resume delivery", () => { }; let customResolved = false; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, deliver: async (message) => { await Promise.resolve(); @@ -147,6 +160,7 @@ describe("approval resume delivery", () => { test("undefined agent throws instead of returning true", async () => { const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => undefined, gate: { resolveSuspended: async () => ({ allow: true }), @@ -163,6 +177,7 @@ describe("approval resume delivery", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, gate: { resolveSuspended: async () => ({ allow: true }), @@ -184,7 +199,11 @@ describe("approval resume delivery", () => { return { allow: true }; }, } as unknown as PermissionGate; - const resume = createApprovalResume({ getAgent: () => agent, gate }); + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => agent, + gate, + }); expect(await resume.handle(SUSPENDED)).toBe(true); expect(delivered).toHaveLength(1); @@ -206,6 +225,7 @@ describe("approval resume generation capture", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, captureGeneration: generation.capture, deliver: (message, stillCurrent) => { @@ -236,6 +256,7 @@ describe("approval resume generation capture", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, captureGeneration: generation.capture, deliver: (message, stillCurrent) => { @@ -271,6 +292,7 @@ describe("approval resume generation capture", () => { }; let current: typeof agentA | typeof agentB = agentA; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => current, captureGeneration: generation.capture, deliver: (message, stillCurrent) => { @@ -325,6 +347,7 @@ describe("approval resume generation capture", () => { }); let finishOverlay: ((outcome: { allow: boolean }) => void) | undefined; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, captureGeneration: generation.capture, registerParkedCancel: (cancel) => { @@ -453,6 +476,7 @@ describe("approval resume persist Allow after interrupt", () => { const delivered: unknown[] = []; const notices: string[] = []; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => ({ deliver: (message: unknown) => delivered.push(message), history: async () => [userTurn()], @@ -488,6 +512,7 @@ describe("approval resume stillCurrent at resolve", () => { const generation = createDeliveryGeneration(); let atResolve: boolean | undefined; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => ({ deliver: () => undefined, history: async () => [userTurn()], @@ -563,6 +588,7 @@ describe("approval resume occupancy until correlation", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, deliver: (message) => enqueue(async () => { @@ -627,6 +653,7 @@ describe("approval resume occupancy until correlation", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, deliver: (message) => enqueue(async () => { @@ -697,6 +724,7 @@ describe("approval resume occupancy until correlation", () => { reloadIfIdle: () => undefined, }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => ({ deliver: () => undefined, history: async () => [userTurn()], diff --git a/tests/unit/tui/approval-reload-during-suspend.test.ts b/tests/unit/tui/approval-reload-during-suspend.test.ts index 93b253c6e..f6125baa8 100644 --- a/tests/unit/tui/approval-reload-during-suspend.test.ts +++ b/tests/unit/tui/approval-reload-during-suspend.test.ts @@ -117,6 +117,7 @@ describe("pendingReload during resolveSuspended vs deliver enqueue", () => { }; const resume = createApprovalResume({ getAgent: () => agent, + resolveParkedCallId: () => "call-ask", deliver: (message) => enqueue(async () => { agent.deliver(message);