From e0a82f7d15145609fd0d0269435e6665f9863506 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 21:05:50 -0700 Subject: [PATCH 1/7] Late-bind operator approval to the live agent --- src/session/approval-resume.ts | 43 +++++-- src/tui/runner/session.ts | 16 ++- .../reactor-approval-suspend.test.ts | 27 +++++ tests/unit/approval-resume.test.ts | 109 +++++++++++++++++- 4 files changed, 179 insertions(+), 16 deletions(-) diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index f21556b31..1ccf19bea 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -108,40 +108,63 @@ function decisionMessage( } export function createApprovalResume(args: { - // Late-bound: the live agent is read at handle() time so rebuilds - // (/clear, model switch) deliver through the current instance. + // Late-bound: the live agent is read at history/deliver time so rebuilds + // (/clear, model switch) deliver through the current instance, not a + // snapshot taken at handle() start. getAgent: () => Pick | undefined; + // TUI session queue. When present, each decision is awaited through this + // seam; exec omits it and uses getAgent().deliver. + deliver?: (message: InboundMessage) => void | Promise; gate: PermissionGate; }): ApprovalResume { const { getAgent, gate } = args; + + const requireAgent = (): Pick => { + const agent = getAgent(); + if (agent === undefined) { + throw new Error("approval resume: no live agent"); + } + return agent; + }; + + const deliverDecision = async (message: InboundMessage): Promise => { + if (args.deliver !== undefined) { + await args.deliver(message); + return; + } + requireAgent().deliver(message); + }; + return { handle: async (result) => { if (result.type !== "suspended") return false; - const agent = getAgent(); - if (agent === undefined) return true; const { correlationId, approvalSnapshot } = result; // Turn-count watermark for the settled guard below: a "approval timed // out" tool result appended after this point means the reactor settled // this very correlation before our decision lands. - const turnsAtSuspend = (await agent.history()).length; + const turnsAtSuspend = (await requireAgent().history()).length; if (approvalSnapshot === undefined) { // A suspension without a snapshot cannot be surfaced; fail closed by // rejecting the parked call so the run does not hang on an invisible // gate. - agent.deliver(decisionMessage(correlationId, "rejected", "approval surface unavailable")); + await deliverDecision( + decisionMessage(correlationId, "rejected", "approval surface unavailable"), + ); return true; } const request = requestFromApprovalSnapshot(approvalSnapshot, correlationId); if (request === null) { - agent.deliver(decisionMessage(correlationId, "rejected", "approval surface unavailable")); + await deliverDecision( + decisionMessage(correlationId, "rejected", "approval surface unavailable"), + ); return true; } const outcome = await gate.resolveSuspended(request); - if (settledAfterSuspend(await agent.history(), turnsAtSuspend)) { + if (settledAfterSuspend(await requireAgent().history(), turnsAtSuspend)) { // The reactor already answered the parked call (its approval timeout // fired while the surface was still up). Delivering now would append // the raw decision JSON as an uncorrelated user turn — drop and log. @@ -149,10 +172,10 @@ export function createApprovalResume(args: { return true; } if (outcome === undefined || !outcome.allow) { - agent.deliver(decisionMessage(correlationId, "rejected", outcome?.message)); + await deliverDecision(decisionMessage(correlationId, "rejected", outcome?.message)); return true; } - agent.deliver(decisionMessage(correlationId, "approved")); + await deliverDecision(decisionMessage(correlationId, "approved")); return true; }, }; diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 832ec8901..d2d05bc86 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -143,10 +143,6 @@ export async function assembleTUISession( reactorGated: true, onGrant: (approval, covers) => emitter.emit("permission.grant", { approval, covers }), }); - const approvalResume = createApprovalResume({ - getAgent: () => state.currentAgent, - gate: permissionGate, - }); const permissionsAdmin = createPermissionsAdmin(permissionGate, config.cwd); @@ -377,6 +373,18 @@ export async function assembleTUISession( // so a rebuild never races an in-flight deliver. const sessionOps = createSessionOperationQueue(); const deliveryGeneration = createDeliveryGeneration(); + const approvalResume = createApprovalResume({ + getAgent: () => state.agentProxy ?? state.currentAgent, + deliver: (message) => { + const stillCurrent = deliveryGeneration.capture(); + return sessionOps.enqueue(async () => { + if (!stillCurrent()) return; + if (state.fatalBuildError !== null) throw state.fatalBuildError; + liveAgent(state).deliver(message); + }); + }, + gate: permissionGate, + }); state.enqueueAgentDeliver = (deliverToLiveAgent: () => void): void => { const stillCurrent = deliveryGeneration.capture(); void sessionOps.enqueue(async () => { diff --git a/tests/integration/reactor-approval-suspend.test.ts b/tests/integration/reactor-approval-suspend.test.ts index 076b50f09..11a0104c5 100644 --- a/tests/integration/reactor-approval-suspend.test.ts +++ b/tests/integration/reactor-approval-suspend.test.ts @@ -92,10 +92,12 @@ describe("integration — reactor approval suspend/resume", () => { const handling = resume.handle(result); await waitForAsk(ctx); expect(ctx.asks.length).toBe(1); + const started = Date.now(); ctx.approve(); expect(await handling).toBe(true); const reply = await turn.reply(); + expect(Date.now() - started).toBeLessThan(1000); // The parked call was re-dispatched and actually executed (approvedOnce // bypass, real tool.start) without a second ask, and its result is not // a permission denial. @@ -132,10 +134,12 @@ describe("integration — reactor approval suspend/resume", () => { const resume = createApprovalResume({ getAgent: () => session.agent, gate: ctx.gate }); const handling = resume.handle(result); await waitForAsk(ctx); + const started = Date.now(); ctx.reject("not today"); expect(await handling).toBe(true); await turn.reply(); + expect(Date.now() - started).toBeLessThan(1000); // resume.tool_result answers the parked call by committing the result // turn directly (upstream does not emit tool.done for it), so the // approver's reason reaches the model through history. @@ -153,6 +157,29 @@ describe("integration — reactor approval suspend/resume", () => { } }); + test.serial("rebuild-then-approve fails loud when the parked agent is closed", async () => { + const ctx = gateWithDeferredApproval(); + const session = await openWith(ctx.gate); + try { + session.harness.scenario.replyOnce("anthropic", { toolCalls: [CURL_CALL] }); + const turn = await runUntilSuspended(session, "Please fetch example.com."); + const { result } = turn; + expect(result.type).toBe("suspended"); + if (result.type !== "suspended") return; + + const resume = createApprovalResume({ getAgent: () => session.agent, gate: ctx.gate }); + const handling = resume.handle(result); + await waitForAsk(ctx); + await session.agent.close(); + const started = Date.now(); + ctx.approve(); + await expect(handling).rejects.toThrow(); + expect(Date.now() - started).toBeLessThan(1000); + } finally { + await closeIntegrationSession(session); + } + }); + test.serial( "headless runs deny ask-tier calls as a block without any approval surface", async () => { diff --git a/tests/unit/approval-resume.test.ts b/tests/unit/approval-resume.test.ts index 02ad7e270..dfe18e9cf 100644 --- a/tests/unit/approval-resume.test.ts +++ b/tests/unit/approval-resume.test.ts @@ -1,7 +1,7 @@ import { describe, expect, test } from "bun:test"; -import type { SendResult } from "@intx/agent"; -import type { ConversationTurn } from "@intx/types/runtime"; +import { AgentClosedError, type SendResult } from "@intx/agent"; +import type { ConversationTurn, InboundMessage } from "@intx/types/runtime"; import type { PermissionGate } from "../../src/permission/gate.js"; import { createApprovalResume } from "../../src/session/approval-resume.js"; @@ -49,6 +49,10 @@ function harness(turns: ConversationTurn[], plantTimeoutOnResolve: boolean) { return { agent, gate, delivered }; } +function correlationHeaders(message: unknown) { + return (message as InboundMessage).headers; +} + describe("approval resume late-decision guard", () => { test("a decision after the reactor settled the correlation is dropped", async () => { const turns = [userTurn()]; @@ -68,6 +72,8 @@ describe("approval resume late-decision guard", () => { expect(delivered).toHaveLength(1); const message = delivered[0] as { content: string }; expect(JSON.parse(message.content)).toEqual({ outcome: "rejected", message: "not today" }); + expect(correlationHeaders(delivered[0]).interchangeCorrelationId).toBe("corr-1"); + expect(correlationHeaders(delivered[0]).messageId).toBe("approval-corr-1"); }); test("an approval timeout from before the suspension does not suppress delivery", async () => { @@ -79,3 +85,102 @@ describe("approval resume late-decision guard", () => { expect(delivered).toHaveLength(1); }); }); + +describe("approval resume late-bind", () => { + test("rebuild-during-wait delivers to the agent present after resolveSuspended", async () => { + const deliveredA: unknown[] = []; + const deliveredB: unknown[] = []; + const agentA = { + deliver: (message: unknown) => deliveredA.push(message), + history: async () => [userTurn()], + }; + const agentB = { + deliver: (message: unknown) => deliveredB.push(message), + history: async () => [userTurn()], + }; + let current: typeof agentA | typeof agentB = agentA; + const gate = { + resolveSuspended: async () => { + current = agentB; + return { allow: true }; + }, + } as unknown as PermissionGate; + const resume = createApprovalResume({ getAgent: () => current, gate }); + + expect(await resume.handle(SUSPENDED)).toBe(true); + expect(deliveredA).toEqual([]); + expect(deliveredB).toHaveLength(1); + expect(correlationHeaders(deliveredB[0]).interchangeCorrelationId).toBe("corr-1"); + expect(correlationHeaders(deliveredB[0]).messageId).toBe("approval-corr-1"); + }); + + test("optional deliver is awaited and used instead of getAgent().deliver", async () => { + const agentDelivered: unknown[] = []; + const customDelivered: unknown[] = []; + const agent = { + deliver: (message: unknown) => agentDelivered.push(message), + history: async () => [userTurn()], + }; + let customResolved = false; + const resume = createApprovalResume({ + getAgent: () => agent, + deliver: async (message) => { + await Promise.resolve(); + customResolved = true; + customDelivered.push(message); + }, + gate: { resolveSuspended: async () => ({ allow: true }) } as unknown as PermissionGate, + }); + + expect(await resume.handle(SUSPENDED)).toBe(true); + expect(customResolved).toBe(true); + expect(agentDelivered).toEqual([]); + expect(customDelivered).toHaveLength(1); + expect(correlationHeaders(customDelivered[0]).interchangeCorrelationId).toBe("corr-1"); + }); + + test("undefined agent throws instead of returning true", async () => { + const resume = createApprovalResume({ + getAgent: () => undefined, + gate: { resolveSuspended: async () => ({ allow: true }) } as unknown as PermissionGate, + }); + await expect(resume.handle(SUSPENDED)).rejects.toThrow(/agent/i); + }); + + test("AgentClosedError from deliver is not swallowed", async () => { + const agent = { + deliver: () => { + throw new AgentClosedError(); + }, + history: async () => [userTurn()], + }; + const resume = createApprovalResume({ + getAgent: () => agent, + gate: { resolveSuspended: async () => ({ allow: true }) } as unknown as PermissionGate, + }); + await expect(resume.handle(SUSPENDED)).rejects.toThrow(AgentClosedError); + }); + + test("an intervening user turn after suspend does not drop a live decision", async () => { + const turns = [userTurn()]; + const delivered: unknown[] = []; + const agent = { + deliver: (message: unknown) => delivered.push(message), + history: async () => turns, + }; + const gate = { + resolveSuspended: async () => { + turns.push(userTurn()); + return { allow: true }; + }, + } as unknown as PermissionGate; + const resume = createApprovalResume({ getAgent: () => agent, gate }); + + expect(await resume.handle(SUSPENDED)).toBe(true); + expect(delivered).toHaveLength(1); + expect(JSON.parse((delivered[0] as { content: string }).content)).toEqual({ + outcome: "approved", + }); + expect(correlationHeaders(delivered[0]).interchangeCorrelationId).toBe("corr-1"); + }); +}); From 19bd9471c751024a480462a02adfe33c8433f111 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 21:05:55 -0700 Subject: [PATCH 2/7] Hold the TUI busy across the approval overlay --- CHANGELOG.md | 2 + docs/ARCHITECTURE.md | 2 +- src/tui/runner/exit.ts | 9 +-- src/tui/runner/state.ts | 13 ++++ src/tui/runner/submit.ts | 16 +++-- .../approval-reload-during-suspend.test.ts | 65 +++++++++++++++++++ 6 files changed, 93 insertions(+), 14 deletions(-) create mode 100644 tests/unit/tui/approval-reload-during-suspend.test.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index df506a5a6..5557ff278 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -42,6 +42,8 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename - A session falls back to the next resolvable provider when the selected default or project-local provider is missing or incomplete. `--provider` still errors. +- Operator approval resume late-binds to the live agent and keeps the TUI + busy across the overlay so a reload cannot drop the parked tool call. ### Changed diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 90dba5a6c..dd91d1bd3 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -382,7 +382,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: `skipPermissions` allows everything; `allow`-tier passes; for `ask`-tier, checks persisted approvals, otherwise requests operator approval. Shell security classifies each chain segment (`||` / `&&` / `|` / `;` / newlines), but the operator is prompted once for the full command block — any unapproved segment fails the whole block, and execution always runs the unsplit original. Safe pipeline tails and pure shell no-ops (`true` / `false` / `:` and bare control-flow keywords stranded by chain-splitting) skip without a prompt. In a non-interactive run an unresolved `ask` becomes a denial. In auto mode: non-shell built-ins in `AUTO_ALLOWED_TOOLS` (writes/edits/deletes, `manage_tasks`, `spawn_agent`, `wait_agents`, …) auto-allow when not path-restricted; for `run_shell` the gate consults the auto-shell policy — a `deny` rule fails the call, an `ask` rule skips the auto-allow shortcut and proceeds to the normal approval flow, and anything unmatched is auto-allowed. Paths outside the workspace and writes under the session state root (`~/.corbits/projects/...` and legacy `.agent-state`) still ask under auto mode. Under `--dangerously-skip-permissions` (forces this process) or `/yolo` (persists as the user-global default via `setSkipPermissions`), the gate auto-allows those same cases, and pre-gate sandboxes (path-escape, shell session cwd retention, `list_dir` / `delete_file` workspace bounds) honor `getSkipPermissions()` live so outside-workspace access is not hard-denied after the gate already allowed it — without rebuilding the plugin stack. Secret-guard path denies and authorization hard blocks still apply. Mutating MCP and unknown built-ins are not blanket-allowed outside skip. Newly granted scopes are appended in memory and persisted. - - **Reactor-gated sessions (main session; `reactorGated: true`).** The gate's decision logic lives in one `decide()` used by both consumers: `evaluate()` (the middleware path below, still used by sub-agents) and `authorizeCall()`, which expresses the decision as the vendored reactor's before-tool authz effect (`src/permission/reactor-authorize.ts` bridges it into `env.authorize`). An `ask` there suspends the call as a reactor `PendingOperation` keyed by a correlationId (persisted through the context store's existing `pendingOperations`); `send()` settles as `suspended` and `src/session/approval-resume.ts` rebuilds the operator request from the approval snapshot, resolves it through the same `requestApproval` seam the TUI overlay uses, and delivers the decision to the reactor on the correlationId signal channel — an approved decision grants a one-shot bypass and the exact parked call re-dispatches; a rejected one answers it with an error result. 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. The TUI stays busy across the overlay and late-binds deliver through the live agent / session queue. 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()`. - **Worker reactor ownership.** `workerPermissionGate` is a reactor-gated view over the parent's live permission gate: grants and policy are shared, not copied or toggled. Worker posix plugins and inherited MCP tools are bound to that view at worker start, so they take the reactor-gated `gateToolCall` path because the view reports `isReactorGated()` — they do not close over the parent's middleware-gated `isReactorGated()`. Deny still blocks; ask/allow skip the middleware prompt. `authorizeCall` on the view never emits `ask` — unresolved approvals become denials that name the permission subject, without invoking an approval callback or suspending, even with an interactive parent; the parent can obtain a grant and retry. Worker control-plane tools (`submit_result`, `ask_director`, and nested fleet verbs other than `spawn_agent`) allow without a parent grant. Authorization and tool execution run under the same async-local worker identity and cwd. Fleet authority remains an independent restriction, not an alternative permission grant. - **matcher** — Approval pattern matching via `@intx/authz` `matchPattern` (`*` wildcards). Exact-command grants store a backslash before each metacharacter; those patterns match by equality after unescape (the package has no escape syntax). diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index f4786330c..9debf1e3b 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -30,6 +30,7 @@ import { hostOf, liveAgent, recordRunError, + runWhileAgentBusy, type RunnerServices, type RunnerState, type SnapshotExtra, @@ -339,15 +340,11 @@ export async function createRunLifecycle( services.emitter.emit("session.title", truncateSessionLabel(state.runTaskTitle)); void persistRunSnapshot("running"); } - state.inFlight++; - try { + return await runWhileAgentBusy(state, async () => { await refreshCodexBeforeSend(); await refreshXaiBeforeSend(); return await liveAgent(state).send(content, opts); - } finally { - state.inFlight--; - reloadIfIdle(); - } + }); }, stream: () => liveAgent(state).stream(), deliver: (message) => { diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index b09674c8a..d393c7a5c 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -254,6 +254,19 @@ export function liveAgent(state: RunnerState): Agent { return agent; } +export async function runWhileAgentBusy( + state: Pick, + op: () => Promise, +): Promise { + state.inFlight++; + try { + return await op(); + } finally { + state.inFlight--; + state.reloadIfIdle?.(); + } +} + export function hostOf(state: RunnerState): RunnerHost { const host = state.host; if (host === undefined) { diff --git a/src/tui/runner/submit.ts b/src/tui/runner/submit.ts index ecd1d9f2f..9ffed8154 100644 --- a/src/tui/runner/submit.ts +++ b/src/tui/runner/submit.ts @@ -38,7 +38,7 @@ import { tuiSendFailureMessage } from "./send-failure-message.js"; import type { ProviderFailureAttempt } from "../provider/failure-attempt.js"; import type { Agent } from "@intx/agent"; import { ASK_DIRECTOR_WAKE_PREFIX } from "../../subagent/fleet-report.js"; -import { hostOf, type RunnerServices, type RunnerState } from "./state.js"; +import { hostOf, runWhileAgentBusy, type RunnerServices, type RunnerState } from "./state.js"; import { LOG_NAMESPACE_ROOT } from "../../branding.js"; const tuiLogger = getLogger([LOG_NAMESPACE_ROOT, "tui"]); @@ -259,12 +259,14 @@ export function createSubmitPath( const attempt = live.attemptIdentity(); const providerFailure = services.providerFailureAttempts.begin(attempt); try { - const result = await live.agentProxy.send(message); - // An ask-tier call parked on the reactor's approval gate settles the - // send early; resolve the operator surface here and deliver the - // decision on the correlationId signal channel so the parked run - // resumes. - await services.approvalResume.handle(result); + await runWhileAgentBusy(state, async () => { + const result = await live.agentProxy.send(message); + // An ask-tier call parked on the reactor's approval gate settles the + // send early; resolve the operator surface here and deliver the + // decision on the correlationId signal channel so the parked run + // resumes. + await services.approvalResume.handle(result); + }); return true; } catch (error) { handleSendFailure(error, attempt, providerFailure); diff --git a/tests/unit/tui/approval-reload-during-suspend.test.ts b/tests/unit/tui/approval-reload-during-suspend.test.ts new file mode 100644 index 000000000..3475c1338 --- /dev/null +++ b/tests/unit/tui/approval-reload-during-suspend.test.ts @@ -0,0 +1,65 @@ +import { describe, expect, test } from "bun:test"; + +import { runWhileAgentBusy, type RunnerState } from "../../../src/tui/runner/state.js"; + +function stubBusyState() { + const rebuilds: number[] = []; + const state: Pick & { pendingReload: boolean } = { + inFlight: 0, + pendingReload: false, + reloadIfIdle: () => { + if (!state.pendingReload || state.inFlight > 0) return; + state.pendingReload = false; + rebuilds.push(state.inFlight); + }, + }; + return { state, rebuilds }; +} + +describe("runWhileAgentBusy vs pendingReload", () => { + test("nested spans hold reloadIfIdle until the outer span finishes", async () => { + const { state, rebuilds } = stubBusyState(); + + const result = await runWhileAgentBusy(state, async () => { + return await runWhileAgentBusy(state, async () => { + state.pendingReload = true; + state.reloadIfIdle?.(); + return "suspended"; + }); + }); + + expect(result).toBe("suspended"); + expect(rebuilds).toEqual([0]); + }); + + test("pendingReload during a deferred overlay-shaped outer op rebuilds only after resolve", async () => { + const { state, rebuilds } = stubBusyState(); + let release: (() => void) | undefined; + const deferred = new Promise((resolve) => { + release = resolve; + }); + + const running = runWhileAgentBusy(state, async () => { + state.pendingReload = true; + state.reloadIfIdle?.(); + await deferred; + return "ok"; + }); + + expect(rebuilds).toEqual([]); + expect(state.inFlight).toBe(1); + release?.(); + expect(await running).toBe("ok"); + expect(rebuilds).toEqual([0]); + expect(state.inFlight).toBe(0); + }); + + test("reloadIfIdle is a no-op when inFlight is already greater than zero", () => { + const { state, rebuilds } = stubBusyState(); + state.inFlight = 2; + state.pendingReload = true; + state.reloadIfIdle?.(); + expect(rebuilds).toEqual([]); + expect(state.pendingReload).toBe(true); + }); +}); From d20c5d5d75dc8d3cc93074565062662a4ad75b75 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 21:41:26 -0700 Subject: [PATCH 3/7] Capture approval generation at overlay start /clear during the permission overlay bumped generation after handle started, but the TUI recaptured at deliver time so an accept still injected the approved decision into the empty session. --- src/session/approval-resume.ts | 24 ++++--- src/tui/runner/session.ts | 4 +- tests/unit/approval-resume.test.ts | 53 +++++++++++++++ .../approval-reload-during-suspend.test.ts | 67 +++++++++++++++++++ 4 files changed, 137 insertions(+), 11 deletions(-) diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index 1ccf19bea..71341f351 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -114,7 +114,10 @@ export function createApprovalResume(args: { getAgent: () => Pick | undefined; // TUI session queue. When present, each decision is awaited through this // seam; exec omits it and uses getAgent().deliver. - deliver?: (message: InboundMessage) => void | Promise; + deliver?: (message: InboundMessage, stillCurrent: () => boolean) => void | Promise; + // TUI: capture at handle() start so /clear during the overlay drops the + // decision instead of delivering into the new session. Exec omits this. + captureGeneration?: () => () => boolean; gate: PermissionGate; }): ApprovalResume { const { getAgent, gate } = args; @@ -127,19 +130,21 @@ export function createApprovalResume(args: { return agent; }; - const deliverDecision = async (message: InboundMessage): Promise => { - if (args.deliver !== undefined) { - await args.deliver(message); - return; - } - requireAgent().deliver(message); - }; - return { handle: async (result) => { if (result.type !== "suspended") return false; + const stillCurrent = args.captureGeneration?.() ?? (() => true); const { correlationId, approvalSnapshot } = result; + const deliverDecision = async (message: InboundMessage): Promise => { + if (!stillCurrent()) return; + if (args.deliver !== undefined) { + await args.deliver(message, stillCurrent); + return; + } + requireAgent().deliver(message); + }; + // Turn-count watermark for the settled guard below: a "approval timed // out" tool result appended after this point means the reactor settled // this very correlation before our decision lands. @@ -164,6 +169,7 @@ export function createApprovalResume(args: { } const outcome = await gate.resolveSuspended(request); + if (!stillCurrent()) return true; if (settledAfterSuspend(await requireAgent().history(), turnsAtSuspend)) { // The reactor already answered the parked call (its approval timeout // fired while the surface was still up). Delivering now would append diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index d2d05bc86..a9baef713 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -375,8 +375,8 @@ export async function assembleTUISession( const deliveryGeneration = createDeliveryGeneration(); const approvalResume = createApprovalResume({ getAgent: () => state.agentProxy ?? state.currentAgent, - deliver: (message) => { - const stillCurrent = deliveryGeneration.capture(); + captureGeneration: deliveryGeneration.capture, + deliver: (message, stillCurrent) => { return sessionOps.enqueue(async () => { if (!stillCurrent()) return; if (state.fatalBuildError !== null) throw state.fatalBuildError; diff --git a/tests/unit/approval-resume.test.ts b/tests/unit/approval-resume.test.ts index dfe18e9cf..7f9ffd1ce 100644 --- a/tests/unit/approval-resume.test.ts +++ b/tests/unit/approval-resume.test.ts @@ -5,6 +5,7 @@ import type { ConversationTurn, InboundMessage } from "@intx/types/runtime"; import type { PermissionGate } from "../../src/permission/gate.js"; import { createApprovalResume } from "../../src/session/approval-resume.js"; +import { createDeliveryGeneration } from "../../src/tui/queued-delivery.js"; const SUSPENDED: SendResult = { type: "suspended", @@ -184,3 +185,55 @@ describe("approval resume late-bind", () => { expect(correlationHeaders(delivered[0]).interchangeCorrelationId).toBe("corr-1"); }); }); + +describe("approval resume generation capture", () => { + test("overlay accept after a generation bump does not deliver", async () => { + const generation = createDeliveryGeneration(); + const delivered: unknown[] = []; + const agent = { + deliver: (message: unknown) => delivered.push(message), + history: async () => [userTurn()], + }; + const resume = createApprovalResume({ + getAgent: () => agent, + captureGeneration: generation.capture, + deliver: (message, stillCurrent) => { + if (!stillCurrent()) return; + delivered.push(message); + }, + gate: { + resolveSuspended: async () => { + generation.bump(); + return { allow: true }; + }, + } as unknown as PermissionGate, + }); + + expect(await resume.handle(SUSPENDED)).toBe(true); + expect(delivered).toEqual([]); + }); + + test("a live generation still delivers after the overlay", async () => { + const generation = createDeliveryGeneration(); + const delivered: unknown[] = []; + const agent = { + deliver: (message: unknown) => delivered.push(message), + history: async () => [userTurn()], + }; + const resume = createApprovalResume({ + getAgent: () => agent, + captureGeneration: generation.capture, + deliver: (message, stillCurrent) => { + if (!stillCurrent()) return; + delivered.push(message); + }, + gate: { resolveSuspended: async () => ({ allow: true }) } as unknown as PermissionGate, + }); + + expect(await resume.handle(SUSPENDED)).toBe(true); + expect(delivered).toHaveLength(1); + expect(JSON.parse((delivered[0] as { content: string }).content)).toEqual({ + outcome: "approved", + }); + }); +}); diff --git a/tests/unit/tui/approval-reload-during-suspend.test.ts b/tests/unit/tui/approval-reload-during-suspend.test.ts index 3475c1338..51e21e617 100644 --- a/tests/unit/tui/approval-reload-during-suspend.test.ts +++ b/tests/unit/tui/approval-reload-during-suspend.test.ts @@ -1,5 +1,11 @@ import { describe, expect, test } from "bun:test"; +import type { SendResult } from "@intx/agent"; +import type { ConversationTurn } from "@intx/types/runtime"; + +import type { PermissionGate } from "../../../src/permission/gate.js"; +import { createApprovalResume } from "../../../src/session/approval-resume.js"; +import { createSessionOperationQueue } from "../../../src/tui/session-operation-queue.js"; import { runWhileAgentBusy, type RunnerState } from "../../../src/tui/runner/state.js"; function stubBusyState() { @@ -63,3 +69,64 @@ describe("runWhileAgentBusy vs pendingReload", () => { expect(state.pendingReload).toBe(true); }); }); + +const SUSPENDED: SendResult = { + type: "suspended", + correlationId: "corr-1", + approvalSnapshot: { name: "run_shell", arguments: { command: "curl -sS https://example.com" } }, +} as unknown as SendResult; + +function userTurn(): ConversationTurn { + return { + role: "user", + content: [{ type: "text", text: "go" }], + timestamp: 0, + } as unknown as ConversationTurn; +} + +describe("pendingReload during resolveSuspended vs deliver enqueue", () => { + test("does not rebuild until handle returns and deliver has run", async () => { + const events: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + const state: Pick & { pendingReload: boolean } = { + inFlight: 0, + pendingReload: false, + reloadIfIdle: () => { + if (!state.pendingReload || state.inFlight > 0) return; + state.pendingReload = false; + void enqueue(async () => { + events.push("rebuild"); + }); + }, + }; + const agent = { + deliver: (_message: unknown) => { + events.push("deliver"); + }, + history: async () => [userTurn()], + }; + const resume = createApprovalResume({ + getAgent: () => agent, + deliver: (message) => + enqueue(async () => { + agent.deliver(message); + }), + gate: { + resolveSuspended: async () => { + state.pendingReload = true; + state.reloadIfIdle?.(); + expect(events).toEqual([]); + return { allow: true }; + }, + } as unknown as PermissionGate, + }); + + await runWhileAgentBusy(state, async () => { + await resume.handle(SUSPENDED); + }); + await awaitTail(); + + expect(events).toEqual(["deliver", "rebuild"]); + expect(state.inFlight).toBe(0); + }); +}); From fee955834f1d17331d29bcc90c27a8294fd738e1 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 22:00:55 -0700 Subject: [PATCH 4/7] Drop overlay approval after interrupt rebuilds the agent Interrupt rebuilt the agent while the permission overlay stayed open, so a later accept or decline still saw a current generation and delivered into the new agent. Bump delivery generation on interrupt the same way session rotation does. --- src/session/approval-resume.ts | 4 +-- src/tui/runner/exit.ts | 3 +++ tests/unit/approval-resume.test.ts | 42 ++++++++++++++++++++++++++++++ 3 files changed, 47 insertions(+), 2 deletions(-) diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index 71341f351..b1b8a4708 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -115,8 +115,8 @@ export function createApprovalResume(args: { // TUI session queue. When present, each decision is awaited through this // seam; exec omits it and uses getAgent().deliver. deliver?: (message: InboundMessage, stillCurrent: () => boolean) => void | Promise; - // TUI: capture at handle() start so /clear during the overlay drops the - // decision instead of delivering into the new session. Exec omits this. + // TUI: capture at handle() start so /clear or interrupt during the overlay + // drops the decision instead of delivering into the rebuilt agent. Exec omits this. captureGeneration?: () => () => boolean; gate: PermissionGate; }): ApprovalResume { diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index 9debf1e3b..0dd3426f6 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -401,6 +401,9 @@ export async function createRunLifecycle( // Close it, drain the old stream, and rebuild a fresh agent so the next send // works. const interrupt = (): void => { + // Overlay stays open across interrupt; bump so a later accept/decline + // cannot late-bind into the rebuilt agent. + services.deliveryGeneration.bump(); state.sendAborted = true; void enqueueOp(async () => { try { diff --git a/tests/unit/approval-resume.test.ts b/tests/unit/approval-resume.test.ts index 7f9ffd1ce..d5b0f36cd 100644 --- a/tests/unit/approval-resume.test.ts +++ b/tests/unit/approval-resume.test.ts @@ -236,4 +236,46 @@ describe("approval resume generation capture", () => { outcome: "approved", }); }); + + async function interruptDuringOverlayThenDecide(outcome: { allow: boolean; message?: string }) { + const generation = createDeliveryGeneration(); + const deliveredA: unknown[] = []; + const deliveredB: unknown[] = []; + const agentA = { + deliver: (message: unknown) => deliveredA.push(message), + history: async () => [userTurn()], + }; + const agentB = { + deliver: (message: unknown) => deliveredB.push(message), + history: async () => [userTurn()], + }; + let current: typeof agentA | typeof agentB = agentA; + const resume = createApprovalResume({ + getAgent: () => current, + captureGeneration: generation.capture, + deliver: (message, stillCurrent) => { + if (!stillCurrent()) return; + current.deliver(message); + }, + gate: { + resolveSuspended: async () => { + generation.bump(); + current = agentB; + return outcome; + }, + } as unknown as PermissionGate, + }); + + expect(await resume.handle(SUSPENDED)).toBe(true); + expect(deliveredA).toEqual([]); + expect(deliveredB).toEqual([]); + } + + test("interrupt during overlay then accept does not deliver to the rebuilt agent", async () => { + await interruptDuringOverlayThenDecide({ allow: true }); + }); + + test("interrupt during overlay then decline does not deliver to the rebuilt agent", async () => { + await interruptDuringOverlayThenDecide({ allow: false, message: "not today" }); + }); }); From 32ee8ae4133091b73c90084ee6ca2fea48d3a92b Mon Sep 17 00:00:00 2001 From: Sawyer Date: Tue, 8 Sep 2026 22:30:21 -0700 Subject: [PATCH 5/7] Assert interrupt bumps delivery generation before rebuild --- src/tui/runner/exit.ts | 75 ++++++++++++++++++++++------------- tests/unit/tui/runner.test.ts | 26 ++++++++++++ 2 files changed, 73 insertions(+), 28 deletions(-) diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index 0dd3426f6..054e7aa6c 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -120,6 +120,22 @@ export function agentRebuildFailure(err: unknown): Error { : new Error(String(err)); } +/** + * Hard-stop interrupt: bump delivery generation, then enqueue the agent rebuild. + * Overlay stays open across interrupt; the bump must happen before enqueue so a + * later accept/decline cannot late-bind into the rebuilt agent. + */ +export function startInterruptRebuild(args: { + deliveryGeneration: { bump: () => void }; + markSendAborted: () => void; + enqueue: (op: () => Promise) => unknown; + rebuild: () => Promise; +}): void { + args.deliveryGeneration.bump(); + args.markSendAborted(); + void args.enqueue(args.rebuild); +} + export function clearsActiveRun(kind: SnapshotKind): boolean { return kind === "run-end"; } @@ -401,36 +417,39 @@ export async function createRunLifecycle( // Close it, drain the old stream, and rebuild a fresh agent so the next send // works. const interrupt = (): void => { - // Overlay stays open across interrupt; bump so a later accept/decline - // cannot late-bind into the rebuilt agent. - services.deliveryGeneration.bump(); - state.sendAborted = true; - void enqueueOp(async () => { - try { - // close() tears down stream consumers before the aborted cycle's - // inference.error is delivered, so the recorder never sees a terminal - // event for the dead cycle — dispose closes it against stray deltas - // and salvages the buffer before that teardown, so it is never lost - // or misattributed to the rebuilt agent's next cycle. - await services.cycleRecorder.dispose("interrupted"); - const closedCleanly = await closeAgentForRebuild(liveAgent(state), "interrupt"); - await state.streamPromise?.catch((err: unknown) => { - tuiLogger.debug("stream drain during interrupt teardown failed: {error}", { - error: err instanceof Error ? err.message : String(err), + startInterruptRebuild({ + deliveryGeneration: services.deliveryGeneration, + markSendAborted: () => { + state.sendAborted = true; + }, + enqueue: enqueueOp, + rebuild: async () => { + try { + // close() tears down stream consumers before the aborted cycle's + // inference.error is delivered, so the recorder never sees a terminal + // event for the dead cycle — dispose closes it against stray deltas + // and salvages the buffer before that teardown, so it is never lost + // or misattributed to the rebuilt agent's next cycle. + await services.cycleRecorder.dispose("interrupted"); + const closedCleanly = await closeAgentForRebuild(liveAgent(state), "interrupt"); + await state.streamPromise?.catch((err: unknown) => { + tuiLogger.debug("stream drain during interrupt teardown failed: {error}", { + error: err instanceof Error ? err.message : String(err), + }); }); - }); - if (!closedCleanly) { - throw new AgentContextLockError(state.workdir); + if (!closedCleanly) { + throw new AgentContextLockError(state.workdir); + } + state.currentAgent = await services.buildAgent(); + services.cycleRecorder.reset(); + state.streamPromise = consumeStream(liveAgent(state).stream(), streamSink); + services.workflowController.reattach(); + state.fatalBuildError = null; + } catch (err) { + recordRunError(state, err); + state.fatalBuildError = agentRebuildFailure(err); } - state.currentAgent = await services.buildAgent(); - services.cycleRecorder.reset(); - state.streamPromise = consumeStream(liveAgent(state).stream(), streamSink); - services.workflowController.reattach(); - state.fatalBuildError = null; - } catch (err) { - recordRunError(state, err); - state.fatalBuildError = agentRebuildFailure(err); - } + }, }); }; state.interrupt = interrupt; diff --git a/tests/unit/tui/runner.test.ts b/tests/unit/tui/runner.test.ts index b440dd48f..846e064ac 100644 --- a/tests/unit/tui/runner.test.ts +++ b/tests/unit/tui/runner.test.ts @@ -5,6 +5,7 @@ import { agentRebuildFailure, closeAgentForRebuild, resumeTranscriptLoadErrorBlock, + startInterruptRebuild, } from "../../../src/tui/runner/exit.js"; import { createTUIEventEmitter, getTUIRunSummaryStatus } from "../../../src/tui/runner/index.js"; import { loadLocalSettingsWriteBase } from "../../../src/tui/runner/settings.js"; @@ -287,3 +288,28 @@ test("a rejecting reload op through the real session-operation-queue never trigg // failing a single assertion. The test above is the harness-compatible half // of that pair: same real queue, same real helpers, proving the fixed shape // produces no such failure. + +// Overlay accept/decline tests stub bump() inside resolveSuspended, so deleting +// the interrupt-site bump would not fail them. Drive the interrupt helper itself. +test("interrupt bumps delivery generation before enqueueing rebuild", () => { + const order: string[] = []; + startInterruptRebuild({ + deliveryGeneration: { + bump: () => { + order.push("bump"); + }, + }, + markSendAborted: () => { + order.push("abort"); + }, + enqueue: (op) => { + order.push("enqueue"); + return op(); + }, + rebuild: async () => { + order.push("rebuild"); + }, + }); + expect(order[0]).toBe("bump"); + expect(order.indexOf("enqueue")).toBeGreaterThan(0); +}); From 7cec6905b2076a5f4ed3d6d735c3a06d8f7ff2ca Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 11:29:43 -0700 Subject: [PATCH 6/7] Drop operator approval when session identity changes --- CHANGELOG.md | 4 + docs/ARCHITECTURE.md | 2 +- src/permission/gate.ts | 42 ++-- src/session/approval-resume.ts | 23 +- src/tui/correlation-acceptance.test.ts | 25 ++ src/tui/correlation-acceptance.ts | 33 +++ src/tui/gate-wire.test.ts | 35 +++ src/tui/gate-wire.ts | 9 +- src/tui/queued-delivery.test.ts | 11 + src/tui/queued-delivery.ts | 12 +- src/tui/request-approval.test.ts | 24 ++ src/tui/request-approval.ts | 15 +- src/tui/runner/exit.ts | 8 +- src/tui/runner/session.ts | 19 +- src/tui/runner/state.ts | 3 + tests/unit/approval-resume.test.ts | 310 ++++++++++++++++++++++--- 16 files changed, 513 insertions(+), 62 deletions(-) create mode 100644 src/tui/correlation-acceptance.test.ts create mode 100644 src/tui/correlation-acceptance.ts diff --git a/CHANGELOG.md b/CHANGELOG.md index 5557ff278..c057ab62b 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -44,6 +44,10 @@ parallel copies under `docs/` or `scripts/notes/`. At cut time: rename still errors. - Operator approval resume late-binds to the live agent and keeps the TUI busy across the overlay so a reload cannot drop the parked tool call. +- Operator approval drops on session identity change. inFlight occupancy + owns idle rebuild; delivery generation owns session identity, so interrupt, + /clear, and /new abort the outstanding overlay, skip minting a grant, and + notify the operator instead of delivering into a rebuilt agent. ### Changed diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index dd91d1bd3..ce5e45b27 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -382,7 +382,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: `skipPermissions` allows everything; `allow`-tier passes; for `ask`-tier, checks persisted approvals, otherwise requests operator approval. Shell security classifies each chain segment (`||` / `&&` / `|` / `;` / newlines), but the operator is prompted once for the full command block — any unapproved segment fails the whole block, and execution always runs the unsplit original. Safe pipeline tails and pure shell no-ops (`true` / `false` / `:` and bare control-flow keywords stranded by chain-splitting) skip without a prompt. In a non-interactive run an unresolved `ask` becomes a denial. In auto mode: non-shell built-ins in `AUTO_ALLOWED_TOOLS` (writes/edits/deletes, `manage_tasks`, `spawn_agent`, `wait_agents`, …) auto-allow when not path-restricted; for `run_shell` the gate consults the auto-shell policy — a `deny` rule fails the call, an `ask` rule skips the auto-allow shortcut and proceeds to the normal approval flow, and anything unmatched is auto-allowed. Paths outside the workspace and writes under the session state root (`~/.corbits/projects/...` and legacy `.agent-state`) still ask under auto mode. Under `--dangerously-skip-permissions` (forces this process) or `/yolo` (persists as the user-global default via `setSkipPermissions`), the gate auto-allows those same cases, and pre-gate sandboxes (path-escape, shell session cwd retention, `list_dir` / `delete_file` workspace bounds) honor `getSkipPermissions()` live so outside-workspace access is not hard-denied after the gate already allowed it — without rebuilding the plugin stack. Secret-guard path denies and authorization hard blocks still apply. Mutating MCP and unknown built-ins are not blanket-allowed outside skip. Newly granted scopes are appended in memory and persisted. - - **Reactor-gated sessions (main session; `reactorGated: true`).** The gate's decision logic lives in one `decide()` used by both consumers: `evaluate()` (the middleware path below, still used by sub-agents) and `authorizeCall()`, which expresses the decision as the vendored reactor's before-tool authz effect (`src/permission/reactor-authorize.ts` bridges it into `env.authorize`). An `ask` there suspends the call as a reactor `PendingOperation` keyed by a correlationId (persisted through the context store's existing `pendingOperations`); `send()` settles as `suspended` and `src/session/approval-resume.ts` rebuilds the operator request from the approval snapshot, resolves it through the same `requestApproval` seam the TUI overlay uses, and delivers the decision to the reactor on the correlationId signal channel — an approved decision grants a one-shot bypass and the exact parked call re-dispatches; a rejected one answers it with an error result. The TUI stays busy across the overlay and late-binds deliver through the live agent / session queue. 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. `inFlight` occupancy owns idle rebuild: the TUI stays busy across the overlay and waits until the correlated resume is accepted (`message.received` / `message.correlated`) or a generation bump `settleAll`s the waiter. Delivery generation owns session identity: interrupt, `/clear`, and `/new` abort the outstanding overlay, skip minting a grant, drop the decision, and surface an operator notice rather than delivering into a rebuilt agent. Under reactor gating the middleware/MCP `gateToolCall` is an execution backstop, not a second copy of `env.authorize`: it consumes the `authorizeCall` verdict only when id, name, and arguments match, and does not re-decide. Deny still blocks and does not call `next`; an `ask` or `allow` skips the middleware prompt so an approved re-dispatch never re-asks. A reused `codex-proxy` id cannot apply an outer `shell` allow to an inner `run_shell` deny. Inner posix runs whose outer tool is not `run_shell` (Codex `apply_patch` proxy) never pass `env.authorize`, so `gateToolCall` decides on that cache miss and still blocks a deny. The headless denial and the stricter chained-command hard-deny are preserved as deny effects (upstream `block`s) decided inside the same `decide()`. - **Worker reactor ownership.** `workerPermissionGate` is a reactor-gated view over the parent's live permission gate: grants and policy are shared, not copied or toggled. Worker posix plugins and inherited MCP tools are bound to that view at worker start, so they take the reactor-gated `gateToolCall` path because the view reports `isReactorGated()` — they do not close over the parent's middleware-gated `isReactorGated()`. Deny still blocks; ask/allow skip the middleware prompt. `authorizeCall` on the view never emits `ask` — unresolved approvals become denials that name the permission subject, without invoking an approval callback or suspending, even with an interactive parent; the parent can obtain a grant and retry. Worker control-plane tools (`submit_result`, `ask_director`, and nested fleet verbs other than `spawn_agent`) allow without a parent grant. Authorization and tool execution run under the same async-local worker identity and cwd. Fleet authority remains an independent restriction, not an alternative permission grant. - **matcher** — Approval pattern matching via `@intx/authz` `matchPattern` (`*` wildcards). Exact-command grants store a backslash before each metacharacter; those patterns match by equality after unescape (the package has no escape syntax). diff --git a/src/permission/gate.ts b/src/permission/gate.ts index a472eee8d..7d3b1a068 100644 --- a/src/permission/gate.ts +++ b/src/permission/gate.ts @@ -67,7 +67,9 @@ function classifyOutcome(outcome: ApprovalOutcome | undefined): ApprovalOutcomeK if (!outcome.allow) { const message = outcome.message ?? ""; if (message.includes("timed out")) return "timeout"; - if (message.includes("no longer running")) return "abort"; + if (message.includes("no longer running") || message.includes("identity changed")) { + return "abort"; + } return "deny"; } return outcome.persist !== undefined ? "allow-with-scope" : "allow-once"; @@ -325,8 +327,12 @@ export interface PermissionGate { // run_shell, colliding reused ids, and tests). executionVerdict: (call: ToolCall) => Promise; // Resolve a suspended reactor approval against the operator (and mint the - // outcome's grant). Returns undefined when no outcome arrived. - resolveSuspended: (request: PermissionRequest) => Promise; + // outcome's grant when the session identity is still current). Returns + // undefined when no outcome arrived. + resolveSuspended: ( + request: PermissionRequest, + stillCurrent?: () => boolean, + ) => Promise; // True when this gate's decisions go through env.authorize (authorizeCall) // rather than evaluate() in the tool-runner middleware. Under reactor gating, // gateToolCall is an execution backstop: it consumes a matching cached @@ -724,7 +730,10 @@ export function createPermissionGate(options: PermissionGateOptions): Permission // span, await the requestApproval seam, settle the log/span, and mint any // grant the outcome carries (never for secret-path shell). Returns undefined // when no outcome arrived (timeout/abort auto-deny paths). - const resolveInteractiveAsk = async (decision: Extract) => { + const resolveInteractiveAsk = async ( + decision: Extract, + stillCurrent?: () => boolean, + ) => { const { request, anySecret, segmentCount } = decision; const askRule = anySecret ? "sensitive-path" : undefined; const ask = approvalLog.ask({ @@ -752,7 +761,7 @@ export function createPermissionGate(options: PermissionGateOptions): Permission finishApprovalWait(telemetry, waitSpanId, request.tool, outcome); ask.settle(classifyOutcome(outcome)); } - if (outcome !== undefined && outcome.allow && !anySecret) { + if (outcome !== undefined && outcome.allow && !anySecret && (stillCurrent?.() ?? true)) { mintGrant(request.tool, outcome); } return outcome; @@ -823,18 +832,21 @@ export function createPermissionGate(options: PermissionGateOptions): Permission // Resolve a suspended reactor approval once the operator answers. The // request is the one authorizeCall built at decision time, so the ask log, // wait span, and grant minting are identical to the middleware path. - const resolveSuspended = (request: PermissionRequest) => { + const resolveSuspended = (request: PermissionRequest, stillCurrent?: () => boolean) => { const anySecret = request.tool === "run_shell" && commandReferencesSensitivePath(request.subject) !== undefined; - return resolveInteractiveAsk({ - kind: "ask", - request, - anySecret, - segmentCount: - request.tool === "run_shell" - ? splitChainedCommand(request.subject).filter((s) => !isShellCommentOnly(s)).length - : 0, - }); + return resolveInteractiveAsk( + { + kind: "ask", + request, + anySecret, + segmentCount: + request.tool === "run_shell" + ? splitChainedCommand(request.subject).filter((s) => !isShellCommentOnly(s)).length + : 0, + }, + stillCurrent, + ); }; const reset = (): void => { diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index b1b8a4708..88ef1a4ac 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -26,6 +26,8 @@ import type { PermissionRequest } from "../permission/types.js"; const logger = getLogger([LOG_NAMESPACE_ROOT, "approval-resume"]); +export const APPROVAL_DROPPED_NOTICE = "Approval dropped because the session changed."; + const ApprovalSnapshotShape = type({ name: "string", "arguments?": "Record", @@ -108,16 +110,19 @@ function decisionMessage( } export function createApprovalResume(args: { - // Late-bound: the live agent is read at history/deliver time so rebuilds - // (/clear, model switch) deliver through the current instance, not a - // snapshot taken at handle() start. + // Live agent at history/deliver time. TUI occupancy holds this identity + // until the correlated resume is accepted; a generation bump aborts the + // gate rather than retargeting a rebuilt agent. getAgent: () => Pick | undefined; // TUI session queue. When present, each decision is awaited through this // seam; exec omits it and uses getAgent().deliver. deliver?: (message: InboundMessage, stillCurrent: () => boolean) => void | Promise; - // TUI: capture at handle() start so /clear or interrupt during the overlay - // drops the decision instead of delivering into the rebuilt agent. Exec omits this. + // TUI: capture at handle() start so interrupt, /clear, or /new during the + // overlay aborts the gate and drops the decision. Exec omits this. captureGeneration?: () => () => boolean; + // TUI: operator-visible notice when an overlay decision is dropped after + // a generation bump. + onDropped?: (text: string) => void; gate: PermissionGate; }): ApprovalResume { const { getAgent, gate } = args; @@ -149,6 +154,7 @@ export function createApprovalResume(args: { // out" tool result appended after this point means the reactor settled // this very correlation before our decision lands. const turnsAtSuspend = (await requireAgent().history()).length; + if (!stillCurrent()) return true; if (approvalSnapshot === undefined) { // A suspension without a snapshot cannot be surfaced; fail closed by @@ -168,8 +174,11 @@ export function createApprovalResume(args: { return true; } - const outcome = await gate.resolveSuspended(request); - if (!stillCurrent()) return true; + const outcome = await gate.resolveSuspended(request, stillCurrent); + if (!stillCurrent()) { + args.onDropped?.(APPROVAL_DROPPED_NOTICE); + return true; + } if (settledAfterSuspend(await requireAgent().history(), turnsAtSuspend)) { // The reactor already answered the parked call (its approval timeout // fired while the surface was still up). Delivering now would append diff --git a/src/tui/correlation-acceptance.test.ts b/src/tui/correlation-acceptance.test.ts new file mode 100644 index 000000000..96e572ee2 --- /dev/null +++ b/src/tui/correlation-acceptance.test.ts @@ -0,0 +1,25 @@ +import { describe, test } from "bun:test"; + +import { createCorrelationAcceptance } from "./correlation-acceptance.js"; + +describe("createCorrelationAcceptance", () => { + test("settle resolves the waiter for that correlation id", async () => { + const acceptance = createCorrelationAcceptance(); + const pending = acceptance.wait("corr-1"); + acceptance.settle("corr-1"); + await pending; + }); + + test("settleAll releases every outstanding waiter", async () => { + const acceptance = createCorrelationAcceptance(); + const first = acceptance.wait("a"); + const second = acceptance.wait("b"); + acceptance.settleAll(); + await Promise.all([first, second]); + }); + + test("settle of an unknown id is a no-op", () => { + const acceptance = createCorrelationAcceptance(); + acceptance.settle("missing"); + }); +}); diff --git a/src/tui/correlation-acceptance.ts b/src/tui/correlation-acceptance.ts new file mode 100644 index 000000000..1ce8c5053 --- /dev/null +++ b/src/tui/correlation-acceptance.ts @@ -0,0 +1,33 @@ +/** + * Occupancy wait for a fire-and-forget Agent.deliver of a correlated + * approval. The reactor accepts the resume asynchronously after deliver + * returns; inFlight must not drop until that acceptance (or an uncorrelated + * pass-through / identity bump) settles the waiter. + */ + +export function createCorrelationAcceptance() { + const waiters = new Map void>(); + + const settle = (correlationId: string): void => { + const resolve = waiters.get(correlationId); + if (resolve === undefined) return; + waiters.delete(correlationId); + resolve(); + }; + + return { + wait(correlationId: string): Promise { + const pending = waiters.get(correlationId); + return new Promise((resolve) => { + waiters.set(correlationId, () => { + pending?.(); + resolve(); + }); + }); + }, + settle, + settleAll(): void { + for (const correlationId of [...waiters.keys()]) settle(correlationId); + }, + }; +} diff --git a/src/tui/gate-wire.test.ts b/src/tui/gate-wire.test.ts index b8691e511..8c35fc218 100644 --- a/src/tui/gate-wire.test.ts +++ b/src/tui/gate-wire.test.ts @@ -19,6 +19,7 @@ import { import { moveOverlaySelection, toggleOverlayExpand } from "./shell/overlay-list.js"; import { streamRowGutter } from "./stream.js"; import { APPROVAL_UNAVAILABLE_MESSAGE } from "./gate-events.js"; +import { SESSION_IDENTITY_ABORT_REASON } from "./queued-delivery.js"; import { approvalOutcomeFromSelection, operatorCancelResult, @@ -1204,6 +1205,40 @@ describe("permission.gate auto-deny", () => { }); }); + test("identity abort reason auto-denies and closes the overlay", async () => { + await withTestRenderer(async (h) => { + const shell = createAppShell(h.renderer, { + terminal: { columns: 80, rows: 24 }, + run: "idle", + }); + const emitter = new EventEmitter(); + const controller = new AbortController(); + let resolved: unknown; + try { + wireGates(emitter, shell); + emitter.emit("permission.gate", { + id: "req-1", + request: baseRequest(), + resolve: (outcome: unknown) => { + resolved = outcome; + }, + signal: controller.signal, + }); + expect(shell.overlayKind).toBe("permissions"); + + controller.abort(SESSION_IDENTITY_ABORT_REASON); + + expect(resolved).toEqual({ + allow: false, + message: SESSION_IDENTITY_ABORT_REASON, + }); + expect(shell.overlayList).toBeNull(); + } finally { + shell.dispose(); + } + }); + }); + test("resolving normally clears the timer instead of firing it later", async () => { await withTestRenderer(async (h) => { const shell = createAppShell(h.renderer, { diff --git a/src/tui/gate-wire.ts b/src/tui/gate-wire.ts index 631488273..169a11ef7 100644 --- a/src/tui/gate-wire.ts +++ b/src/tui/gate-wire.ts @@ -412,10 +412,15 @@ export function wireGates( settle({ allow: false, message }); }; function onAbort(): void { - autoDeny("tool no longer running; permission request denied"); + const reason = ev.signal?.reason; + autoDeny( + typeof reason === "string" && reason.length > 0 + ? reason + : "tool no longer running; permission request denied", + ); } if (ev.signal?.aborted === true) { - autoDeny("tool no longer running; permission request denied"); + onAbort(); return; } ev.signal?.addEventListener("abort", onAbort, { once: true }); diff --git a/src/tui/queued-delivery.test.ts b/src/tui/queued-delivery.test.ts index c5d54c7e4..78c7155a0 100644 --- a/src/tui/queued-delivery.test.ts +++ b/src/tui/queued-delivery.test.ts @@ -5,6 +5,7 @@ import { createLeftoverSend, createLiveSteerDeliver, routeQueuedDelivery, + SESSION_IDENTITY_ABORT_REASON, } from "./queued-delivery.js"; import { createSessionOperationQueue } from "./session-operation-queue.js"; @@ -111,6 +112,16 @@ describe("createDeliveryGeneration", () => { expect(first()).toBe(false); expect(second()).toBe(false); }); + + test("bump aborts the prior identity signal and mints a fresh one", () => { + const generation = createDeliveryGeneration(); + const prior = generation.signal(); + expect(prior.aborted).toBe(false); + generation.bump(); + expect(prior.aborted).toBe(true); + expect(prior.reason).toBe(SESSION_IDENTITY_ABORT_REASON); + expect(generation.signal().aborted).toBe(false); + }); }); describe("createLiveSteerDeliver", () => { diff --git a/src/tui/queued-delivery.ts b/src/tui/queued-delivery.ts index 88a12854d..abaa482ba 100644 --- a/src/tui/queued-delivery.ts +++ b/src/tui/queued-delivery.ts @@ -33,16 +33,26 @@ export function routeQueuedDelivery(args: RouteQueuedDeliveryArgs): ProductHostD }; } -export function createDeliveryGeneration() { +export const SESSION_IDENTITY_ABORT_REASON = "session identity changed; approval request denied"; + +export function createDeliveryGeneration(onBump?: () => void) { let generation = 0; + let identity = new AbortController(); return { bump(): void { generation += 1; + const previous = identity; + identity = new AbortController(); + previous.abort(SESSION_IDENTITY_ABORT_REASON); + onBump?.(); }, capture(): () => boolean { const captured = generation; return () => captured === generation; }, + signal(): AbortSignal { + return identity.signal; + }, }; } diff --git a/src/tui/request-approval.test.ts b/src/tui/request-approval.test.ts index 8185a475f..8258bbaf6 100644 --- a/src/tui/request-approval.test.ts +++ b/src/tui/request-approval.test.ts @@ -117,6 +117,30 @@ describe("createGateRequestApproval", () => { expect(ids[0]?.length).toBeGreaterThan(0); expect(ids[1]).not.toBe(ids[0]); }); + + test("merges identitySignal so a generation bump aborts the overlay signal", async () => { + const identity = new AbortController(); + let captured: PermissionGateEvent | undefined; + const requestApproval = createGateRequestApproval({ + emitGate: (event) => { + captured = event; + return true; + }, + approvalTimeout: noTimeout, + identitySignal: () => identity.signal, + }); + const pending = requestApproval(request); + expect(captured?.signal).toBeDefined(); + expect(captured?.signal?.aborted).toBe(false); + identity.abort("session identity changed; approval request denied"); + expect(captured?.signal?.aborted).toBe(true); + expect(captured?.signal?.reason).toBe("session identity changed; approval request denied"); + captured?.resolve({ + allow: false, + message: "session identity changed; approval request denied", + }); + expect((await pending).allow).toBe(false); + }); }); // attachApprovalBudget is the mechanism createGateRequestApproval builds on diff --git a/src/tui/request-approval.ts b/src/tui/request-approval.ts index 6f8a31225..59de74724 100644 --- a/src/tui/request-approval.ts +++ b/src/tui/request-approval.ts @@ -15,6 +15,12 @@ export interface CreateGateRequestApprovalArgs { * a future generalized auto-continue mechanism owns re-arming this. */ approvalTimeout: () => { timeoutMs: number; timeoutMessage: string } | undefined; + /** + * Session-identity abort. A generation bump (interrupt, /clear, /new) + * aborts this signal so the outstanding overlay denies through the existing + * gate abort path instead of remaining as a ghost accept. + */ + identitySignal?: () => AbortSignal; } const logger = getLogger([LOG_NAMESPACE_ROOT, "tui", "permission"]); @@ -63,6 +69,12 @@ export function attachApprovalBudget( * Always attaches the budget signal so a timeout with waitForApproval off * dismisses the modal instead of leaving a ghost. */ +function mergeAbortSignals(a?: AbortSignal, b?: AbortSignal): AbortSignal | undefined { + if (a === undefined) return b; + if (b === undefined) return a; + return AbortSignal.any([a, b]); +} + export function createGateRequestApproval(args: CreateGateRequestApprovalArgs): RequestApproval { return (request: PermissionRequest) => new Promise((resolve) => { @@ -71,12 +83,13 @@ export function createGateRequestApproval(args: CreateGateRequestApprovalArgs): kind: "permission", }); const timeout = args.approvalTimeout(); + const merged = mergeAbortSignals(signal, args.identitySignal?.()); const event: PermissionGateEvent = { id: randomUUID(), request, resolve: finish, ...(timeout !== undefined ? timeout : {}), - ...(signal !== undefined ? { signal } : {}), + ...(merged !== undefined ? { signal: merged } : {}), }; if (!args.emitGate(event)) { // Pre-mount or post-unmount: no gate queue exists, so the prompt would diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index 054e7aa6c..e717595a3 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -122,8 +122,8 @@ export function agentRebuildFailure(err: unknown): Error { /** * Hard-stop interrupt: bump delivery generation, then enqueue the agent rebuild. - * Overlay stays open across interrupt; the bump must happen before enqueue so a - * later accept/decline cannot late-bind into the rebuilt agent. + * The bump aborts the outstanding permission gate (overlay dismissed, no grant) + * before enqueue so a later accept cannot mint into the rebuilt identity. */ export function startInterruptRebuild(args: { deliveryGeneration: { bump: () => void }; @@ -219,6 +219,10 @@ export async function createRunLifecycle( let eventForSink = event; if (event.type === "message.received") { providerFailureAttempts.advanceToNextMessage(); + const correlationId = event.data.message.headers.interchangeCorrelationId; + if (correlationId !== undefined) services.correlationAcceptance.settle(correlationId); + } else if (event.type === "message.correlated") { + services.correlationAcceptance.settle(event.data.correlationId); } else if (event.type === "inference.start" || event.type === "inference.done") { providerFailureAttempts.reset(); } else if (event.type === "inference.error") { diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index a9baef713..d5001f2da 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -60,6 +60,7 @@ import { createModelSummarizer, type SummaryContext } from "../../session/summar import { createSessionCostAccumulator } from "../../cost/session-cost.js"; import { createSessionOperationQueue } from "../session-operation-queue.js"; import { createDeliveryGeneration } from "../queued-delivery.js"; +import { createCorrelationAcceptance } from "../correlation-acceptance.js"; import { createAgentToolset, type MCPServerState, type OperatorResult } from "../../agent/tools.js"; import type { ToolAvailability } from "../../agent/tool-search.js"; import { detectLanguageServerAvailable } from "../../agent/lsp-availability.js"; @@ -124,6 +125,9 @@ export async function assembleTUISession( const approvalTimeout = (): { timeoutMs: number; timeoutMessage: string } | undefined => undefined; + const correlationAcceptance = createCorrelationAcceptance(); + const deliveryGeneration = createDeliveryGeneration(() => correlationAcceptance.settleAll()); + const { gate: permissionGate } = await assembleSessionGate({ cwd: config.cwd, sessionId: state.sessionId, @@ -133,6 +137,7 @@ export async function assembleTUISession( requestApproval: createGateRequestApproval({ emitGate: (event) => emitter.emit("permission.gate", event), approvalTimeout, + identitySignal: () => deliveryGeneration.signal(), }), getActiveProviderModel: () => `${state.config.providerName}:${state.config.model}`, onPersistNotice: (text) => state.approvalPersistNotice.notify?.(text), @@ -372,15 +377,24 @@ export async function assembleTUISession( // Reload, interrupt, compaction continuation, and proxy deliver share one queue // so a rebuild never races an in-flight deliver. const sessionOps = createSessionOperationQueue(); - const deliveryGeneration = createDeliveryGeneration(); const approvalResume = createApprovalResume({ getAgent: () => state.agentProxy ?? state.currentAgent, captureGeneration: deliveryGeneration.capture, + onDropped: (text) => state.systemNotice?.(text), deliver: (message, stillCurrent) => { return sessionOps.enqueue(async () => { if (!stillCurrent()) return; if (state.fatalBuildError !== null) throw state.fatalBuildError; - liveAgent(state).deliver(message); + const correlationId = message.headers.interchangeCorrelationId; + const accepted = + correlationId === undefined ? undefined : correlationAcceptance.wait(correlationId); + try { + liveAgent(state).deliver(message); + await accepted; + } catch (err) { + if (correlationId !== undefined) correlationAcceptance.settle(correlationId); + throw err; + } }); }, gate: permissionGate, @@ -522,6 +536,7 @@ export async function assembleTUISession( sessionCost, sessionOps, deliveryGeneration, + correlationAcceptance, buildSessionSources, providerFailureAttempts, baseToolCount, diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index d393c7a5c..662d62d7b 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -123,6 +123,9 @@ export interface RunnerServices { typeof import("../session-operation-queue.js").createSessionOperationQueue >; deliveryGeneration: ReturnType; + correlationAcceptance: ReturnType< + typeof import("../correlation-acceptance.js").createCorrelationAcceptance + >; buildSessionSources: () => import("../../session/assemble-runtime.js").LiveSessionSources; providerFailureAttempts: ReturnType< typeof import("../provider/failure-attempt.js").createProviderFailureAttemptTracker diff --git a/tests/unit/approval-resume.test.ts b/tests/unit/approval-resume.test.ts index d5b0f36cd..56c30d3f3 100644 --- a/tests/unit/approval-resume.test.ts +++ b/tests/unit/approval-resume.test.ts @@ -3,9 +3,21 @@ import { describe, expect, test } from "bun:test"; import { AgentClosedError, type SendResult } from "@intx/agent"; import type { ConversationTurn, InboundMessage } from "@intx/types/runtime"; -import type { PermissionGate } from "../../src/permission/gate.js"; -import { createApprovalResume } from "../../src/session/approval-resume.js"; -import { createDeliveryGeneration } from "../../src/tui/queued-delivery.js"; +import { createPermissionGate, type PermissionGate } from "../../src/permission/gate.js"; +import type { Approval, ApprovalScope, PermissionRequest } from "../../src/permission/types.js"; +import { + APPROVAL_DROPPED_NOTICE, + createApprovalResume, +} from "../../src/session/approval-resume.js"; +import { createCorrelationAcceptance } from "../../src/tui/correlation-acceptance.js"; +import type { PermissionGateEvent } from "../../src/tui/gate-events.js"; +import { + createDeliveryGeneration, + SESSION_IDENTITY_ABORT_REASON, +} from "../../src/tui/queued-delivery.js"; +import { createGateRequestApproval } from "../../src/tui/request-approval.js"; +import { runWhileAgentBusy } from "../../src/tui/runner/state.js"; +import { createSessionOperationQueue } from "../../src/tui/session-operation-queue.js"; const SUSPENDED: SendResult = { type: "suspended", @@ -87,34 +99,7 @@ describe("approval resume late-decision guard", () => { }); }); -describe("approval resume late-bind", () => { - test("rebuild-during-wait delivers to the agent present after resolveSuspended", async () => { - const deliveredA: unknown[] = []; - const deliveredB: unknown[] = []; - const agentA = { - deliver: (message: unknown) => deliveredA.push(message), - history: async () => [userTurn()], - }; - const agentB = { - deliver: (message: unknown) => deliveredB.push(message), - history: async () => [userTurn()], - }; - let current: typeof agentA | typeof agentB = agentA; - const gate = { - resolveSuspended: async () => { - current = agentB; - return { allow: true }; - }, - } as unknown as PermissionGate; - const resume = createApprovalResume({ getAgent: () => current, gate }); - - expect(await resume.handle(SUSPENDED)).toBe(true); - expect(deliveredA).toEqual([]); - expect(deliveredB).toHaveLength(1); - expect(correlationHeaders(deliveredB[0]).interchangeCorrelationId).toBe("corr-1"); - expect(correlationHeaders(deliveredB[0]).messageId).toBe("approval-corr-1"); - }); - +describe("approval resume delivery", () => { test("optional deliver is awaited and used instead of getAgent().deliver", async () => { const agentDelivered: unknown[] = []; const customDelivered: unknown[] = []; @@ -279,3 +264,266 @@ describe("approval resume generation capture", () => { await interruptDuringOverlayThenDecide({ allow: false, message: "not today" }); }); }); + +const persistAllow: ApprovalScope = { + id: "project-curl", + label: "Allow curl *", + pattern: "curl *", + grant: "project", +}; + +function persistRequest(): PermissionRequest { + return { + tool: "run_shell", + action: "Run shell command", + subject: "curl -sS https://example.com", + scopes: [persistAllow], + }; +} + +describe("approval resume identity abort", () => { + test("generation bump aborts the merged overlay signal with the identity reason", async () => { + const generation = createDeliveryGeneration(); + let captured: PermissionGateEvent | undefined; + const requestApproval = createGateRequestApproval({ + emitGate: (event) => { + captured = event; + return true; + }, + approvalTimeout: () => undefined, + identitySignal: () => generation.signal(), + }); + const pending = requestApproval(persistRequest()); + expect(captured?.signal?.aborted).toBe(false); + generation.bump(); + expect(captured?.signal?.aborted).toBe(true); + expect(captured?.signal?.reason).toBe(SESSION_IDENTITY_ABORT_REASON); + captured?.resolve({ allow: false, message: SESSION_IDENTITY_ABORT_REASON }); + await pending; + }); +}); + +describe("approval resume persist Allow after interrupt", () => { + test("drops persist Allow: no grant, overlay dismissed, operator notice", async () => { + const generation = createDeliveryGeneration(); + const persisted: Approval[] = []; + let overlay: PermissionGateEvent | undefined; + let overlayReady: (() => void) | undefined; + const waitForOverlay = new Promise((resolve) => { + overlayReady = resolve; + }); + const requestApproval = createGateRequestApproval({ + emitGate: (event) => { + overlay = event; + overlayReady?.(); + event.signal?.addEventListener( + "abort", + () => { + overlay = undefined; + event.resolve({ + allow: false, + message: + typeof event.signal?.reason === "string" + ? event.signal.reason + : SESSION_IDENTITY_ABORT_REASON, + }); + }, + { once: true }, + ); + return true; + }, + approvalTimeout: () => undefined, + identitySignal: () => generation.signal(), + }); + const gate = createPermissionGate({ + approvals: [], + interactive: true, + skipPermissions: false, + reactorGated: true, + requestApproval, + persist: (approval) => { + persisted.push(approval); + }, + }); + const delivered: unknown[] = []; + const notices: string[] = []; + const resume = createApprovalResume({ + getAgent: () => ({ + deliver: (message: unknown) => delivered.push(message), + history: async () => [userTurn()], + }), + captureGeneration: generation.capture, + onDropped: (text) => notices.push(text), + gate, + }); + + const pending = resume.handle(SUSPENDED); + await waitForOverlay; + expect(overlay).toBeDefined(); + + const dismissed = overlay; + generation.bump(); + dismissed?.resolve({ allow: true, persist: persistAllow }); + + expect(await pending).toBe(true); + expect(overlay).toBeUndefined(); + expect(persisted).toEqual([]); + expect(gate.getApprovals()).toEqual([]); + expect(notices).toEqual([APPROVAL_DROPPED_NOTICE]); + expect(delivered).toEqual([]); + }); +}); + +describe("approval resume stillCurrent at resolve", () => { + test("handle forwards the capture-at-start stillCurrent into resolveSuspended", async () => { + const generation = createDeliveryGeneration(); + let atResolve: boolean | undefined; + const resume = createApprovalResume({ + getAgent: () => ({ + deliver: () => undefined, + history: async () => [userTurn()], + }), + captureGeneration: generation.capture, + gate: { + resolveSuspended: async (_request: PermissionRequest, stillCurrent?: () => boolean) => { + expect(stillCurrent?.()).toBe(true); + generation.bump(); + atResolve = stillCurrent?.(); + return { allow: true, persist: persistAllow }; + }, + } as unknown as PermissionGate, + }); + + expect(await resume.handle(SUSPENDED)).toBe(true); + expect(atResolve).toBe(false); + }); + + test("resolveSuspended skips mintGrant when stillCurrent is false after persist Allow", async () => { + const generation = createDeliveryGeneration(); + const stillCurrent = generation.capture(); + const persisted: Approval[] = []; + const gate = createPermissionGate({ + approvals: [], + interactive: true, + skipPermissions: false, + reactorGated: true, + persist: (approval) => { + persisted.push(approval); + }, + requestApproval: async () => { + generation.bump(); + return { allow: true, persist: persistAllow }; + }, + }); + + const outcome = await gate.resolveSuspended(persistRequest(), stillCurrent); + expect(outcome?.allow).toBe(true); + expect(stillCurrent()).toBe(false); + expect(persisted).toEqual([]); + expect(gate.getApprovals()).toEqual([]); + }); +}); + +describe("approval resume occupancy until correlation", () => { + test("inFlight holds idle rebuild until the correlated resume is accepted", async () => { + const events: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + const correlationAcceptance = createCorrelationAcceptance(); + let delivered: (() => void) | undefined; + const waitUntilDelivered = new Promise((resolve) => { + delivered = resolve; + }); + const state = { + inFlight: 0, + pendingReload: false, + reloadIfIdle: () => { + if (!state.pendingReload || state.inFlight > 0) return; + state.pendingReload = false; + void enqueue(async () => { + events.push("rebuild"); + }); + }, + }; + const agent = { + deliver: (_message: unknown) => { + events.push("deliver"); + }, + history: async () => [userTurn()], + }; + const resume = createApprovalResume({ + getAgent: () => agent, + deliver: (message) => + enqueue(async () => { + const correlationId = message.headers.interchangeCorrelationId; + const accepted = + correlationId === undefined ? undefined : correlationAcceptance.wait(correlationId); + agent.deliver(message); + delivered?.(); + await accepted; + }), + gate: { + resolveSuspended: async () => { + state.pendingReload = true; + state.reloadIfIdle?.(); + return { allow: true }; + }, + } as unknown as PermissionGate, + }); + + const running = runWhileAgentBusy(state, async () => { + await resume.handle(SUSPENDED); + }); + + await waitUntilDelivered; + expect(events).toEqual(["deliver"]); + expect(state.inFlight).toBe(1); + expect(state.pendingReload).toBe(true); + + correlationAcceptance.settle("corr-1"); + await running; + await awaitTail(); + + expect(events).toEqual(["deliver", "rebuild"]); + expect(state.inFlight).toBe(0); + }); + + test("generation bump settleAll releases occupancy without a correlation event", async () => { + const correlationAcceptance = createCorrelationAcceptance(); + const generation = createDeliveryGeneration(() => correlationAcceptance.settleAll()); + let waiting: (() => void) | undefined; + const waitUntilWaiting = new Promise((resolve) => { + waiting = resolve; + }); + const state = { + inFlight: 0, + pendingReload: false, + reloadIfIdle: () => undefined, + }; + const resume = createApprovalResume({ + getAgent: () => ({ + deliver: () => undefined, + history: async () => [userTurn()], + }), + captureGeneration: generation.capture, + deliver: (message) => { + const correlationId = message.headers.interchangeCorrelationId; + if (correlationId === undefined) return; + const accepted = correlationAcceptance.wait(correlationId); + waiting?.(); + return accepted; + }, + gate: { + resolveSuspended: async () => ({ allow: true }), + } as unknown as PermissionGate, + }); + + const running = runWhileAgentBusy(state, async () => { + await resume.handle(SUSPENDED); + }); + await waitUntilWaiting; + expect(state.inFlight).toBe(1); + generation.bump(); + await running; + expect(state.inFlight).toBe(0); + }); +}); From 94211013b804a63850554c2e03fb707ec389aef7 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Wed, 9 Sep 2026 19:38:59 -0700 Subject: [PATCH 7/7] Reject parked approval on interrupt and hold occupancy until tool start --- src/session/approval-resume.ts | 117 +++++++++++------- src/tui/correlation-acceptance.test.ts | 44 ++++++- src/tui/correlation-acceptance.ts | 75 +++++++++++- src/tui/runner/exit.ts | 8 +- src/tui/runner/session.ts | 11 +- tests/unit/approval-resume.test.ts | 163 ++++++++++++++++++++++++- 6 files changed, 360 insertions(+), 58 deletions(-) diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index 88ef1a4ac..c557e631f 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -123,6 +123,9 @@ export function createApprovalResume(args: { // TUI: operator-visible notice when an overlay decision is dropped after // a generation bump. onDropped?: (text: string) => void; + // TUI: interrupt/clear bump this to reject the parked call on the old + // agent before close/rebuild. Cleared when handle returns. + registerParkedCancel?: (cancel: (() => void) | undefined) => void; gate: PermissionGate; }): ApprovalResume { const { getAgent, gate } = args; @@ -139,59 +142,83 @@ export function createApprovalResume(args: { handle: async (result) => { if (result.type !== "suspended") return false; const stillCurrent = args.captureGeneration?.() ?? (() => true); + const parkedAgent = requireAgent(); const { correlationId, approvalSnapshot } = result; - const deliverDecision = async (message: InboundMessage): Promise => { - if (!stillCurrent()) return; - if (args.deliver !== undefined) { - await args.deliver(message, stillCurrent); - return; - } - requireAgent().deliver(message); + let cancelled = false; + const cancelParked = (): void => { + if (cancelled) return; + cancelled = true; + parkedAgent.deliver(decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE)); }; + args.registerParkedCancel?.(cancelParked); - // Turn-count watermark for the settled guard below: a "approval timed - // out" tool result appended after this point means the reactor settled - // this very correlation before our decision lands. - const turnsAtSuspend = (await requireAgent().history()).length; - if (!stillCurrent()) return true; - - if (approvalSnapshot === undefined) { - // A suspension without a snapshot cannot be surfaced; fail closed by - // rejecting the parked call so the run does not hang on an invisible - // gate. - await deliverDecision( - decisionMessage(correlationId, "rejected", "approval surface unavailable"), - ); - return true; - } + const dropParked = (): void => { + args.onDropped?.(APPROVAL_DROPPED_NOTICE); + cancelParked(); + }; - const request = requestFromApprovalSnapshot(approvalSnapshot, correlationId); - if (request === null) { - await deliverDecision( - decisionMessage(correlationId, "rejected", "approval surface unavailable"), - ); - return true; - } + try { + const deliverDecision = async (message: InboundMessage): Promise => { + if (!stillCurrent()) return; + if (args.deliver !== undefined) { + await args.deliver(message, stillCurrent); + return; + } + requireAgent().deliver(message); + }; + + // Turn-count watermark for the settled guard below: a "approval timed + // out" tool result appended after this point means the reactor settled + // this very correlation before our decision lands. + const turnsAtSuspend = (await parkedAgent.history()).length; + if (!stillCurrent()) { + dropParked(); + return true; + } - const outcome = await gate.resolveSuspended(request, stillCurrent); - if (!stillCurrent()) { - args.onDropped?.(APPROVAL_DROPPED_NOTICE); - return true; - } - if (settledAfterSuspend(await requireAgent().history(), turnsAtSuspend)) { - // The reactor already answered the parked call (its approval timeout - // fired while the surface was still up). Delivering now would append - // the raw decision JSON as an uncorrelated user turn — drop and log. - logger.warn`late approval decision dropped correlation=${correlationId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`; - return true; - } - if (outcome === undefined || !outcome.allow) { - await deliverDecision(decisionMessage(correlationId, "rejected", outcome?.message)); + if (approvalSnapshot === undefined) { + // A suspension without a snapshot cannot be surfaced; fail closed by + // rejecting the parked call so the run does not hang on an invisible + // gate. + args.registerParkedCancel?.(undefined); + await deliverDecision( + decisionMessage(correlationId, "rejected", "approval surface unavailable"), + ); + return true; + } + + const request = requestFromApprovalSnapshot(approvalSnapshot, correlationId); + if (request === null) { + args.registerParkedCancel?.(undefined); + await deliverDecision( + decisionMessage(correlationId, "rejected", "approval surface unavailable"), + ); + return true; + } + + const outcome = await gate.resolveSuspended(request, stillCurrent); + if (!stillCurrent()) { + dropParked(); + return true; + } + args.registerParkedCancel?.(undefined); + if (settledAfterSuspend(await requireAgent().history(), turnsAtSuspend)) { + // The reactor already answered the parked call (its approval timeout + // fired while the surface was still up). Delivering now would append + // the raw decision JSON as an uncorrelated user turn — drop and log. + logger.warn`late approval decision dropped correlation=${correlationId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`; + return true; + } + if (outcome === undefined || !outcome.allow) { + await deliverDecision(decisionMessage(correlationId, "rejected", outcome?.message)); + return true; + } + await deliverDecision(decisionMessage(correlationId, "approved")); return true; + } finally { + args.registerParkedCancel?.(undefined); } - await deliverDecision(decisionMessage(correlationId, "approved")); - return true; }, }; } diff --git a/src/tui/correlation-acceptance.test.ts b/src/tui/correlation-acceptance.test.ts index 96e572ee2..c27f7d87c 100644 --- a/src/tui/correlation-acceptance.test.ts +++ b/src/tui/correlation-acceptance.test.ts @@ -1,7 +1,21 @@ -import { describe, test } from "bun:test"; +import { describe, expect, test } from "bun:test"; import { createCorrelationAcceptance } from "./correlation-acceptance.js"; +function approvedMessage(correlationId: string) { + return { + headers: { interchangeCorrelationId: correlationId }, + content: JSON.stringify({ outcome: "approved" }), + }; +} + +function rejectedMessage(correlationId: string) { + return { + headers: { interchangeCorrelationId: correlationId }, + content: JSON.stringify({ outcome: "rejected" }), + }; +} + describe("createCorrelationAcceptance", () => { test("settle resolves the waiter for that correlation id", async () => { const acceptance = createCorrelationAcceptance(); @@ -22,4 +36,32 @@ describe("createCorrelationAcceptance", () => { const acceptance = createCorrelationAcceptance(); acceptance.settle("missing"); }); + + test("an approved correlation does not settle until tool.start", async () => { + const acceptance = createCorrelationAcceptance(); + const pending = acceptance.wait("corr-1"); + let settled = false; + void pending.then(() => { + settled = true; + }); + acceptance.observe({ + type: "message.correlated", + data: { correlationId: "corr-1", message: approvedMessage("corr-1") }, + }); + await Promise.resolve(); + expect(settled).toBe(false); + acceptance.observe({ type: "tool.start", data: { call: { id: "call-1" } } }); + await pending; + expect(settled).toBe(true); + }); + + test("a rejected correlation settles at message.correlated", async () => { + const acceptance = createCorrelationAcceptance(); + const pending = acceptance.wait("corr-1"); + acceptance.observe({ + type: "message.correlated", + data: { correlationId: "corr-1", message: rejectedMessage("corr-1") }, + }); + await pending; + }); }); diff --git a/src/tui/correlation-acceptance.ts b/src/tui/correlation-acceptance.ts index 1ce8c5053..ccb525bc0 100644 --- a/src/tui/correlation-acceptance.ts +++ b/src/tui/correlation-acceptance.ts @@ -2,19 +2,72 @@ * Occupancy wait for a fire-and-forget Agent.deliver of a correlated * approval. The reactor accepts the resume asynchronously after deliver * returns; inFlight must not drop until that acceptance (or an uncorrelated - * pass-through / identity bump) settles the waiter. + * pass-through / identity bump) settles the waiter. An approved re-dispatch + * is not idle at message.correlated — occupancy holds until tool.start. */ +import { ApprovalDecision } from "@intx/types"; +import { type } from "arktype"; + +export interface CorrelationStreamEvent { + type: string; + data?: unknown; +} + +function isApprovedDecision(content: string | undefined): boolean { + if (content === undefined) return false; + let raw: unknown; + try { + raw = JSON.parse(content); + } catch { + return false; + } + const decision = ApprovalDecision(raw); + if (decision instanceof type.errors) return false; + return decision.outcome === "approved"; +} + +function correlatedMessage(data: unknown): { + correlationId: string | undefined; + content: string | undefined; + receivedCorrelationId: string | undefined; +} { + if (data === null || typeof data !== "object") { + return { correlationId: undefined, content: undefined, receivedCorrelationId: undefined }; + } + const record = data as { + correlationId?: unknown; + message?: { + headers?: { interchangeCorrelationId?: unknown }; + content?: unknown; + }; + }; + return { + correlationId: typeof record.correlationId === "string" ? record.correlationId : undefined, + content: typeof record.message?.content === "string" ? record.message.content : undefined, + receivedCorrelationId: + typeof record.message?.headers?.interchangeCorrelationId === "string" + ? record.message.headers.interchangeCorrelationId + : undefined, + }; +} + export function createCorrelationAcceptance() { const waiters = new Map void>(); + const holdUntilToolStart = new Set(); const settle = (correlationId: string): void => { + holdUntilToolStart.delete(correlationId); const resolve = waiters.get(correlationId); if (resolve === undefined) return; waiters.delete(correlationId); resolve(); }; + const settleHeldForToolStart = (): void => { + for (const correlationId of [...holdUntilToolStart]) settle(correlationId); + }; + return { wait(correlationId: string): Promise { const pending = waiters.get(correlationId); @@ -27,7 +80,27 @@ export function createCorrelationAcceptance() { }, settle, settleAll(): void { + holdUntilToolStart.clear(); for (const correlationId of [...waiters.keys()]) settle(correlationId); }, + observe(event: CorrelationStreamEvent): void { + if (event.type === "tool.start") { + settleHeldForToolStart(); + return; + } + const fields = correlatedMessage(event.data); + if (event.type === "message.received") { + if (fields.receivedCorrelationId !== undefined) settle(fields.receivedCorrelationId); + return; + } + if (event.type === "message.correlated") { + if (fields.correlationId === undefined) return; + if (isApprovedDecision(fields.content)) { + holdUntilToolStart.add(fields.correlationId); + return; + } + settle(fields.correlationId); + } + }, }; } diff --git a/src/tui/runner/exit.ts b/src/tui/runner/exit.ts index e717595a3..bdca53ad4 100644 --- a/src/tui/runner/exit.ts +++ b/src/tui/runner/exit.ts @@ -219,11 +219,9 @@ export async function createRunLifecycle( let eventForSink = event; if (event.type === "message.received") { providerFailureAttempts.advanceToNextMessage(); - const correlationId = event.data.message.headers.interchangeCorrelationId; - if (correlationId !== undefined) services.correlationAcceptance.settle(correlationId); - } else if (event.type === "message.correlated") { - services.correlationAcceptance.settle(event.data.correlationId); - } else if (event.type === "inference.start" || event.type === "inference.done") { + } + services.correlationAcceptance.observe(event); + if (event.type === "inference.start" || event.type === "inference.done") { providerFailureAttempts.reset(); } else if (event.type === "inference.error") { const error = event.data.error; diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index d5001f2da..05a19ac98 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -126,7 +126,11 @@ export async function assembleTUISession( undefined; const correlationAcceptance = createCorrelationAcceptance(); - const deliveryGeneration = createDeliveryGeneration(() => correlationAcceptance.settleAll()); + const parkedApprovalCancel = { fn: undefined as (() => void) | undefined }; + const deliveryGeneration = createDeliveryGeneration(() => { + parkedApprovalCancel.fn?.(); + correlationAcceptance.settleAll(); + }); const { gate: permissionGate } = await assembleSessionGate({ cwd: config.cwd, @@ -378,9 +382,12 @@ export async function assembleTUISession( // so a rebuild never races an in-flight deliver. const sessionOps = createSessionOperationQueue(); const approvalResume = createApprovalResume({ - getAgent: () => state.agentProxy ?? state.currentAgent, + getAgent: () => state.currentAgent, captureGeneration: deliveryGeneration.capture, onDropped: (text) => state.systemNotice?.(text), + registerParkedCancel: (cancel) => { + parkedApprovalCancel.fn = cancel; + }, deliver: (message, stillCurrent) => { return sessionOps.enqueue(async () => { if (!stillCurrent()) return; diff --git a/tests/unit/approval-resume.test.ts b/tests/unit/approval-resume.test.ts index 56c30d3f3..05886cb31 100644 --- a/tests/unit/approval-resume.test.ts +++ b/tests/unit/approval-resume.test.ts @@ -16,6 +16,7 @@ import { SESSION_IDENTITY_ABORT_REASON, } from "../../src/tui/queued-delivery.js"; import { createGateRequestApproval } from "../../src/tui/request-approval.js"; +import { startInterruptRebuild } from "../../src/tui/runner/exit.js"; import { runWhileAgentBusy } from "../../src/tui/runner/state.js"; import { createSessionOperationQueue } from "../../src/tui/session-operation-queue.js"; @@ -172,7 +173,7 @@ describe("approval resume delivery", () => { }); describe("approval resume generation capture", () => { - test("overlay accept after a generation bump does not deliver", async () => { + test("overlay accept after a generation bump rejects the parked call on the old agent", async () => { const generation = createDeliveryGeneration(); const delivered: unknown[] = []; const agent = { @@ -195,7 +196,11 @@ describe("approval resume generation capture", () => { }); expect(await resume.handle(SUSPENDED)).toBe(true); - expect(delivered).toEqual([]); + expect(delivered).toHaveLength(1); + expect(JSON.parse((delivered[0] as { content: string }).content)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); }); test("a live generation still delivers after the overlay", async () => { @@ -252,7 +257,11 @@ describe("approval resume generation capture", () => { }); expect(await resume.handle(SUSPENDED)).toBe(true); - expect(deliveredA).toEqual([]); + expect(deliveredA).toHaveLength(1); + expect(JSON.parse((deliveredA[0] as { content: string }).content)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); expect(deliveredB).toEqual([]); } @@ -263,6 +272,69 @@ describe("approval resume generation capture", () => { test("interrupt during overlay then decline does not deliver to the rebuilt agent", async () => { await interruptDuringOverlayThenDecide({ allow: false, message: "not today" }); }); + + test("interrupt during overlay rejects the parked call before rebuild enqueue", async () => { + const events: string[] = []; + const parkedCancel = { fn: undefined as (() => void) | undefined }; + const generation = createDeliveryGeneration(() => parkedCancel.fn?.()); + const { enqueue, awaitTail } = createSessionOperationQueue(); + const delivered: unknown[] = []; + const agent = { + deliver: (message: unknown) => { + events.push("reject"); + delivered.push(message); + }, + history: async () => [userTurn()], + }; + let overlayReady: (() => void) | undefined; + const waitForOverlay = new Promise((resolve) => { + overlayReady = resolve; + }); + let finishOverlay: ((outcome: { allow: boolean }) => void) | undefined; + const resume = createApprovalResume({ + getAgent: () => agent, + captureGeneration: generation.capture, + registerParkedCancel: (cancel) => { + parkedCancel.fn = cancel; + }, + onDropped: () => events.push("dropped"), + gate: { + resolveSuspended: () => { + overlayReady?.(); + return new Promise((resolve) => { + finishOverlay = resolve; + }); + }, + } as unknown as PermissionGate, + }); + + const handling = resume.handle(SUSPENDED); + await waitForOverlay; + startInterruptRebuild({ + deliveryGeneration: generation, + markSendAborted: () => { + events.push("abort"); + }, + enqueue: (op) => { + events.push("enqueue"); + return enqueue(op); + }, + rebuild: async () => { + events.push("rebuild"); + }, + }); + finishOverlay?.({ allow: false }); + + expect(await handling).toBe(true); + await awaitTail(); + expect(events.indexOf("reject")).toBeGreaterThanOrEqual(0); + expect(events.indexOf("reject")).toBeLessThan(events.indexOf("enqueue")); + expect(events).toContain("rebuild"); + expect(JSON.parse((delivered[0] as { content: string }).content)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); + }); }); const persistAllow: ApprovalScope = { @@ -370,7 +442,11 @@ describe("approval resume persist Allow after interrupt", () => { expect(persisted).toEqual([]); expect(gate.getApprovals()).toEqual([]); expect(notices).toEqual([APPROVAL_DROPPED_NOTICE]); - expect(delivered).toEqual([]); + expect(delivered).toHaveLength(1); + expect(JSON.parse((delivered[0] as { content: string }).content)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); }); }); @@ -487,6 +563,85 @@ describe("approval resume occupancy until correlation", () => { expect(state.inFlight).toBe(0); }); + test("approved correlation holds idle rebuild until tool.start", async () => { + const events: string[] = []; + const { enqueue, awaitTail } = createSessionOperationQueue(); + const correlationAcceptance = createCorrelationAcceptance(); + let delivered: (() => void) | undefined; + const waitUntilDelivered = new Promise((resolve) => { + delivered = resolve; + }); + const state = { + inFlight: 0, + pendingReload: false, + reloadIfIdle: () => { + if (!state.pendingReload || state.inFlight > 0) return; + state.pendingReload = false; + void enqueue(async () => { + events.push("rebuild"); + }); + }, + }; + const agent = { + deliver: (_message: unknown) => { + events.push("deliver"); + }, + history: async () => [userTurn()], + }; + const resume = createApprovalResume({ + getAgent: () => agent, + deliver: (message) => + enqueue(async () => { + const correlationId = message.headers.interchangeCorrelationId; + const accepted = + correlationId === undefined ? undefined : correlationAcceptance.wait(correlationId); + agent.deliver(message); + delivered?.(); + await accepted; + }), + gate: { + resolveSuspended: async () => { + state.pendingReload = true; + state.reloadIfIdle?.(); + return { allow: true }; + }, + } as unknown as PermissionGate, + }); + + const running = runWhileAgentBusy(state, async () => { + await resume.handle(SUSPENDED); + }); + + await waitUntilDelivered; + expect(events).toEqual(["deliver"]); + expect(state.inFlight).toBe(1); + expect(state.pendingReload).toBe(true); + + correlationAcceptance.observe({ + type: "message.correlated", + data: { + correlationId: "corr-1", + message: { + headers: { interchangeCorrelationId: "corr-1" }, + content: JSON.stringify({ outcome: "approved" }), + }, + }, + }); + await Promise.resolve(); + expect(events).toEqual(["deliver"]); + expect(state.inFlight).toBe(1); + + correlationAcceptance.observe({ + type: "tool.start", + data: { call: { id: "call-ask" } }, + }); + await running; + await awaitTail(); + + expect(events).toEqual(["deliver", "rebuild"]); + expect(state.inFlight).toBe(0); + }); + test("generation bump settleAll releases occupancy without a correlation event", async () => { const correlationAcceptance = createCorrelationAcceptance(); const generation = createDeliveryGeneration(() => correlationAcceptance.settleAll());