diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index ff4ba3c78..5862a1b2a 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -411,7 +411,7 @@ tool call - **command** — Splits chained commands for security classification and derives command-shape approval scopes. Multi-segment chains only offer an exact-command persist pattern (a prefix like `npm *` must not cover `npm i && rm -rf /` later). - **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 in order: catastrophic authorization denial, `skipPermissions`, then the normal tier, grant, and auto policies. Catastrophic commands remain denied under every mode. Skip mode, enabled for the process by `--dangerously-skip-permissions` or `--yolo` and persisted to the active settings source by TUI `/yolo` (machine-wide only for the default user-global source, not an explicit `--config` source), allows the call before normal policy evaluation. Therefore `--auto --yolo` behaves as yolo: auto-mode file-mutation asks and denials do not run. Otherwise, `allow`-tier passes; for `ask`-tier, the gate checks persisted approvals and then 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, `web_fetch`, `web_search`, `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 process-only `--dangerously-skip-permissions` / `--yolo` or TUI `/yolo` persisted in the active settings source (machine-wide only for the default user-global source), 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. Static secret-guard denies for path-keyed tools and authorization hard blocks still apply. Sensitive shell references require approval in normal and auto modes, but skip mode allows them without approval after catastrophic authorization checks. 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. Idle rebuild waits on three signals. `inFlight` occupancy keeps the TUI busy across the overlay until the correlated resume is accepted (`message.received` / `message.correlated`) or a generation bump `settleAll`s the waiter. The reactor's message-run bracket (`message.run.started` to `message.run.ended`) stays open while the parked run suspends and while the resumed tool and next inference execute, so a deferred reload (for example, MCP tools landing while the overlay is up) runs only after the run reaches `reply`, `wait`, or `done`. A decision delivered to a reactor that rehydrated the parked approval from disk resumes with no run open, so the TUI also holds rebuilds from that `message.correlated` until the reactor announces a terminal (`connector.reply`, a run end, the chat director's silent settle of a dropped approval, a fatal `reactor.error`, or `reactor.done`). Every agent build logs its reason (`initial`, `mcp-tools-changed`, `interrupt`, `session-rotation`). 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 the operator's decision into a rebuilt agent. A rebuild that keeps the session (interrupt, or an MCP tool-set reload) instead rejects each dropped approval its store still shows parked on the rebuilt agent, again on every such rebuild until the store shows the call closed, and the chat director commits that result without starting a turn. `/clear` and `/new` forget dropped approvals, since the new session never loads the old one's parked calls. 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()`. + - **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. Idle rebuild waits on three signals. `inFlight` occupancy keeps the TUI busy across the overlay until the correlated resume is accepted (`message.received` / `message.correlated`) or a generation bump `settleAll`s the waiter. The reactor's message-run bracket (`message.run.started` to `message.run.ended`) stays open while the parked run suspends and while the resumed tool and next inference execute, so a deferred reload (for example, MCP tools landing while the overlay is up) runs only after the run reaches `reply`, `wait`, or `done`. A decision delivered to a reactor that rehydrated the parked approval from disk resumes with no run open, so the TUI also holds rebuilds from that `message.correlated` until the reactor announces a terminal (`connector.reply`, a run end, the chat director's silent settle of a dropped approval, a fatal `reactor.error`, or `reactor.done`). Every agent build logs its reason (`initial`, `mcp-tools-changed`, `interrupt`, `session-rotation`). 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 the operator's decision into a rebuilt agent. A rebuild that keeps the session (interrupt, or an MCP tool-set reload) instead rejects each dropped approval its store still shows parked on the rebuilt agent, again on every such rebuild until the store shows the call closed, and the chat director commits that result without starting a turn. An operator decision handed to a reactor that closes before acting on it, or whose delivery fails or never sees its acceptance, leaves its call parked the same way and is settled the same way, except that an approved call whose tool had started (the TUI reports each `tool.start` to the resume) is answered as interrupted, not dropped, because it may have run and the model should check before retrying. `/clear` and `/new` forget dropped approvals, since the new session never loads the old one's parked calls. 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. While the overlay is open, resume watches history for that exact-call timeout and aborts the overlay signal so the gate auto-denies, occupancy (`inFlight`) unsticks, and no late decision is delivered. 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. diff --git a/e2e/reactor-approval-reload.test.ts b/e2e/reactor-approval-reload.test.ts index 3d2fce4cc..6d596100c 100644 --- a/e2e/reactor-approval-reload.test.ts +++ b/e2e/reactor-approval-reload.test.ts @@ -17,6 +17,7 @@ import { createPermissionGate } from "../src/permission/gate.js"; import { createReactorAuthorize } from "../src/permission/reactor-authorize.js"; import { APPROVAL_DROPPED_NOTICE, + APPROVAL_INTERRUPTED_NOTICE, createApprovalResume, resolveParkedCallIdFromStore, type ApprovalResume, @@ -25,6 +26,11 @@ import { DROPPED_APPROVAL_SETTLED_EVENT, type DroppedApprovalCalls, } from "../src/agent/director.js"; +import { + ApprovalDeliveryTimeoutError, + createApprovalDeliverer, + createQueuedApprovalDeliver, +} from "../src/tui/approval-delivery.js"; import { createDeliveryGeneration, createSessionOperationQueue, @@ -68,6 +74,23 @@ function linearClient(calls: Record[]): MCPClient { }; } +/** A save that stays in flight until released, whatever the abort signal. */ +function stuckLinearClient(calls: Record[]) { + let release: (() => void) | undefined; + const released = new Promise((resolve) => { + release = resolve; + }); + const client: MCPClient = { + ...linearClient(calls), + async callResult(_toolName, args) { + calls.push(args); + await released; + return { blocks: [{ type: "text", text: "SAVED-1" }], isError: false }; + }, + }; + return { client, release: () => release?.() }; +} + function labelClient(): MCPClient { return { serverName: "linear", @@ -427,13 +450,34 @@ 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" + /** After the reactor accepted an approval, while its loop is busy. */ + | "decision-accepted" + /** + * After an approval reached a busy loop but the deliverer gave up waiting + * to observe its acceptance. + */ + | "acceptance-timed-out" + /** After an approved call's tool started, before its result is saved. */ + | "tool-running"; + /** - * 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 }, @@ -445,21 +489,41 @@ async function dropThenInterrupt( * slower than that reactor's own. */ deadlinePassesDuringRebuild?: boolean; + interruptWhile?: InterruptPoint; + prepare?: (session: IntegrationSession) => Promise; } = {}, ): Promise { + 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, + ...(opts.prepare !== undefined ? { prepare: opts.prepare } : {}), + }); const parkedCancel = { fn: undefined as (() => void) | undefined }; const generation = createDeliveryGeneration(() => parkedCancel.fn?.()); let releaseCallId: (() => void) | undefined; const callIdResolved = new Promise((resolve) => { releaseCallId = resolve; }); + const { enqueue, awaitTail } = createSessionOperationQueue(); + let decisionQueued: (() => void) | undefined; + const queued = new Promise((resolve) => { + decisionQueued = resolve; + }); const resume = createApprovalResume({ getAgent: () => live.session.agent, gate: ctx.gate, @@ -468,12 +532,89 @@ 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: + interruptWhile === "acceptance-timed-out" + ? createApprovalDeliverer({ + deliverToAgent: (message) => live.session.agent.deliver(message), + acceptance: { + wait: () => new Promise(() => undefined), + settle: () => undefined, + abandon: () => undefined, + }, + timeoutMs: 50, + }) + : { + deliver: async (message) => live.session.agent.deliver(message), + }, + }), }); claimFrom = resume; + // As the TUI's stream sink does for every event it sees. + void (async () => { + for await (const event of live.session.agent.stream()) { + if (event.type === "tool.start") resume.toolStarted(event.data.call.id); + } + })().catch(() => undefined); 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((resolve) => { + releaseBusy = resolve; + }), + ); + ctx.approve(); + await queued; + } + if (interruptWhile === "tool-running") { + ctx.approve(); + expect(await handling).toBe(true); + await waitFor( + () => parkedEvents.some((e) => e.type === "tool.start"), + "tool start", + ); + } + if ( + interruptWhile === "decision-accepted" || + interruptWhile === "acceptance-timed-out" + ) { + // A second message's held inference keeps the loop busy, so the close's + // abort is served ahead of the accepted approval. + void live.session.agent.send("Still there?").catch(() => undefined); + await waitFor( + () => live.session.harness.scenario.matchedRequests().length === 2, + "second message's inference", + ); + ctx.approve(); + if (interruptWhile === "decision-accepted") + expect(await handling).toBe(true); + else + await expect(handling).rejects.toBeInstanceOf( + ApprovalDeliveryTimeoutError, + ); + await waitFor( + () => parkedEvents.some((e) => e.type === "message.correlated"), + "approval accepted", + ); + } + let events: ReactorEmittedEvent[] = []; + let parkedAtRebuild: (string | undefined)[] = []; startInterruptRebuild({ deliveryGeneration: generation, markSendAborted: () => undefined, @@ -482,6 +623,11 @@ async function dropThenInterrupt( if (opts.deadlinePassesDuringRebuild === true) { clock.shift(deadline + 1_000); } + // As the TUI does: prepared from the closed agent's store. + await live.session.agent.close(); + 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,7 +646,9 @@ async function dropThenInterrupt( }, }); releaseCallId?.(); - expect(await handling).toBe(true); + releaseBusy?.(); + if (interruptWhile !== "acceptance-timed-out") + expect(await handling).toBe(true); await awaitTail(); await waitFor( @@ -524,6 +672,9 @@ async function dropThenInterrupt( deadline, clock, resume, + correlationId: suspended.correlationId, + parkedEvents, + parkedAtRebuild, }; } @@ -997,4 +1148,101 @@ 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 { + 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, + ); + + test.serial( + "an approval a closing reactor accepted but never ran settles with no model turn", + async () => { + const reloaded = await dropThenInterrupt( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + NATIVE_ANTHROPIC, + { interruptWhile: "decision-accepted" }, + ); + + await expectSettledWithoutRunning(reloaded); + }, + 20_000, + ); + + test.serial( + "an approved call whose tool is still running is not rejected as dropped", + async () => { + const calls: Record[] = []; + const stuck = stuckLinearClient(calls); + cleanups.push(() => stuck.release()); + const reloaded = await dropThenInterrupt( + { name: SAVE_ISSUE, args: { title: "in flight" } }, + NATIVE_ANTHROPIC, + { + interruptWhile: "tool-running", + prepare: async (session) => { + session.toolset.dynamicRunner.addTools( + mcpClientTools(stuck.client), + ); + await session.rebuild(); + }, + }, + ); + const { session, callId, parkedEvents } = reloaded; + + expect(calls).toEqual([{ title: "in flight" }]); + expect(parkedEvents.filter((e) => e.type === "tool.start")).toHaveLength( + 1, + ); + const settled = toolResultsFor(await session.agent.history(), callId); + expect(settled).toHaveLength(1); + expect(settled.join("")).not.toContain(APPROVAL_DROPPED_NOTICE); + expect(settled.join("")).toContain(APPROVAL_INTERRUPTED_NOTICE); + }, + 20_000, + ); + + test.serial( + "an approval whose acceptance was never observed settles with no model turn", + async () => { + const reloaded = await dropThenInterrupt( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + NATIVE_ANTHROPIC, + { interruptWhile: "acceptance-timed-out" }, + ); + + await expectSettledWithoutRunning(reloaded); + }, + 20_000, + ); }); diff --git a/src/agent/director.ts b/src/agent/director.ts index f52850b60..42ab5cf59 100644 --- a/src/agent/director.ts +++ b/src/agent/director.ts @@ -395,14 +395,15 @@ function applyManageTasksToolCall( // reserved prefixes (inference., tool., reactor., fork.). export const CHAT_TASKS_CHANGED_EVENT = "custom.chat.tasks.changed"; export const CHAT_TOOLS_ACTIVATE_EVENT = "custom.chat.tools.activate"; -/** Marks a dropped approval settled with no turn, since `wait` announces nothing. */ +/** Marks an approval a rebuilt agent settled with no turn, since `wait` announces nothing. */ export const DROPPED_APPROVAL_SETTLED_EVENT = "custom.chat.approval.dropped-settled"; /** - * Calls whose parked approval an interrupt or a worker's expiring ask - * dropped. `claim` is true once for each, so its result settles the call - * without a turn nobody asked for. + * Calls whose parked approval the reactor that parked them may never + * settle: dropped by an interrupt or a worker's expiring ask, or handed to a + * reactor that closed before acting on the decision. `claim` is true once + * for each, so its result settles the call without a turn nobody asked for. */ export interface DroppedApprovalCalls { claim: (callId: string) => boolean; @@ -498,8 +499,8 @@ class ChatDirectorImpl extends DefaultDirector { // Forget cached permission denies on the next inbound user message. Session // wiring points this at PermissionGate.clearDenials; unset in unit tests. private clearDenials: (() => void) | undefined; - // Shared across rebuilds by the session, so the director that loads a - // dropped call recognizes its result. + // Shared across rebuilds by the session, so the director that loads one of + // these calls recognizes its result. private droppedApprovalCalls: DroppedApprovalCalls | undefined; // Consecutive tool-call-only turns (reset by any text or fresh user // message). Drives the soft check-in nudge at toolOnlyTurnNudgeAt — a diff --git a/src/exec/runner.ts b/src/exec/runner.ts index 51b18fbf0..89d7f1d96 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -1214,8 +1214,9 @@ export async function runExec(config: Config): Promise { 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; diff --git a/src/session/approval-resume-interrupt.test.ts b/src/session/approval-resume-interrupt.test.ts index 110f3fdd5..aeea35dec 100644 --- a/src/session/approval-resume-interrupt.test.ts +++ b/src/session/approval-resume-interrupt.test.ts @@ -18,6 +18,7 @@ import type { } from "../permission/types.js"; import { APPROVAL_DROPPED_NOTICE, + APPROVAL_INTERRUPTED_NOTICE, createApprovalResume, } from "./approval-resume.js"; import { createCorrelationAcceptance } from "../tui/correlation-acceptance.js"; @@ -27,6 +28,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 +137,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 +169,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 +206,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 +465,279 @@ describe("approval resume dropped approvals", () => { }); }); +describe("approval resume decisions a closing reactor never ran", () => { + function handedOverResume() { + const oldAgentDelivered: InboundMessage[] = []; + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => ({ + deliver: (message: InboundMessage) => oldAgentDelivered.push(message), + history: async () => [userTurn()], + }), + gate: { + resolveSuspended: async () => ({ allow: true }), + } as unknown as PermissionGate, + }); + return { resume, oldAgentDelivered }; + } + + test("an approval whose delivery failed before the hand-over is rejected on the rebuilt agent with no turn", async () => { + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => ({ + deliver: () => undefined, + history: async () => [userTurn()], + }), + deliver: async () => { + throw new Error("agent build failed"); + }, + gate: { + resolveSuspended: async () => ({ allow: true }), + } as unknown as PermissionGate, + }); + await expect(resume.handle(SUSPENDED)).rejects.toThrow( + "agent build failed", + ); + expect(resume.status("corr-1")).toBe("idle"); + + await resume.prepareRebuild(parkedStore().storage); + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + parkedStore().storage, + ); + + expect(firstDeliveredContent(rebuilt)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); + expect(resume.droppedCalls.claim("call-ask")).toBe(true); + }); + + test("an approval handed over but still parked in the rebuilt store is rejected there with no turn", async () => { + const { resume, oldAgentDelivered } = handedOverResume(); + expect(await resume.handle(SUSPENDED)).toBe(true); + expect(firstDeliveredContent(oldAgentDelivered)).toEqual({ + outcome: "approved", + }); + + await resume.prepareRebuild(parkedStore().storage); + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + parkedStore().storage, + ); + + expect(firstDeliveredContent(rebuilt)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); + expect(resume.droppedCalls.claim("call-ask")).toBe(true); + }); + + test("an approval whose tool started is told it was interrupted, not dropped, on the rebuilt agent", async () => { + // The start is announced while the delivery waits for it, before the + // delivery resolves. + const resumeRef: { current?: ReturnType } = {}; + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => ({ + deliver: () => undefined, + history: async () => [userTurn()], + }), + deliver: async () => { + resumeRef.current?.toolStarted("call-ask"); + return "delivered" as const; + }, + gate: { + resolveSuspended: async () => ({ allow: true }), + } as unknown as PermissionGate, + }); + resumeRef.current = resume; + expect(await resume.handle(SUSPENDED)).toBe(true); + + await resume.prepareRebuild(parkedStore().storage); + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + parkedStore().storage, + ); + + expect(firstDeliveredContent(rebuilt)).toEqual({ + outcome: "rejected", + message: APPROVAL_INTERRUPTED_NOTICE, + }); + expect(resume.droppedCalls.claim("call-ask")).toBe(true); + }); + + test("a tool start announced after the hand-over still tells the rebuilt agent the call was interrupted", async () => { + const { resume } = handedOverResume(); + expect(await resume.handle(SUSPENDED)).toBe(true); + resume.toolStarted("call-ask"); + + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + parkedStore().storage, + ); + + expect(firstDeliveredContent(rebuilt)).toEqual({ + outcome: "rejected", + message: APPROVAL_INTERRUPTED_NOTICE, + }); + }); + + test("a tool start for a call no decision was offered for is ignored", async () => { + const { resume } = handedOverResume(); + resume.toolStarted("call-ask"); + resume.toolStarted("call-elsewhere"); + expect(await resume.handle(SUSPENDED)).toBe(true); + + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + parkedStore().storage, + ); + + expect(firstDeliveredContent(rebuilt)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); + }); + + test("an approval whose tool started and closed on disk is forgotten", async () => { + const { resume } = handedOverResume(); + expect(await resume.handle(SUSPENDED)).toBe(true); + resume.toolStarted("call-ask"); + const { store, storage } = parkedStore(); + store.parked = false; + await resume.prepareRebuild(storage); + + expect(resume.droppedCalls.claim("call-ask")).toBe(false); + store.parked = true; + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + storage, + ); + expect(rebuilt).toEqual([]); + }); + + test("an approval handed over and closed on disk is forgotten before the rebuild, claim and all", async () => { + const { resume } = handedOverResume(); + expect(await resume.handle(SUSPENDED)).toBe(true); + const { store, storage } = parkedStore(); + await resume.prepareRebuild(storage); + store.parked = false; + + await resume.prepareRebuild(storage); + + expect(resume.droppedCalls.claim("call-ask")).toBe(false); + store.parked = true; + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + storage, + ); + expect(rebuilt).toEqual([]); + }); + + test("an approval handed over and closed on disk is forgotten", async () => { + const { resume } = handedOverResume(); + expect(await resume.handle(SUSPENDED)).toBe(true); + const { store, storage } = parkedStore(); + store.parked = false; + + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + storage, + ); + store.parked = true; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + storage, + ); + + expect(rebuilt).toEqual([]); + }); +}); + +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((resolve) => { + releaseBusy = resolve; + }), + ); + + let decisionQueued: (() => void) | undefined; + const queued = new Promise((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 +937,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 +947,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 +1058,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 }), diff --git a/src/session/approval-resume.test.ts b/src/session/approval-resume.test.ts index a5fea5534..2d6681e6e 100644 --- a/src/session/approval-resume.test.ts +++ b/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"), @@ -794,6 +796,7 @@ describe("suspended approval recovery", () => { droppedCalls: t.resume.droppedCalls, prepareRebuild: t.resume.prepareRebuild, settleOn: t.resume.settleOn, + toolStarted: t.resume.toolStarted, forgetUnsettled: t.resume.forgetUnsettled, }, }); @@ -1413,6 +1416,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( diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index 53b603bd7..f27b55893 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -48,6 +48,16 @@ const logger = getLogger([LOG_NAMESPACE_ROOT, "approval-resume"]); export const APPROVAL_DROPPED_NOTICE = "Approval dropped because the session changed."; +/** + * The result of an approved call whose tool had started when the session + * rebuilt and whose result was never saved. The reactor puts "denied by + * approver: " in front of a rejection's message, so this says it is not one. + * The operator did approve the call and it may have finished, so the model + * checks before it retries. + */ +export const APPROVAL_INTERRUPTED_NOTICE = + "Not a denial: the operator approved this call, but the session changed while the tool was running. It may have completed; check before retrying."; + const ApprovalSnapshotShape = type({ name: "string", "arguments?": "Record", @@ -71,10 +81,12 @@ export interface ApprovalResume { /** Handed to every director the session builds. */ readonly droppedCalls: DroppedApprovalCalls; /** - * Before a rebuild that keeps the session: registers each unsettled - * approval's call, resolving from `storage` any the drop landed too early - * to know, so a rehydrated deadline that fires before `settleOn` reads the - * rebuilt store still settles it without a turn. + * Between the outgoing agent's close and the next build, for a rebuild + * that keeps the session: registers the call of each unsettled approval + * `storage` still shows parked, so a rehydrated deadline that fires before + * `settleOn` reads the rebuilt store still settles it without a turn, and + * forgets the ones it shows closed. No reactor holds the store then, so a + * closed call can never produce another result. */ prepareRebuild: (storage: Pick) => Promise; /** @@ -83,11 +95,17 @@ export interface ApprovalResume { * store shows closed. One still parked is rejected again on every rebuild * until the store shows it closed, since a reactor closed before accepting * the rejection discards it; a rejection of a settled call is unmatched. + * One whose tool started is told it was interrupted, not dropped. */ settleOn: ( agent: Pick, storage: Pick, ) => Promise; + /** + * The tool of `callId` started. A decision handed over for it is then + * settled as interrupted, not dropped: the call may have run. + */ + toolStarted: (callId: string) => void; /** On rotation: nothing in the new session can settle these. */ forgetUnsettled: () => void; } @@ -283,13 +301,38 @@ export async function resolveParkedCallIdFromStore( storage: Pick, correlationId: string, ): Promise { + return (await parkedCallIds(storage)).get(correlationId); +} + +/** + * Each approval's parked call id from one load. A correlation with more than + * one approval resolves to none, so the gate's identity check and the + * rebuild settle never pick different calls. + */ +async function parkedCallIds( + storage: Pick, +): Promise> { const { pendingOperations } = await storage.load(); - const matches = pendingOperations.filter( - (operation) => + const counts = new Map(); + for (const operation of pendingOperations) { + if (operation.kind !== "approval") continue; + counts.set( + operation.correlationId, + (counts.get(operation.correlationId) ?? 0) + 1, + ); + } + const ids = new Map(); + for (const operation of pendingOperations) { + const callId = operation.suspendedCall?.id; + if ( operation.kind === "approval" && - operation.correlationId === correlationId, - ); - return matches.length === 1 ? matches[0]?.suspendedCall?.id : undefined; + counts.get(operation.correlationId) === 1 && + callId !== undefined + ) { + ids.set(operation.correlationId, callId); + } + } + return ids; } export type SuspendedApprovalResolution = @@ -442,12 +485,18 @@ function decisionMessage( } satisfies InboundMessage; } +/** + * What a host deliverer did with a decision: `skipped` when `stillCurrent()` + * was false by the time it would have sent, so nothing reached the reactor. + */ +export type ApprovalDeliveryOutcome = "delivered" | "skipped"; + export function createApprovalResume(args: { getAgent: () => Pick | undefined; deliver?: ( message: InboundMessage, stillCurrent: () => boolean, - ) => void | Promise; + ) => ApprovalDeliveryOutcome | Promise; captureGeneration?: () => () => boolean; onDropped?: (text: string) => void; registerParkedCancel?: (cancel: (() => void) | undefined) => void; @@ -478,11 +527,14 @@ export function createApprovalResume(args: { // concurrent duplicate handle shares the one in-flight outcome instead of // opening a second gate, so no waiter is lost and none double-resumes. const inflight = new Map>(); - // Approvals a generation bump dropped, keyed by correlation id with the - // parked call id once known. The reject a drop sends the old agent can be + // Approvals a generation bump dropped or whose decision was handed to a + // reactor or failed to deliver, keyed by correlation id with the parked + // call id once known. The reject a drop sends the old agent can be // discarded by its closing reactor, or never sent when the drop lands - // before the parked call resolves, so each stays here until a rebuilt - // agent's store shows its call closed. + // before the parked call resolves; a reactor closed after accepting a + // decision but before running it, or a delivery that failed, leaves the + // call parked on disk. Each stays here until a rebuilt agent's store shows + // its call closed. const unsettled = new Map(); const markUnsettled = ( correlationId: string, @@ -490,8 +542,23 @@ export function createApprovalResume(args: { ): void => { unsettled.set(correlationId, callId ?? unsettled.get(correlationId)); }; - // Entries leave only by claim or rotation: a store read can show a call - // closed between its result being written and the director claiming it. + // Calls a decision was offered for, with whether their tool started. The + // store shows a started call parked until its result is saved, the same as + // one the closing reactor never ran, so the start is the only way to tell + // them apart. The start is announced before the delivery that waits for + // it settles, so a call is watched before its decision is delivered. + const watchedCalls = new Map(); + const forgetWatched = (callId: string | undefined): void => { + if (callId !== undefined) watchedCalls.delete(callId); + }; + const settleNotice = (callId: string): string => + watchedCalls.get(callId) === true + ? APPROVAL_INTERRUPTED_NOTICE + : APPROVAL_DROPPED_NOTICE; + // settleOn never removes an entry: with the rebuilt reactor running, its + // store read can show a call closed between the result being written and + // the director claiming it. Entries leave by claim, by rotation, or when + // prepareRebuild finds the call closed with no reactor on the store. const silentCalls = new Set(); const droppedCalls: DroppedApprovalCalls = { claim: (callId) => silentCalls.delete(callId), @@ -543,16 +610,38 @@ export function createApprovalResume(args: { if (timeoutResult(initialHistory, parkedCallId)) return true; canReject = true; + // The parked cancel is unregistered before a decision is delivered, + // so a bump from here on drops it without cancelParked. Record that + // drop the moment the delivery is skipped: a rebuild queued behind the + // delivery must find it. + const currentOrDropped = (): boolean => { + if (stillCurrent()) return true; + markUnsettled(correlationId, parkedCallId); + return false; + }; const deliverDecision = async ( message: InboundMessage, ): Promise => { - if (!stillCurrent()) return false; - if (args.deliver !== undefined) { - await args.deliver(message, stillCurrent); - } else { - parkedAgent.deliver(message); + if (!currentOrDropped()) return false; + if (!watchedCalls.has(parkedCallId)) + watchedCalls.set(parkedCallId, false); + try { + if (args.deliver !== undefined) { + const outcome = await args.deliver(message, currentOrDropped); + if (outcome === "skipped") return false; + } else { + parkedAgent.deliver(message); + } + } catch (err) { + // A deliverer can throw after the hand-over, when it gives up + // observing the acceptance, so the reactor may still hold the + // decision; one that throws before it leaves the call just as + // parked. Either way a rebuild must find it. + markUnsettled(correlationId, parkedCallId); + throw err; } handedOver.add(correlationId); + markUnsettled(correlationId, parkedCallId); return true; }; const request = @@ -686,34 +775,44 @@ export function createApprovalResume(args: { : "idle", droppedCalls, prepareRebuild: async (storage) => { + if (unsettled.size === 0) return; + const parked = await parkedCallIds(storage); for (const [correlationId, known] of [...unsettled]) { - const callId = - known ?? (await resolveParkedCallIdFromStore(storage, correlationId)); - if (callId === undefined) continue; - markUnsettled(correlationId, callId); + const callId = parked.get(correlationId); + if (callId === undefined) { + unsettled.delete(correlationId); + if (known !== undefined) silentCalls.delete(known); + forgetWatched(known); + continue; + } + unsettled.set(correlationId, callId); silentCalls.add(callId); } }, settleOn: async (agent, storage) => { + if (unsettled.size === 0) return; + const parked = await parkedCallIds(storage); for (const correlationId of [...unsettled.keys()]) { - const callId = await resolveParkedCallIdFromStore( - storage, - correlationId, - ); + const callId = parked.get(correlationId); if (callId === undefined) { + forgetWatched(unsettled.get(correlationId)); unsettled.delete(correlationId); continue; } unsettled.set(correlationId, callId); silentCalls.add(callId); agent.deliver( - decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE), + decisionMessage(correlationId, "rejected", settleNotice(callId)), ); } }, + toolStarted: (callId) => { + if (watchedCalls.has(callId)) watchedCalls.set(callId, true); + }, forgetUnsettled: () => { unsettled.clear(); silentCalls.clear(); + watchedCalls.clear(); }, handle: (result) => { if (result.type !== "suspended") return Promise.resolve(false); diff --git a/src/tui/approval-delivery.test.ts b/src/tui/approval-delivery.test.ts index ce37b8f6d..5093d2ff9 100644 --- a/src/tui/approval-delivery.test.ts +++ b/src/tui/approval-delivery.test.ts @@ -5,6 +5,7 @@ import { createSessionOperationQueue } from "./delivery-queue.js"; import { ApprovalDeliveryTimeoutError, createApprovalDeliverer, + createQueuedApprovalDeliver, } from "./approval-delivery.js"; import { redactedCorrelationId } from "../session/redacted-correlation.js"; @@ -189,3 +190,45 @@ describe("approval delivery acceptance bound", () => { expect(err.message).toContain("rejected"); }); }); + +describe("queued approval deliver", () => { + function queuedDeliver(fatal: Error | null = null) { + const delivered: InboundMessage[] = []; + const deliver = createQueuedApprovalDeliver({ + enqueue: createSessionOperationQueue().enqueue, + fatalBuildError: () => fatal, + deliverer: { + deliver: async (message) => { + delivered.push(message); + }, + }, + }); + return { deliver, delivered }; + } + + test("reports a decision the session still wants as delivered", async () => { + const { deliver, delivered } = queuedDeliver(); + const message = decisionMessage("corr-1", "rejected"); + + expect(await deliver(message, () => true)).toBe("delivered"); + expect(delivered).toEqual([message]); + }); + + test("reports a decision the session moved past while it waited as skipped", async () => { + const { deliver, delivered } = queuedDeliver(); + + expect( + await deliver(decisionMessage("corr-1", "rejected"), () => false), + ).toBe("skipped"); + expect(delivered).toEqual([]); + }); + + test("throws the build error instead of delivering to a failed agent", async () => { + const { deliver, delivered } = queuedDeliver(new Error("build failed")); + + await expect( + deliver(decisionMessage("corr-1", "rejected"), () => true), + ).rejects.toThrow("build failed"); + expect(delivered).toEqual([]); + }); +}); diff --git a/src/tui/approval-delivery.ts b/src/tui/approval-delivery.ts index c192e7d55..f4c04c81b 100644 --- a/src/tui/approval-delivery.ts +++ b/src/tui/approval-delivery.ts @@ -19,6 +19,7 @@ import type { InboundMessage } from "@intx/types/runtime"; import { getLogger } from "@intx/log"; import { LOG_NAMESPACE_ROOT } from "../branding.js"; +import type { ApprovalDeliveryOutcome } from "../session/approval-resume.js"; import { redactedCorrelationId } from "../session/redacted-correlation.js"; const logger = getLogger([LOG_NAMESPACE_ROOT, "approval-delivery"]); @@ -159,3 +160,29 @@ export function createApprovalDeliverer( }, }; } + +/** + * Serializes a decision behind the session's other operations and reports + * whether it reached the deliverer. `skipped` means the session changed + * while it waited, so nothing reached the reactor. + */ +export function createQueuedApprovalDeliver(deps: { + enqueue: (op: () => Promise) => Promise; + fatalBuildError: () => Error | null; + deliverer: ApprovalDeliverer; +}): ( + message: InboundMessage, + stillCurrent: () => boolean, +) => Promise { + return async (message, stillCurrent) => { + let outcome: ApprovalDeliveryOutcome = "skipped"; + await deps.enqueue(async () => { + if (!stillCurrent()) return; + const fatal = deps.fatalBuildError(); + if (fatal !== null) throw fatal; + await deps.deliverer.deliver(message); + outcome = "delivered"; + }); + return outcome; + }; +} diff --git a/src/tui/runner/exit.test.ts b/src/tui/runner/exit.test.ts index 3e5ba3280..498569790 100644 --- a/src/tui/runner/exit.test.ts +++ b/src/tui/runner/exit.test.ts @@ -340,6 +340,7 @@ function stubSendLifecycle(agent: Agent): { approvalResume: { prepareRebuild: async () => undefined, settleOn: async () => undefined, + toolStarted: () => undefined, forgetUnsettled: () => undefined, }, } as unknown as RunnerServices; @@ -426,6 +427,10 @@ describe("stream sink routes parked approvals", () => { async function runWith(events: readonly unknown[]) { const { state, services } = stubSendLifecycle(agentStreaming(events)); const observed: { gate: unknown; stillCurrent: () => boolean }[] = []; + const started: string[] = []; + Object.assign(services.approvalResume, { + toolStarted: (callId: string) => started.push(callId), + }); Object.assign(services, { providerFailureAttempts: { current: () => undefined, @@ -440,7 +445,7 @@ describe("stream sink routes parked approvals", () => { }); await createRunLifecycle(state, services); await state.streamPromise; - return { observed, services }; + return { observed, started, services }; } test("hands a gate-blocked event to the approval recovery with the current generation", async () => { @@ -465,6 +470,15 @@ describe("stream sink routes parked approvals", () => { expect(observed[0]?.stillCurrent()).toBe(false); }); + test("tells the approval resume which call's tool started", async () => { + const { started } = await runWith([ + { type: "tool.start", seq: 1, data: { call: { id: "call-ask" } } }, + { type: "reactor.start", seq: 2, data: {} }, + ]); + + expect(started).toEqual(["call-ask"]); + }); + test("ignores events that are not a parked gate", async () => { const { observed } = await runWith([ { type: "reactor.start", seq: 1, data: {} }, @@ -940,6 +954,14 @@ function recordRebuildHooks( newest = recordingAgent([]); return newest; }) as unknown as RunnerServices["buildAgent"]; + const outgoing = state.currentAgent; + if (outgoing !== undefined) { + const close = outgoing.close.bind(outgoing); + outgoing.close = async () => { + calls.push("close"); + await close(); + }; + } services.approvalResume = { prepareRebuild: async (storage: unknown) => { calls.push( @@ -953,6 +975,7 @@ function recordRebuildHooks( : "settle-elsewhere", ); }, + toolStarted: () => undefined, forgetUnsettled: () => calls.push("forget"), } as unknown as RunnerServices["approvalResume"]; state.currentStorage = {} as RunnerState["currentStorage"]; @@ -960,7 +983,7 @@ function recordRebuildHooks( } describe("approvals dropped by a generation bump", () => { - test("an interrupt prepares the drops before building, then settles them on the new agent", async () => { + test("an interrupt prepares the drops once the old agent closes, then settles them on the new agent", async () => { const directorHolder: RunnerServices["directorHolder"] = {}; const agent = recordingAgent([]); const { state, services } = stubSendLifecycle(agent); @@ -971,7 +994,7 @@ describe("approvals dropped by a generation bump", () => { defined(state.interrupt, "interrupt")(); await services.sessionOps.awaitTail(); - expect(calls).toEqual(["prepare", "build", "settle"]); + expect(calls).toEqual(["close", "prepare", "build", "settle"]); }); test("rotation forgets approvals dropped in the old session", () => { @@ -1203,7 +1226,7 @@ describe("reload-if-idle waits for the reactor's message run", () => { expect(state.pendingReload).toBe(false); }); - test("an MCP rebuild prepares the drops before building, then settles them on the new agent", async () => { + test("an MCP rebuild prepares the drops once the old agent closes, then settles them on the new agent", async () => { const { state, services } = await idleLifecycle(); const calls = recordRebuildHooks(state, services); @@ -1211,7 +1234,7 @@ describe("reload-if-idle waits for the reactor's message run", () => { defined(state.reloadIfIdle, "reloadIfIdle")(); await services.sessionOps.awaitTail(); - expect(calls).toEqual(["prepare", "build", "settle"]); + expect(calls).toEqual(["close", "prepare", "build", "settle"]); }); /** Late MCP connects that mount tools and report the change. */ diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index 8e89e4594..1e5a5e12d 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -400,6 +400,9 @@ export async function createRunLifecycle( providerFailureAttempts.advanceToNextMessage(); } services.correlationAcceptance.observe(event); + if (event.type === "tool.start") { + services.approvalResume.toolStarted(event.data.call.id); + } if (event.type === "reactor.gate.blocked") { // A gate that parks outside a send() has no caller to hand the // suspension to the operator; route it through the approval path here. @@ -515,10 +518,11 @@ export async function createRunLifecycle( } state.currentAgent = await services.buildAgent(); }; - // Every rebuild that keeps the session reloads a dropped approval that is - // still parked, so each one settles it on the new agent: prepared from the - // outgoing store before teardown, settled once the stream is attached so - // the settle event reaches the sink. + // Every rebuild that keeps the session reloads any approval still parked + // that this session dropped or handed to a reactor that closed before + // acting on it, so each one settles it on the new agent: prepared from the + // closed agent's store before the build, settled once the stream is + // attached so the settle event reaches the sink. const prepareDrops = async (): Promise => { const storage = state.currentStorage; if (storage === null) return; @@ -585,7 +589,6 @@ export async function createRunLifecycle( { ...drift, sessionId: state.sessionId }, ); try { - await prepareDrops(); const old = liveAgent(state); const closedCleanly = await closeAgentForRebuild(old, "reload"); await state.streamPromise?.catch( @@ -594,6 +597,7 @@ export async function createRunLifecycle( if (!closedCleanly) { throw new AgentContextLockError(state.workdir); } + await prepareDrops(); await buildLiveAgent("mcp-tools-changed"); resyncIdleWithFleetFlag(services); state.streamPromise = consumeStream( @@ -738,7 +742,6 @@ export async function createRunLifecycle( // inference.error is delivered, so the recorder never sees a // terminal event. Dispose before that teardown salvages the buffer // so it is never lost or misattributed to the next cycle. - await prepareDrops(); await services.cycleRecorder.dispose("interrupted"); const closedCleanly = await closeAgentForRebuild( liveAgent(state), @@ -753,6 +756,7 @@ export async function createRunLifecycle( if (!closedCleanly) { throw new AgentContextLockError(state.workdir); } + await prepareDrops(); await buildLiveAgent("interrupt"); resyncIdleWithFleetFlag(services); services.cycleRecorder.reset(); diff --git a/src/tui/runner/queued-approval-delivery.test.ts b/src/tui/runner/queued-approval-delivery.test.ts index 26f648ed4..df03c9652 100644 --- a/src/tui/runner/queued-approval-delivery.test.ts +++ b/src/tui/runner/queued-approval-delivery.test.ts @@ -12,8 +12,8 @@ // // Uses the real createDeliverRouting + createSubmitPath (submit.ts), the real // session queue / generation, the real createApprovalResume, and the real -// createApprovalDeliverer. The approvalResume `deliver` closure is copied -// verbatim from src/tui/runner/session.ts (createApprovalResume wiring). +// createApprovalDeliverer behind createQueuedApprovalDeliver, as +// src/tui/runner/session.ts wires them. import { expect, test } from "bun:test"; import type { Agent, SendResult } from "@intx/agent"; @@ -23,7 +23,10 @@ import { createDeliveryGeneration, createSessionOperationQueue, } from "../delivery-queue.js"; -import { createApprovalDeliverer } from "../approval-delivery.js"; +import { + createApprovalDeliverer, + createQueuedApprovalDeliver, +} from "../approval-delivery.js"; import { createApprovalResume, createSuspendedApprovalRecovery, @@ -134,14 +137,11 @@ function harness(opts: HarnessOptions = {}) { resolveParkedCallId: () => PARKED_CALL, captureGeneration: deliveryGeneration.capture, parkedTimeoutPollMs: 5, - // Verbatim from src/tui/runner/session.ts. - deliver: (message, stillCurrent) => { - return sessionOps.enqueue(async () => { - if (!stillCurrent()) return; - if (state.fatalBuildError !== null) throw state.fatalBuildError; - await approvalDeliverer.deliver(message); - }); - }, + deliver: createQueuedApprovalDeliver({ + enqueue: sessionOps.enqueue, + fatalBuildError: () => state.fatalBuildError, + deliverer: approvalDeliverer, + }), gate, }); const suspendedApprovalRecovery = createSuspendedApprovalRecovery({ diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 48989b994..4a9138aa4 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -98,7 +98,10 @@ import { type AgentDeliveryResult, } from "../delivery-queue.js"; import { createCorrelationAcceptance } from "../correlation-acceptance.js"; -import { createApprovalDeliverer } from "../approval-delivery.js"; +import { + createApprovalDeliverer, + createQueuedApprovalDeliver, +} from "../approval-delivery.js"; import { createAgentToolset, type MCPServerState, @@ -573,13 +576,11 @@ export async function assembleTUISession( parkedApprovalCancel.fn = cancel; }, registerOverlayAbort: parkedOverlay.registerOverlayAbort, - deliver: (message, stillCurrent) => { - return sessionOps.enqueue(async () => { - if (!stillCurrent()) return; - if (state.fatalBuildError !== null) throw state.fatalBuildError; - await approvalDeliverer.deliver(message); - }); - }, + deliver: createQueuedApprovalDeliver({ + enqueue: sessionOps.enqueue, + fatalBuildError: () => state.fatalBuildError, + deliverer: approvalDeliverer, + }), gate: permissionGate, cwd: config.cwd, extraDeniedPaths: [config.globalSettingsPath],