diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index b8d6f926f..ff4ba3c78 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, 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 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()`. + - **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()`. - **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/integration-harness.ts b/e2e/integration-harness.ts index 4f88e4c8e..cacebb770 100644 --- a/e2e/integration-harness.ts +++ b/e2e/integration-harness.ts @@ -39,6 +39,7 @@ import { denyLateTools } from "../src/agent/late-tool-authorize.js"; import { createChatDirector, type ChatDirector, + type DroppedApprovalCalls, } from "../src/agent/director.js"; import { OPERATOR_ORIGINATED_FLAG } from "../src/agent/message-provenance.js"; import { @@ -159,6 +160,8 @@ export interface OpenIntegrationSessionOpts { depsOverrides?: Partial; /** Reuse an existing session cwd, so the agent loads that session's store. */ cwd?: string; + /** The session's dropped approvals, shared by every build's director as in the TUI. */ + droppedApprovalCalls?: DroppedApprovalCalls; } export async function openIntegrationSession( @@ -242,6 +245,7 @@ export async function openIntegrationSession( }, ); d.setClearDenials(() => opts.permissionGate.clearDenials()); + d.setDroppedApprovalCalls(opts.droppedApprovalCalls); directorHolder.current = d; return d; }, diff --git a/e2e/reactor-approval-reload.test.ts b/e2e/reactor-approval-reload.test.ts index 6a7fc8e38..3d2fce4cc 100644 --- a/e2e/reactor-approval-reload.test.ts +++ b/e2e/reactor-approval-reload.test.ts @@ -9,16 +9,27 @@ import { afterEach, describe, expect, spyOn, test } from "bun:test"; import type { ReactorEmittedEvent } from "@intx/inference"; -import type { ConversationTurn } from "@intx/types/runtime"; +import type { ConversationTurn, InferenceSource } from "@intx/types/runtime"; import type { MCPClient } from "../src/mcp/client.js"; import { mcpClientTools } from "../src/mcp/plugin.js"; import { createPermissionGate } from "../src/permission/gate.js"; import { createReactorAuthorize } from "../src/permission/reactor-authorize.js"; import { + APPROVAL_DROPPED_NOTICE, createApprovalResume, resolveParkedCallIdFromStore, + type ApprovalResume, } from "../src/session/approval-resume.js"; +import { + DROPPED_APPROVAL_SETTLED_EVENT, + type DroppedApprovalCalls, +} from "../src/agent/director.js"; +import { + createDeliveryGeneration, + createSessionOperationQueue, +} from "../src/tui/delivery-queue.js"; +import { startInterruptRebuild } from "../src/tui/runner/exit.js"; import { closeIntegrationSession, openIntegrationSession, @@ -26,6 +37,7 @@ import { runUntilDone, runUntilSuspended, type IntegrationSession, + type OpenIntegrationSessionOpts, } from "./integration-harness.js"; // Ask-tier, but answered locally: nothing listens on the discard port. @@ -119,6 +131,10 @@ function holdLongTimers() { realClearTimeout(handle); }) as typeof clearTimeout); return { + /** Move the clock to `at` without firing held timers. */ + shift(at: number): void { + offset = at - realNow(); + }, /** Move the clock to `at` and fire every held timer due by then. */ advanceTo(at: number): void { offset = at - realNow(); @@ -253,84 +269,271 @@ interface DecideThenReloadOpts { } /** - * Parks `call` on an ask, settles it per `opts.decision` (approve by - * default), and once `reloadWhen` holds for the settled run, rebuilds the - * agent on the same store. `reloadWhen` sees only events from after the - * decision. + * Opens a session and parks `call` on an ask. `live.session` is the agent's + * current session; a rebuild replaces it, and cleanup closes whichever is + * current. */ -async function decideThenReload( +async function parkOnAsk( call: { name: string; args: Record }, - reloadWhen: ( - session: IntegrationSession, - events: ReactorEmittedEvent[], - ) => boolean, - opts: DecideThenReloadOpts = {}, -): Promise { + opts: Pick & { + adapter?: Adapter; + droppedApprovalCalls?: DroppedApprovalCalls; + }, +) { + const provider = opts.adapter?.provider ?? "anthropic"; // Held from before the park so the expire case can fire the original gate // timer; the approve and reject cases clear it unfired. const clock = holdLongTimers(); cleanups.push(() => clock.restore()); const ctx = deferredApprovalGate(); const sessionOpts = { + ...opts.adapter?.sessionOpts, permissionGate: ctx.gate, authorize: createReactorAuthorize(ctx.gate), pinToolSetAtBuild: true, + ...(opts.droppedApprovalCalls !== undefined + ? { droppedApprovalCalls: opts.droppedApprovalCalls } + : {}), }; - let session = await openIntegrationSession(sessionOpts); - cleanups.push(() => closeIntegrationSession(session)); - await opts.prepare?.(session); + const live = { session: await openIntegrationSession(sessionOpts) }; + cleanups.push(() => closeIntegrationSession(live.session)); + await opts.prepare?.(live.session); - session.harness.scenario.replyOnce("anthropic", { + live.session.harness.scenario.replyOnce(provider, { toolCalls: [call], }); - if (opts.replyAfterDecision === undefined) session.harness.scenario.stall(); + if (opts.replyAfterDecision === undefined) + live.session.harness.scenario.stall(); else - session.harness.scenario.replyOnce("anthropic", { + live.session.harness.scenario.replyOnce(provider, { text: opts.replyAfterDecision, }); - const turn = await runUntilSuspended(session, "Probe the local port."); - if (turn.result.type !== "suspended") throw new Error("did not park"); - const { correlationId } = turn.result; - const parked = (await session.storage.load()).pendingOperations.find( - (op) => op.correlationId === correlationId, + const turn = await runUntilSuspended(live.session, "Probe the local port."); + const suspended = turn.result; + if (suspended.type !== "suspended") throw new Error("did not park"); + const parked = (await live.session.storage.load()).pendingOperations.find( + (op) => op.correlationId === suspended.correlationId, ); const callId = parked?.suspendedCall?.id; const deadline = parked?.timeoutAt; if (callId === undefined || deadline === undefined) { throw new Error("parked call has no persisted call id or deadline"); } + return { + live, + sessionOpts, + ctx, + clock, + suspended, + events: turn.events, + callId, + deadline, + }; +} + +/** + * Parks `call` on an ask, settles it per `opts.decision` (approve by + * default), and once `reloadWhen` holds for the settled run, rebuilds the + * agent on the same store. `reloadWhen` sees only events from after the + * decision. + */ +async function decideThenReload( + call: { name: string; args: Record }, + reloadWhen: ( + session: IntegrationSession, + events: ReactorEmittedEvent[], + ) => boolean, + opts: DecideThenReloadOpts = {}, +): Promise { + const { live, sessionOpts, ctx, clock, suspended, events, callId, deadline } = + await parkOnAsk(call, opts); const decision = opts.decision ?? "approve"; - const decidedAt = turn.events.length; + const decidedAt = events.length; if (decision === "expire") { clock.advanceTo(deadline + 1_000); } else { const resume = createApprovalResume({ - getAgent: () => session.agent, + getAgent: () => live.session.agent, gate: ctx.gate, resolveParkedCallId: (id) => - resolveParkedCallIdFromStore(session.storage, id), + resolveParkedCallIdFromStore(live.session.storage, id), }); - const handling = resume.handle(turn.result); + const handling = resume.handle(suspended); await waitFor(ctx.asked, "approval prompt"); - opts.afterApprovalOpened?.(session); + opts.afterApprovalOpened?.(live.session); if (decision === "approve") ctx.approve(); else ctx.reject(); expect(await handling).toBe(true); } - void session.harness + void live.session.harness .run({ wallClockBudgetMs: 30_000 }) .catch(() => undefined); - const before = session; + const before = live.session; await waitFor( - () => reloadWhen(before, turn.events.slice(decidedAt)), + () => reloadWhen(before, events.slice(decidedAt)), "reload point", ); - if (opts.rebuildInPlace === true) await session.rebuild(); - else session = await rebuildIntegrationSession(session, sessionOpts); - return { session, events: collect(session), callId, deadline, clock }; + if (opts.rebuildInPlace === true) await live.session.rebuild(); + else + live.session = await rebuildIntegrationSession(live.session, sessionOpts); + return { + session: live.session, + events: collect(live.session), + callId, + deadline, + clock, + }; +} + +interface Adapter { + name: string; + provider: "anthropic" | "openai"; + sessionOpts: Partial; +} + +const OPENAI_SOURCE: InferenceSource = { + id: "openai:integration", + provider: "openai", + baseURL: "https://api.openai.com/v1", + credentialId: "openai:integration", + model: "gpt-integration", +}; + +const NATIVE_ANTHROPIC: Adapter = { + name: "native Anthropic", + provider: "anthropic", + sessionOpts: {}, +}; + +const ADAPTERS: Adapter[] = [ + NATIVE_ANTHROPIC, + { + name: "OpenAI-shaped", + provider: "openai", + sessionOpts: { + sources: [OPENAI_SOURCE], + credentialRecords: { + [OPENAI_SOURCE.credentialId]: { + provenance: { kind: "api-key" }, + material: { secret: "openai-integration-key" }, + }, + }, + }, + }, +]; + +interface Dropped extends Reloaded { + resume: ApprovalResume; +} + +/** + * 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. + */ +async function dropThenInterrupt( + call: { name: string; args: Record }, + adapter: Adapter, + opts: { + /** + * The interrupt lands just before the deadline: it has passed by the + * time the rebuilt reactor rehydrates, and the settle's store read is + * slower than that reactor's own. + */ + deadlinePassesDuringRebuild?: boolean; + } = {}, +): Promise { + // 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 parkedCancel = { fn: undefined as (() => void) | undefined }; + const generation = createDeliveryGeneration(() => parkedCancel.fn?.()); + let releaseCallId: (() => void) | undefined; + const callIdResolved = new Promise((resolve) => { + releaseCallId = resolve; + }); + const resume = createApprovalResume({ + getAgent: () => live.session.agent, + gate: ctx.gate, + captureGeneration: generation.capture, + registerParkedCancel: (cancel) => { + parkedCancel.fn = cancel; + }, + resolveParkedCallId: () => callIdResolved.then(() => callId), + }); + claimFrom = resume; + const handling = resume.handle(suspended); + + const { enqueue, awaitTail } = createSessionOperationQueue(); + let events: ReactorEmittedEvent[] = []; + startInterruptRebuild({ + deliveryGeneration: generation, + markSendAborted: () => undefined, + enqueue, + rebuild: async () => { + if (opts.deadlinePassesDuringRebuild === true) { + clock.shift(deadline + 1_000); + } + await resume.prepareRebuild(live.session.storage); + live.session = await rebuildIntegrationSession(live.session, sessionOpts); + events = collect(live.session); + const storage = live.session.storage; + await resume.settleOn( + live.session.agent, + opts.deadlinePassesDuringRebuild === true + ? { + load: async () => { + await new Promise((resolve) => setTimeout(resolve, 50)); + return storage.load(); + }, + } + : storage, + ); + }, + }); + releaseCallId?.(); + expect(await handling).toBe(true); + await awaitTail(); + + await waitFor( + () => events.some((e) => String(e.type) === DROPPED_APPROVAL_SETTLED_EVENT), + "dropped approval settled", + ); + // The settle is announced before the wait terminal's cycle commit lands. + const settledAt = performance.now(); + while ( + toolResultsFor(await committedTurns(live.session), callId).length === 0 + ) { + if (performance.now() - settledAt > 5_000) { + throw new Error("timed out: dropped approval committed"); + } + await new Promise((resolve) => setTimeout(resolve, 5)); + } + return { + session: live.session, + events, + callId, + deadline, + clock, + resume, + }; +} + +/** The turns in the store's newest commit. */ +async function committedTurns( + session: IntegrationSession, +): Promise { + const [newest] = await session.agent.checkpoints(1); + if (newest === undefined) throw new Error("no commits"); + return session.agent.readAt(newest.hash); } // Past both the original deadline and a full approval window freshly armed @@ -601,37 +804,19 @@ describe("integration: a rejection the reactor accepted but never ran, then reop test.serial( "the reopened agent loads the denial once and no parked call, and the old deadline adds no result", async () => { - const clock = holdLongTimers(); - cleanups.push(() => clock.restore()); - const ctx = deferredApprovalGate(); - const sessionOpts = { - permissionGate: ctx.gate, - authorize: createReactorAuthorize(ctx.gate), - pinToolSetAtBuild: true, - }; - const live = { session: await openIntegrationSession(sessionOpts) }; - cleanups.push(() => closeIntegrationSession(live.session)); - live.session.harness.scenario.replyOnce("anthropic", { - toolCalls: [ - { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, - ], - }); - live.session.harness.scenario.stall(); - - const turn = await runUntilSuspended( - live.session, - "Probe the local port.", - ); - const suspended = turn.result; - if (suspended.type !== "suspended") throw new Error("did not park"); - const parked = (await live.session.storage.load()).pendingOperations.find( - (op) => op.correlationId === suspended.correlationId, + const { + live, + sessionOpts, + ctx, + clock, + suspended, + events, + callId, + deadline, + } = await parkOnAsk( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + {}, ); - const callId = parked?.suspendedCall?.id; - const deadline = parked?.timeoutAt; - if (callId === undefined || deadline === undefined) { - throw new Error("parked call has no persisted call id or deadline"); - } void live.session.agent.send("Still there?").catch(() => undefined); await waitFor( @@ -649,7 +834,7 @@ describe("integration: a rejection the reactor accepted but never ran, then reop ctx.reject(); expect(await handling).toBe(true); await waitFor( - () => turn.events.some((e) => e.type === "message.correlated"), + () => events.some((e) => e.type === "message.correlated"), "rejection accepted", ); @@ -742,3 +927,74 @@ describe("integration: an ask settled without running, then reloading the agent 20_000, ); }); + +describe("integration: an ask dropped by an interrupt, then settled on the rebuilt agent", () => { + for (const adapter of ADAPTERS) { + test.serial( + `${adapter.name}: the drop settles with no model turn, and the next send carries it once`, + async () => { + const reloaded = await dropThenInterrupt( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + adapter, + ); + const { session, callId, events, resume } = reloaded; + const requests = () => session.harness.scenario.matchedRequests(); + + 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(toolResultsFor(await committedTurns(session), callId)).toEqual( + settled, + ); + expect(requests()).toHaveLength(0); + expect(events.filter((e) => e.type === "inference.start")).toEqual([]); + expect(resume.droppedCalls.claim(callId)).toBe(false); + + await passDeadline(reloaded); + expect(toolResultsFor(await session.agent.history(), callId)).toEqual( + settled, + ); + expect(requests()).toHaveLength(0); + + session.harness.scenario.replyOnce(adapter.provider, { + text: "Noted.", + }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Noted."); + const [request] = requests(); + expect(requests()).toHaveLength(1); + if (request === undefined) throw new Error("no request"); + const body = await (request.clone() as unknown as Request).text(); + // Once as the tool call, once as its result. + expect(body.split(callId).length - 1).toBe(2); + expect(body.indexOf(APPROVAL_DROPPED_NOTICE)).toBeGreaterThan(-1); + expect(body.indexOf(APPROVAL_DROPPED_NOTICE)).toBeLessThan( + body.lastIndexOf("Continue."), + ); + expect(events.filter((e) => e.type === "reactor.error")).toEqual([]); + }, + 20_000, + ); + } + + test.serial( + "a drop whose deadline passes during the rebuild settles with no model turn", + async () => { + const reloaded = await dropThenInterrupt( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + NATIVE_ANTHROPIC, + { deadlinePassesDuringRebuild: true }, + ); + const { session, callId, events } = reloaded; + + expect((await session.storage.load()).pendingOperations).toEqual([]); + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + expect(session.harness.scenario.matchedRequests()).toHaveLength(0); + expect(events.filter((e) => e.type === "inference.start")).toEqual([]); + }, + 20_000, + ); +}); diff --git a/src/agent/director.test.ts b/src/agent/director.test.ts index 3ab402ed7..506c7df56 100644 --- a/src/agent/director.test.ts +++ b/src/agent/director.test.ts @@ -9,6 +9,7 @@ import { askOperatorDefinition, CHAT_TASKS_CHANGED_EVENT, CHAT_TOOLS_ACTIVATE_EVENT, + DROPPED_APPROVAL_SETTLED_EVENT, createChatDirector, toolSetDigest, } from "./director.js"; @@ -2729,3 +2730,60 @@ describe("tool-discipline rules on the wire", () => { expect(await chatPromptSentFor("claude-sonnet-4")).toBe("BASE PROMPT"); }); }); + +describe("dropped approval results", () => { + function resumedResult(callId: string): ReactorInboundEvent { + return { + type: "resume.tool_result", + result: { + callId, + content: "denied by approver: dropped", + isError: true, + }, + } as unknown as ReactorInboundEvent; + } + + test("a dropped approval's result is committed once, with no new turn even with open tasks", async () => { + const director = createChatDirector("base", [], {}); + director.restoreTasks([{ id: "t1", title: "open work", status: "doing" }]); + const dropped = new Set(["call-dropped"]); + director.setDroppedApprovalCalls({ + claim: (callId) => dropped.delete(callId), + }); + + const actions = await director.decide( + resumedResult("call-dropped"), + mockState, + stubReactorCapabilities, + ); + + expect(actions).toEqual([ + { type: "checkpoint", message: "dropped-approval" }, + { + type: "emit", + eventType: DROPPED_APPROVAL_SETTLED_EVENT, + data: { callId: "call-dropped" }, + }, + { type: "wait" }, + ]); + expect(dropped.size).toBe(0); + }); + + test("any other resumed result still re-infers", async () => { + const director = createChatDirector("base", [], {}); + const dropped = new Set(["call-dropped"]); + director.setDroppedApprovalCalls({ + claim: (callId) => dropped.delete(callId), + }); + + const actions = await director.decide( + resumedResult("call-other"), + mockState, + stubReactorCapabilities, + ); + + const list = Array.isArray(actions) ? actions : [actions]; + expect(list.some((action) => action.type === "infer")).toBe(true); + expect([...dropped]).toEqual(["call-dropped"]); + }); +}); diff --git a/src/agent/director.ts b/src/agent/director.ts index 12c267449..f52850b60 100644 --- a/src/agent/director.ts +++ b/src/agent/director.ts @@ -395,6 +395,18 @@ 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. */ +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. + */ +export interface DroppedApprovalCalls { + claim: (callId: string) => boolean; +} export const COMPACTION_FOLD_NONCONVERGED_EVENT = "custom.compaction.fold_nonconverged"; export const CREDENTIAL_RECOVERY_INTERCHANGE_TYPE = @@ -486,6 +498,9 @@ 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. + 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 // nudge, not a stop. @@ -665,6 +680,10 @@ class ChatDirectorImpl extends DefaultDirector { this.clearDenials = clear; } + setDroppedApprovalCalls(calls: DroppedApprovalCalls | undefined): void { + this.droppedApprovalCalls = calls; + } + armCredentialRecoveryContinuation(generation: number): void { this.credentialRecoveryGeneration = generation; } @@ -839,6 +858,21 @@ class ChatDirectorImpl extends DefaultDirector { state: ReactorState, capabilities: ReactorCapabilities, ): Promise { + // Settled before compaction, loop protection and the nudges, any of + // which could turn the wait back into a turn the operator never asked + // for. No send is waiting on it, so a bare wait cannot hang one. + if ( + event.type === "resume.tool_result" && + this.droppedApprovalCalls?.claim(event.result.callId) === true + ) { + return [ + capabilities.checkpoint("dropped-approval"), + capabilities.emit(DROPPED_APPROVAL_SETTLED_EVENT, { + callId: event.result.callId, + }), + capabilities.wait(), + ]; + } this.coordinatorRethrowNoted = false; try { const settled = ensureCycleSettlesWithReply( @@ -1326,6 +1360,7 @@ export interface ChatDirector extends ReactorDirector { setWorkflowCoordinator(coordinator: WorkflowCoordinator | undefined): void; setAllowIdleWithFleet(value: boolean): void; setClearDenials(clear: (() => void) | undefined): void; + setDroppedApprovalCalls(calls: DroppedApprovalCalls | undefined): void; armCredentialRecoveryContinuation(generation: number): void; cancelCredentialRecoveryContinuation(generation: number): void; getTasks(): Task[]; diff --git a/src/session/approval-resume-interrupt.test.ts b/src/session/approval-resume-interrupt.test.ts index 834a989ef..110f3fdd5 100644 --- a/src/session/approval-resume-interrupt.test.ts +++ b/src/session/approval-resume-interrupt.test.ts @@ -1,6 +1,10 @@ import { describe, expect, test } from "bun:test"; -import type { ConversationTurn, InboundMessage } from "@intx/types/runtime"; +import type { + ContextStore, + ConversationTurn, + InboundMessage, +} from "@intx/types/runtime"; import { APPROVAL_TIMEOUT_RESULT_TEXT } from "../permission/decline-markers.js"; import { @@ -295,6 +299,167 @@ describe("approval resume generation capture", () => { }); }); +// The store a rebuilt agent loads; `store.parked` flips the dropped call closed. +function parkedStore() { + const store = { parked: true }; + return { + store, + storage: { + load: async () => + ({ + pendingOperations: store.parked + ? [ + { + kind: "approval", + correlationId: "corr-1", + suspendedCall: { id: "call-ask" }, + }, + ] + : [], + }) as unknown as Awaited>, + }, + }; +} + +describe("approval resume dropped approvals", () => { + function droppableResume( + resolveParkedCallId: () => string | Promise, + ) { + const parkedCancel = { fn: undefined as (() => void) | undefined }; + const generation = createDeliveryGeneration(() => parkedCancel.fn?.()); + const delivered: unknown[] = []; + let overlayReady: (() => void) | undefined; + const overlayOpened = new Promise((resolve) => { + overlayReady = resolve; + }); + const resume = createApprovalResume({ + resolveParkedCallId, + getAgent: () => ({ + deliver: (message: unknown) => delivered.push(message), + history: async () => [userTurn()], + }), + captureGeneration: generation.capture, + registerParkedCancel: (cancel) => { + parkedCancel.fn = cancel; + }, + gate: { + resolveSuspended: () => { + overlayReady?.(); + return new Promise(() => undefined); + }, + } as unknown as PermissionGate, + }); + return { resume, generation, parkedCancel, delivered, overlayOpened }; + } + + test("a drop while the prompt is open is rejected once on the rebuilt agent", async () => { + const { resume, generation, parkedCancel, overlayOpened } = droppableResume( + () => "call-ask", + ); + + void resume.handle(SUSPENDED); + await overlayOpened; + const cancel = parkedCancel.fn; + generation.bump(); + cancel?.(); + + const rebuilt: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => rebuilt.push(message) }, + parkedStore().storage, + ); + expect(rebuilt).toHaveLength(1); + expect(rebuilt[0]?.headers.interchangeCorrelationId).toBe("corr-1"); + expect(firstDeliveredContent(rebuilt)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); + expect(resume.droppedCalls.claim("call-ask")).toBe(true); + expect(resume.droppedCalls.claim("call-ask")).toBe(false); + }); + + test("a drop before the parked call resolves sends the old agent nothing and still registers its call", async () => { + let resolveCallId: ((id: string) => void) | undefined; + const { resume, generation, delivered } = droppableResume( + () => + new Promise((resolve) => { + resolveCallId = resolve; + }), + ); + + const handling = resume.handle(SUSPENDED); + generation.bump(); + resolveCallId?.("call-ask"); + + expect(await handling).toBe(true); + expect(delivered).toEqual([]); + // Registered before the rebuilt agent's store is read, so a rehydrated + // deadline that fires first still settles without a turn. + await resume.prepareRebuild(parkedStore().storage); + expect(resume.droppedCalls.claim("call-ask")).toBe(true); + }); + + test("a drop whose call id never resolved is registered from the outgoing store", async () => { + const { resume, generation } = droppableResume( + () => new Promise(() => undefined), + ); + void resume.handle(SUSPENDED); + generation.bump(); + + await resume.prepareRebuild(parkedStore().storage); + + expect(resume.droppedCalls.claim("call-ask")).toBe(true); + }); + + test("a drop is rejected again on every rebuild until the store shows the call closed", async () => { + const { resume, generation, overlayOpened } = droppableResume( + () => "call-ask", + ); + void resume.handle(SUSPENDED); + await overlayOpened; + generation.bump(); + const { store, storage } = parkedStore(); + // A reactor closed before it accepts the rejection swallows it, so the + // store still shows the call parked. + const delivered: InboundMessage[] = []; + const agent = { + deliver: (message: InboundMessage) => delivered.push(message), + }; + + await resume.settleOn(agent, storage); + expect(delivered).toHaveLength(1); + await resume.settleOn(agent, storage); + expect(delivered).toHaveLength(2); + + store.parked = false; + await resume.settleOn(agent, storage); + expect(delivered).toHaveLength(2); + store.parked = true; + await resume.settleOn(agent, storage); + expect(delivered).toHaveLength(2); + }); + + test("rotation forgets drops and their calls", async () => { + const { resume, generation, overlayOpened } = droppableResume( + () => "call-ask", + ); + void resume.handle(SUSPENDED); + await overlayOpened; + generation.bump(); + await resume.prepareRebuild(parkedStore().storage); + + resume.forgetUnsettled(); + + const delivered: InboundMessage[] = []; + await resume.settleOn( + { deliver: (message) => delivered.push(message) }, + parkedStore().storage, + ); + expect(delivered).toEqual([]); + expect(resume.droppedCalls.claim("call-ask")).toBe(false); + }); +}); + const persistAllow: ApprovalScope = { id: "project-curl", label: "Allow curl *", diff --git a/src/session/approval-resume.test.ts b/src/session/approval-resume.test.ts index 557c965fb..a5fea5534 100644 --- a/src/session/approval-resume.test.ts +++ b/src/session/approval-resume.test.ts @@ -791,6 +791,10 @@ describe("suspended approval recovery", () => { throw new Error("deliver failed: transport closed"); }, status: t.resume.status, + droppedCalls: t.resume.droppedCalls, + prepareRebuild: t.resume.prepareRebuild, + settleOn: t.resume.settleOn, + forgetUnsettled: t.resume.forgetUnsettled, }, }); failing.capture(result, () => true); diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index 1b16782ae..53b603bd7 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -22,6 +22,7 @@ import { getLogger } from "@intx/log"; import { LOG_NAMESPACE_ROOT } from "../branding.js"; import { canonicalToolName } from "../agent/canonical-tool-name.js"; +import type { DroppedApprovalCalls } from "../agent/director.js"; import { inspectShellSecretReference, shellSecretInspectionRequiresApproval, @@ -67,6 +68,28 @@ export interface ApprovalResume { * has been handed over, so a fresh `handle` would open the operator surface. */ status: (correlationId: string) => "idle" | "in-flight" | "handed-over"; + /** 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. + */ + prepareRebuild: (storage: Pick) => Promise; + /** + * After that rebuild: rejects each unsettled approval still parked in + * `storage` on `agent`, the agent built from it, and forgets the ones the + * 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. + */ + settleOn: ( + agent: Pick, + storage: Pick, + ) => Promise; + /** On rotation: nothing in the new session can settle these. */ + forgetUnsettled: () => void; } /** @@ -455,6 +478,24 @@ 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 + // 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. + const unsettled = new Map(); + const markUnsettled = ( + correlationId: string, + callId: string | undefined, + ): 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. + const silentCalls = new Set(); + const droppedCalls: DroppedApprovalCalls = { + claim: (callId) => silentCalls.delete(callId), + }; const settleSuspended = async ( result: Extract, ): Promise => { @@ -465,10 +506,12 @@ export function createApprovalResume(args: { const { correlationId, approvalSnapshot } = result; let cancelled = false; let canReject = false; + let knownCallId: string | undefined; const stillCurrent = (): boolean => !cancelled && generationCurrent(); const cancelParked = (): void => { if (cancelled) return; cancelled = true; + markUnsettled(correlationId, knownCallId); if (canReject) { parkedAgent.deliver( decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE), @@ -478,12 +521,15 @@ export function createApprovalResume(args: { const dropParked = (): void => { args.onDropped?.(APPROVAL_DROPPED_NOTICE); cancelParked(); + // A bump while the call id was resolving recorded the drop without it. + markUnsettled(correlationId, knownCallId); }; args.registerParkedCancel?.(cancelParked); let overlayAbort: AbortController | undefined; try { // The resolver captures the paired store synchronously before its first await. const parkedCallId = await args.resolveParkedCallId(correlationId); + knownCallId = parkedCallId; if (!stillCurrent()) { dropParked(); return true; @@ -638,6 +684,37 @@ export function createApprovalResume(args: { : inflight.has(correlationId) ? "in-flight" : "idle", + droppedCalls, + prepareRebuild: async (storage) => { + for (const [correlationId, known] of [...unsettled]) { + const callId = + known ?? (await resolveParkedCallIdFromStore(storage, correlationId)); + if (callId === undefined) continue; + markUnsettled(correlationId, callId); + silentCalls.add(callId); + } + }, + settleOn: async (agent, storage) => { + for (const correlationId of [...unsettled.keys()]) { + const callId = await resolveParkedCallIdFromStore( + storage, + correlationId, + ); + if (callId === undefined) { + unsettled.delete(correlationId); + continue; + } + unsettled.set(correlationId, callId); + silentCalls.add(callId); + agent.deliver( + decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE), + ); + } + }, + forgetUnsettled: () => { + unsettled.clear(); + silentCalls.clear(); + }, handle: (result) => { if (result.type !== "suspended") return Promise.resolve(false); const { correlationId } = result; diff --git a/src/session/assemble-runtime.ts b/src/session/assemble-runtime.ts index 076856462..e06fef922 100644 --- a/src/session/assemble-runtime.ts +++ b/src/session/assemble-runtime.ts @@ -55,7 +55,11 @@ import { stickyExtraInstructionsFromRecords, lastCycleSourceFromRunModel, } from "../agent/compaction.js"; -import { createChatDirector, type ChatDirector } from "../agent/director.js"; +import { + createChatDirector, + type ChatDirector, + type DroppedApprovalCalls, +} from "../agent/director.js"; import { createDoomLoopCorrectiveNote } from "../agent/doom-loop-note.js"; import type { AgentToolset } from "../agent/tools.js"; import { createAgentWithLiveToolDispatch } from "../agent/live-tool-dispatch.js"; @@ -555,6 +559,8 @@ export interface ChatAgentWiring { * declined this turn. Same-turn reactor retries still short-circuit. */ clearDenials?: () => void; + /** The session's dropped approvals; every build's director claims from it. */ + droppedApprovalCalls?: DroppedApprovalCalls; getProvider: () => { providerName: string; model: string }; /** Pre-created holder so the workflow controller can close over it first. */ directorHolder?: { instance?: ChatDirector }; @@ -672,6 +678,7 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent { if (tasksFromPrev !== undefined) d.restoreTasks(tasksFromPrev); if (wiring.clearDenials !== undefined) d.setClearDenials(wiring.clearDenials); + d.setDroppedApprovalCalls(wiring.droppedApprovalCalls); directorHolder.instance = d; return d; }, diff --git a/src/tui/runner/exit.test.ts b/src/tui/runner/exit.test.ts index 462999141..3e5ba3280 100644 --- a/src/tui/runner/exit.test.ts +++ b/src/tui/runner/exit.test.ts @@ -10,7 +10,11 @@ import { readSourceCredentialMaterial, registerSourceCredentialRecord, } from "../../config/source-credentials.js"; -import { createChatDirector, type ChatDirector } from "../../agent/director.js"; +import { + createChatDirector, + DROPPED_APPROVAL_SETTLED_EVENT, + type ChatDirector, +} from "../../agent/director.js"; import * as sessionIndex from "../../session/index.js"; import { createSubAgentSessionStore, @@ -333,6 +337,11 @@ function stubSendLifecycle(agent: Agent): { directorHolder: {}, subAgentSessions: { cancelAll: async () => [], list: () => [] }, activeRunHandle: { task: "", startedAt: 0, model: "" }, + approvalResume: { + prepareRebuild: async () => undefined, + settleOn: async () => undefined, + forgetUnsettled: () => undefined, + }, } as unknown as RunnerServices; return { state, services }; } @@ -915,6 +924,66 @@ describe("rebuild re-syncs idle-with-fleet", () => { }); }); +/** + * Records, in order, each agent build and the rebuild hooks the runner calls + * on the approval resume. A settle notes whether it got the newest agent and + * the session's current store. + */ +function recordRebuildHooks( + state: RunnerState, + services: RunnerServices, +): string[] { + const calls: string[] = []; + let newest: unknown; + services.buildAgent = (async () => { + calls.push("build"); + newest = recordingAgent([]); + return newest; + }) as unknown as RunnerServices["buildAgent"]; + services.approvalResume = { + prepareRebuild: async (storage: unknown) => { + calls.push( + storage === state.currentStorage ? "prepare" : "prepare-elsewhere", + ); + }, + settleOn: async (agent: unknown, storage: unknown) => { + calls.push( + agent === newest && storage === state.currentStorage + ? "settle" + : "settle-elsewhere", + ); + }, + forgetUnsettled: () => calls.push("forget"), + } as unknown as RunnerServices["approvalResume"]; + state.currentStorage = {} as RunnerState["currentStorage"]; + return calls; +} + +describe("approvals dropped by a generation bump", () => { + test("an interrupt prepares the drops before building, then settles them on the new agent", async () => { + const directorHolder: RunnerServices["directorHolder"] = {}; + const agent = recordingAgent([]); + const { state, services } = stubSendLifecycle(agent); + wireRebuildServices(services, directorHolder, agent); + await createRunLifecycle(state, services); + const calls = recordRebuildHooks(state, services); + + defined(state.interrupt, "interrupt")(); + await services.sessionOps.awaitTail(); + + expect(calls).toEqual(["prepare", "build", "settle"]); + }); + + test("rotation forgets approvals dropped in the old session", () => { + const { state, services } = stubSendLifecycle(recordingAgent([])); + const calls = recordRebuildHooks(state, services); + + resetSessionForRotation(state, services); + + expect(calls).toEqual(["forget"]); + }); +}); + describe("reload-if-idle waits for the reactor's message run", () => { /** * An agent whose stream the test feeds one event at a time; close() ends @@ -1111,6 +1180,40 @@ describe("reload-if-idle waits for the reactor's message run", () => { expect(state.fatalBuildError).toBeNull(); }); + test("a dropped approval settled with no turn releases the hold its correlation set", async () => { + const { state, services, push, builds } = await idleLifecycle(); + push({ + type: "message.correlated", + seq: 3, + data: { message: {}, correlationId: "corr-1" }, + }); + await drainStream(); + state.pendingReload = true; + defined(state.reloadIfIdle, "reloadIfIdle")(); + await services.sessionOps.awaitTail(); + expect(builds()).toBe(1); + push({ + type: DROPPED_APPROVAL_SETTLED_EVENT, + seq: 4, + data: { callId: "call-ask" }, + }); + await drainStream(); + await services.sessionOps.awaitTail(); + expect(builds()).toBe(2); + expect(state.pendingReload).toBe(false); + }); + + test("an MCP rebuild prepares the drops before building, then settles them on the new agent", async () => { + const { state, services } = await idleLifecycle(); + const calls = recordRebuildHooks(state, services); + + state.pendingReload = true; + defined(state.reloadIfIdle, "reloadIfIdle")(); + await services.sessionOps.awaitTail(); + + expect(calls).toEqual(["prepare", "build", "settle"]); + }); + /** Late MCP connects that mount tools and report the change. */ function wireLateMcp( state: RunnerState, diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index d6e8b4934..8e89e4594 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -13,6 +13,7 @@ import { import { getLogger } from "@intx/log"; import { consumeStream } from "../../session/stream-consumer.js"; import { COMPACTION_CONTINUATION_EVENT } from "../../agent/compaction.js"; +import { DROPPED_APPROVAL_SETTLED_EVENT } from "../../agent/director.js"; import { buildCompactionContinuationMessage, createContinuationGate, @@ -38,6 +39,7 @@ import { truncateSessionLabel, } from "../../session/session-label.js"; import { printResumeHint } from "../../session/resume-hint.js"; +import type { ApprovalResume } from "../../session/approval-resume.js"; import { clearActiveDisposeHost } from "../../session/active-host.js"; import { syncRunStateHandle } from "../../session/active-run.js"; import { startRunHeartbeat } from "../../session/run-liveness.js"; @@ -73,11 +75,14 @@ export function resetSessionForRotation( services: Pick< RunnerServices, "deliveryGeneration" | "emitter" | "subAgentSessions" - >, + > & { approvalResume: Pick }, ): Promise { let cancelledWorkers: Promise = Promise.resolve([]); const reset = (): void => { services.deliveryGeneration.bump(); + // Nothing in the new session can settle the old one's parked call; it + // stays parked on disk, exactly as a quit leaves it. + services.approvalResume.forgetUnsettled(); cancelFeedbackCapture(); services.emitter.emit("session.clear"); cancelledWorkers = services.subAgentSessions.cancelAll("Session cleared"); @@ -377,8 +382,9 @@ export async function createRunLifecycle( // correlated decision without opening a message run, so no bracket covers // the re-dispatched tool or the inference after it. Hold rebuilds from // that correlation until the reactor reaches a terminal it announces: a - // reply, a run end, a fatal error, or shutdown. A bare `wait` terminal - // emits nothing, so the hold then lasts until the next run ends. + // reply, a run end, a dropped approval's silent settle, a fatal error, or + // shutdown. Any other bare `wait` terminal emits nothing, so the hold then + // lasts until the next run ends. let resumingOutsideRun = false; const streamSink = ( event: Parameters[0], @@ -477,6 +483,7 @@ export async function createRunLifecycle( } if ( event.type === "connector.reply" || + event.type === DROPPED_APPROVAL_SETTLED_EVENT || event.type === "reactor.done" || (event.type === "reactor.error" && event.data.fatal) ) { @@ -508,6 +515,24 @@ 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. + const prepareDrops = async (): Promise => { + const storage = state.currentStorage; + if (storage === null) return; + await services.approvalResume + .prepareRebuild(storage) + .catch(caughtDebug(tuiLogger, "preparing dropped approvals failed")); + }; + const settleDrops = async (): Promise => { + const storage = state.currentStorage; + if (storage === null) return; + await services.approvalResume + .settleOn(liveAgent(state), storage) + .catch(caughtDebug(tuiLogger, "settling dropped approvals failed")); + }; const agentBusy = (): boolean => state.inFlight > 0 || openMessageRuns.size > 0 || resumingOutsideRun; @@ -560,6 +585,7 @@ 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( @@ -576,6 +602,7 @@ export async function createRunLifecycle( ); // The rebuild made a fresh director; re-attach the active workflow. services.workflowHost.reattach(); + await settleDrops(); } catch (err) { recordRunError(state, err); state.fatalBuildError = agentRebuildFailure(err); @@ -711,6 +738,7 @@ 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), @@ -733,6 +761,7 @@ export async function createRunLifecycle( streamSink, ); services.workflowHost.reattach(); + await settleDrops(); state.fatalBuildError = null; } catch (err) { recordRunError(state, err); diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index ed9dc2123..48989b994 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -723,6 +723,7 @@ export async function assembleTUISession( // before any worker ever spawns. allowIdleWithFleet: false, clearDenials: () => permissionGate.clearDenials(), + droppedApprovalCalls: approvalResume.droppedCalls, getProvider: () => state.config, directorHolder, getWorkdir: () => state.workdir, diff --git a/src/tui/runner/wiring.ask-wake.test.ts b/src/tui/runner/wiring.ask-wake.test.ts index 22ec1b832..902b2b244 100644 --- a/src/tui/runner/wiring.ask-wake.test.ts +++ b/src/tui/runner/wiring.ask-wake.test.ts @@ -222,7 +222,12 @@ for (const phase of ["settled", "prequeued", "deferred"] as const) { resetting = true; resetSessionForRotation( { withFleetPublicationSuspended: publisher.withSuspended }, - { deliveryGeneration, emitter, subAgentSessions: store }, + { + approvalResume: { forgetUnsettled: () => undefined }, + deliveryGeneration, + emitter, + subAgentSessions: store, + }, ); resetting = false; expect(scheduled).toHaveLength(beforeScheduled);