diff --git a/docs/ARCHITECTURE.md b/docs/ARCHITECTURE.md index 3e4fdcb7e..11006cdaf 100644 --- a/docs/ARCHITECTURE.md +++ b/docs/ARCHITECTURE.md @@ -14,6 +14,21 @@ The reactor (from `@intx/agent`) drives a single agent turn-by-turn. Each turn i This repeats until the director emits `capabilities.done()`. +Explicit approval messages use `headers.interchangeType` (`approval.granted` or +`approval.denied`) as intent, not authentication. The reactor accepts one only +while its exact pending operation and matching gate remain live, before both +epoch deadlines (equality expires), with a body agreeing with the header. Typed +messages without an eligible suspended call, expired/consumed decisions and +concurrent duplicates are discarded before body parsing, never ordinary input. +Validator false/throw also discards typed decisions but leaves a still-live +operation retryable; untyped and nonapproval response routing is unchanged. +After validation and synchronous grant callbacks, the reactor rechecks ownership, +deadlines and lifecycle, then consumes the operation and enqueues its continuation +before publishing `message.correlated`. The claim is released on every settled +exit; a never-settling validator retains it. Extension side effects are not rolled +back. Discards produce no acceptance receipt: the TUI's existing bounded 30-second +acceptance wait remains unchanged. + ### Reactor Events (Partial) This table is not the full reactor/stream event vocabulary. `tool.done` is @@ -393,6 +408,7 @@ tool call - **auto-shell-policy** — Constrains `run_shell` even when auto mode would otherwise rubber-stamp it. Before matching, `expandShellSubjects` peels `bash`/`sh`/`zsh -c`, `xargs` utility tails, and transparent prefixes (`env`, `nice`, `timeout`, …) so rules see the real payload; an unparseable wrapper (variable expansion or command substitution) sets an opaque flag that forces `ask`. Effects: `deny` blocks outright (file mutations through ad-hoc tooling — output redirection, `tee`, `sed -i`/`perl -i`, interpreter inline programs or heredocs — which must instead go through `write_file`/`edit_file`); `ask` declines to auto-allow and falls through to the operator prompt (recursive `rm`, dependency installs and remote runners: npm/yarn/pnpm/bun, pip, cargo, go, brew, npx/bunx, …, force or uncontained `git worktree` ops, shell that references a sensitive path such as `.env` or a private key, and opaque wrappers). Contained non-force `git worktree add`/`remove`/`prune` and read-only `list` auto-allow (sibling destinations like `../corbits-dispatch-wts/…` included; absolute outside, `~`, globs, and credential basenames still ask). Deny beats ask when multiple subjects match. Quoted spans are stripped before pattern matching so a quoted `>` or install word in an argument is not flagged, and program names are matched only in command position. Adding a table category is a one-line rule append in `AUTO_SHELL_RULES`. - **gate** — Evaluates a call: `skipPermissions` allows everything; `allow`-tier passes; for `ask`-tier, checks persisted approvals, otherwise requests operator approval. Shell security classifies each chain segment (`||` / `&&` / `|` / `;` / newlines), but the operator is prompted once for the full command block — any unapproved segment fails the whole block, and execution always runs the unsplit original. Safe pipeline tails and pure shell no-ops (`true` / `false` / `:` and bare control-flow keywords stranded by chain-splitting) skip without a prompt. In a non-interactive run an unresolved `ask` becomes a denial. In auto mode: non-shell built-ins in `AUTO_ALLOWED_TOOLS` (writes/edits/deletes, `manage_tasks`, `spawn_agent`, `wait_agents`, …) auto-allow when not path-restricted; for `run_shell` the gate consults the auto-shell policy — a `deny` rule fails the call, an `ask` rule skips the auto-allow shortcut and proceeds to the normal approval flow, and anything unmatched is auto-allowed. Paths outside the workspace on path-arg tools are denied at authorize time (the same sandbox path-escape enforces at execution, so the gate does not show an Accept overlay that cannot succeed). Writes under the in-workspace session state root (legacy `.agent-state`) still ask under auto mode. Under `--dangerously-skip-permissions` (forces this process) or `/yolo` (persists as the user-global default via `setSkipPermissions`), the gate auto-allows those same cases, and pre-gate sandboxes (path-escape, shell session cwd retention, `list_dir` / `delete_file` workspace bounds) honor `getSkipPermissions()` live so outside-workspace access is not hard-denied after the gate already allowed it — without rebuilding the plugin stack. Secret-guard path denies and authorization hard blocks still apply. Mutating MCP and unknown built-ins are not blanket-allowed outside skip. Newly granted scopes are appended in memory and persisted. - **Reactor-gated sessions (main session; `reactorGated: true`).** The gate's decision logic lives in one `decide()` used by both consumers: `evaluate()` (the middleware path below, still used by sub-agents) and `authorizeCall()`, which expresses the decision as the vendored reactor's before-tool authz effect (`src/permission/reactor-authorize.ts` bridges it into `env.authorize`). An `ask` there suspends the call as a reactor `PendingOperation` keyed by a correlationId (persisted through the context store's existing `pendingOperations`); `send()` settles as `suspended` and `src/session/approval-resume.ts` rebuilds the operator request from the approval snapshot, resolves it through the same `requestApproval` seam the TUI overlay uses, and delivers the decision to the reactor on the correlationId signal channel — an approved decision grants a one-shot bypass and the exact parked call re-dispatches; a rejected one answers it with an error result. `inFlight` occupancy owns idle rebuild: the TUI stays busy across the overlay and waits until the correlated resume is accepted (`message.received` / `message.correlated`) or a generation bump `settleAll`s the waiter. Delivery generation owns session identity: interrupt, `/clear`, and `/new` abort the outstanding overlay, skip minting a grant, drop the decision, and surface an operator notice rather than delivering into a rebuilt agent. Under reactor gating the middleware/MCP `gateToolCall` is an execution backstop, not a second copy of `env.authorize`: it consumes the `authorizeCall` verdict only when id, name, and arguments match, and does not re-decide. Deny still blocks and does not call `next`; an `ask` or `allow` skips the middleware prompt so an approved re-dispatch never re-asks. A reused `codex-proxy` id cannot apply an outer `shell` allow to an inner `run_shell` deny. Inner posix runs whose outer tool is not `run_shell` (Codex `apply_patch` proxy) never pass `env.authorize`, so `gateToolCall` decides on that cache miss and still blocks a deny. The headless denial and the stricter chained-command hard-deny are preserved as deny effects (upstream `block`s) decided inside the same `decide()`. + - **Approval resume identity.** Before opening the operator gate, resume captures the session generation and agent/store pair and resolves the correlation exactly once through `ContextStore.load().pendingOperations` to one approval operation's `suspendedCall.id`. Missing or duplicate mappings produce no gate or delivery; store errors propagate with registration cleanup. Every decision path first checks captured history for an exact-call approval timeout, and operator decisions check again after the gate. Identical tool names and arguments never establish identity. Cancellation during lookup cannot deliver a rejection. This suppresses observed exact timeouts, not all expired correlations: the reactor removes correlation state before the queued timeout result publishes, and expiration can also race the final history check or TUI delivery queue. Atomic stale-decision admission remains reactor-owned work tracked separately in CL-8000. - **Worker reactor ownership.** `workerPermissionGate` is a reactor-gated view over the parent's live permission gate: grants and policy are shared, not copied or toggled. Worker posix plugins and inherited MCP tools are bound to that view at worker start, so they take the reactor-gated `gateToolCall` path because the view reports `isReactorGated()` — they do not close over the parent's middleware-gated `isReactorGated()`. Deny still blocks; ask/allow skip the middleware prompt. `authorizeCall` on the view never emits `ask` — unresolved approvals become denials that name the permission subject, without invoking an approval callback or suspending, even with an interactive parent; the parent can obtain a grant and retry. Worker control-plane tools (`submit_result`, `ask_director`, and nested fleet verbs other than `spawn_agent`) allow without a parent grant. Authorization and tool execution run under the same async-local worker identity and cwd. Fleet authority remains an independent restriction, not an alternative permission grant. - **matcher** — Approval pattern matching via `@intx/authz` `matchPattern` (`*` wildcards). Exact-command grants store a backslash before each metacharacter; those patterns match by equality after unescape (the package has no escape syntax). diff --git a/src/exec/runner.ts b/src/exec/runner.ts index e347a8fc0..07e607eb5 100644 --- a/src/exec/runner.ts +++ b/src/exec/runner.ts @@ -110,7 +110,10 @@ import { loadSessionChatPrompt, skillDirsFromEnabledPlugins, } from "../session/runtime-assembly.js"; -import { createApprovalResume } from "../session/approval-resume.js"; +import { + createApprovalResume, + resolveParkedCallIdFromStore, +} from "../session/approval-resume.js"; import { createReactorAuthorize } from "../permission/reactor-authorize.js"; import { assembleChatAgent, @@ -932,6 +935,9 @@ export async function runExec(config: Config): Promise { agent = currentAgent; // Local non-null handle after init (currentAgent stays mutable for compaction deliver). const activeAgent = currentAgent; + const activeStorage = currentStorage; + if (activeStorage === null) + throw new Error("approval resume: no context store"); if (agentToolset.connectMCP !== undefined) { await agentToolset @@ -1051,10 +1057,10 @@ export async function runExec(config: Config): Promise { const sendResult = await activeAgent.send(operatorTaskMessage(task)); // A suspension must not park silently in exec: the approval resume owns // the terminal prompt flow and delivers the decision to the reactor. - // No resolveParkedCallId: the vendored reactor exposes no - // correlationId-to-call lookup, so the history heuristic is the path. await createApprovalResume({ getAgent: () => activeAgent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(activeStorage, correlationId), gate: permissionGate, }).handle(sendResult); sendCompleted = true; diff --git a/src/session/approval-resume.test.ts b/src/session/approval-resume.test.ts index 489483d24..558e9c442 100644 --- a/src/session/approval-resume.test.ts +++ b/src/session/approval-resume.test.ts @@ -1,14 +1,21 @@ -import { describe, expect, test } from "bun:test"; +import { describe, expect, mock, test } from "bun:test"; import type { Agent, SendResult } from "@intx/agent"; import type { ApprovalSnapshot, + ContextStore, + PendingOperation, ConversationTurn, InboundMessage, } from "@intx/types/runtime"; import { APPROVAL_TIMEOUT_RESULT_TEXT } from "../permission/decline-markers.js"; import type { PermissionGate } from "../permission/gate.js"; -import { createApprovalResume } from "./approval-resume.js"; +import { + APPROVAL_DROPPED_NOTICE, + createApprovalResume, + resolveParkedCallIdFromStore, +} from "./approval-resume.js"; +import { createSessionOperationQueue } from "../tui/delivery-queue.js"; function assistantTurn( calls: { id: string; name: string; command: string }[], @@ -77,17 +84,20 @@ function setup(args: { const resume = createApprovalResume({ getAgent: () => agent as Pick, gate, - ...(args.resolveParkedCallId !== undefined - ? { resolveParkedCallId: args.resolveParkedCallId } - : {}), + resolveParkedCallId: + args.resolveParkedCallId ?? + ((correlationId) => (correlationId === "corr-A" ? "call-A" : undefined)), }); return { resume, delivered }; } -function decisionBody(message: InboundMessage): { outcome: string } { +function decisionBody(message: InboundMessage): { + outcome: string; + message?: string; +} { if (message.content === undefined) throw new Error("expected a decision body"); - return JSON.parse(message.content) as { outcome: string }; + return JSON.parse(message.content) as { outcome: string; message?: string }; } function deliveredCorrelationId(message: InboundMessage): string { @@ -97,6 +107,59 @@ function deliveredCorrelationId(message: InboundMessage): string { return correlationId; } +describe("approval decision intent headers", () => { + for (const allow of [true, false]) { + test(`preserves ${allow ? "granted" : "denied"} intent and correlation through the session queue`, async () => { + const delivered: InboundMessage[] = []; + const queue = createSessionOperationQueue(); + let release!: () => void; + const held = new Promise((resolve) => { + release = resolve; + }); + queue.enqueue(() => held); + const agent = { + history: async () => [], + deliver: (message: InboundMessage) => { + delivered.push(message); + }, + }; + const gate = { + resolveSuspended: async () => ({ allow, message: "decision reason" }), + } as unknown as PermissionGate; + const resume = createApprovalResume({ + getAgent: () => agent, + gate, + resolveParkedCallId: () => "typed-call", + deliver: (message) => + queue.enqueue(async () => { + agent.deliver(message); + }), + }); + const handling = resume.handle( + suspension("typed-correlation", "echo alpha"), + ); + expect(delivered).toHaveLength(0); + release(); + expect(await handling).toBe(true); + await queue.awaitTail(); + expect(delivered).toHaveLength(1); + const message = delivered[0]; + if (message === undefined) throw new Error("expected decision"); + expect(message.headers.interchangeType).toBe( + allow ? "approval.granted" : "approval.denied", + ); + expect(deliveredCorrelationId(message)).toBe("typed-correlation"); + expect(decisionBody(message).outcome).toBe( + allow ? "approved" : "rejected", + ); + if (!allow) + expect(JSON.parse(message.content ?? "").message).toBe( + "decision reason", + ); + }); + } +}); + describe("approval-resume parallel-parked approvals", () => { test("delivers A's decision when a different parked call's approval times out", async () => { const { resume, delivered } = setup({ @@ -185,14 +248,13 @@ describe("approval-resume parallel-parked approvals", () => { expect(delivered).toHaveLength(1); }); - // Known limitation of the exact-one history derivation: with identical - // name+args twins, the answered own call drops out of the candidates and - // the derivation resolves to the unanswered twin, so the settled check - // misses and the genuinely-late decision is delivered. Telling the twins - // apart needs the resolveParkedCallId lookup; this test locks the current - // shape so a future fix can flip it to a drop. - test("identical name+args twin: own timeout with unanswered twin delivers", async () => { + test("identical name+args twin: own timeout with unanswered twin drops", async () => { + const pending = new Map([ + ["corr-A", "call-A"], + ["corr-B", "call-B"], + ]); const { resume, delivered } = setup({ + resolveParkedCallId: (correlationId) => pending.get(correlationId), preTurns: [ assistantTurn([ { id: "call-A", name: "run_shell", command: "echo same" }, @@ -200,6 +262,7 @@ describe("approval-resume parallel-parked approvals", () => { ]), ], onGate: (turns) => { + pending.delete("corr-A"); turns.push(timeoutTurn("call-A")); }, }); @@ -207,6 +270,436 @@ describe("approval-resume parallel-parked approvals", () => { const handled = await resume.handle(suspension("corr-A", "echo same")); expect(handled).toBe(true); + expect(delivered).toHaveLength(0); + expect(await resume.handle(suspension("corr-B", "echo same"))).toBe(true); + expect(delivered).toHaveLength(1); + const message = delivered[0]; + if (message === undefined) throw new Error("expected B decision"); + expect(deliveredCorrelationId(message)).toBe("corr-B"); + }); +}); + +function storeWith( + pendingOperations: PendingOperation[], +): Pick { + return { + load: async () => ({ + turns: [], + pendingOperations, + tokenUsage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + thinking: 0, + }, + connectorState: null, + }), + }; +} + +const operation: PendingOperation = { + correlationId: "corr-A", + kind: "approval", + registeredAt: 1, + gateId: "gate-A", + suspendedCall: { + id: "call-A", + name: "run_shell", + arguments: { command: "echo same" }, + }, +}; + +describe("persisted approval identity", () => { + test("requires exactly one matching approval with a suspended call", async () => { + expect( + await resolveParkedCallIdFromStore(storeWith([operation]), "corr-A"), + ).toBe("call-A"); + expect( + await resolveParkedCallIdFromStore(storeWith([operation]), "corr-B"), + ).toBeUndefined(); + expect( + await resolveParkedCallIdFromStore(storeWith([]), "corr-A"), + ).toBeUndefined(); + expect( + await resolveParkedCallIdFromStore( + storeWith([operation, operation]), + "corr-A", + ), + ).toBeUndefined(); + const { suspendedCall: _call, ...withoutCall } = operation; + expect( + await resolveParkedCallIdFromStore(storeWith([withoutCall]), "corr-A"), + ).toBeUndefined(); + expect( + await resolveParkedCallIdFromStore( + storeWith([operation, withoutCall]), + "corr-A", + ), + ).toBeUndefined(); + }); +}); + +test("missing or ambiguous stored identity never falls back to an identical history call", async () => { + for (const pending of [[], [operation, operation]]) { + const deliver = mock((_message: InboundMessage): void => undefined); + const resolveSuspended = mock(async () => ({ allow: true })); + const resume = createApprovalResume({ + getAgent: () => ({ + deliver, + history: async () => [ + assistantTurn([ + { id: "call-B", name: "run_shell", command: "echo same" }, + ]), + ], + }), + resolveParkedCallId: (id) => + resolveParkedCallIdFromStore(storeWith(pending), id), + gate: { resolveSuspended } as unknown as PermissionGate, + }); + expect(await resume.handle(suspension("corr-A", "echo same"))).toBe(true); + expect(resolveSuspended).not.toHaveBeenCalled(); + expect(deliver).not.toHaveBeenCalled(); + } +}); + +function deferredResume() { + const lookup = Promise.withResolvers(); + const deliver = mock((_message: InboundMessage): void => undefined); + const history = mock(async (): Promise => []); + const resolveSuspended = mock(async () => ({ allow: true })); + const resolveParkedCallId = mock(() => lookup.promise); + const onDropped = mock((_text: string): void => undefined); + let cancel: (() => void) | undefined; + let current = true; + const resume = createApprovalResume({ + getAgent: () => ({ deliver, history }), + resolveParkedCallId, + gate: { resolveSuspended } as unknown as PermissionGate, + captureGeneration: () => () => current, + onDropped, + registerParkedCancel: (registered) => { + cancel = registered; + }, + }); + return { + lookup, + deliver, + onDropped, + history, + resolveSuspended, + resolveParkedCallId, + resume, + cancel: () => cancel?.(), + registered: () => cancel, + invalidate: () => { + current = false; + }, + }; +} + +describe("approval identity ordering", () => { + test("resolves once before opening the gate", async () => { + const ctx = deferredResume(); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.history).not.toHaveBeenCalled(); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.resolveParkedCallId).toHaveBeenCalledTimes(1); + expect(ctx.resolveSuspended).toHaveBeenCalledTimes(1); + expect(ctx.deliver).toHaveBeenCalledTimes(1); + expect(ctx.registered()).toBeUndefined(); + }); + + for (const cancel of [true, false]) { + test(`${cancel ? "registered cancellation" : "generation change"} during lookup never delivers or opens gate`, async () => { + const ctx = deferredResume(); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + expect(ctx.registered()).toBeDefined(); + if (cancel) ctx.cancel(); + else ctx.invalidate(); + expect(ctx.deliver).not.toHaveBeenCalled(); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.deliver).not.toHaveBeenCalled(); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.registered()).toBeUndefined(); + }); + } + + test("generation change during initial history never delivers", async () => { + const ctx = deferredResume(); + ctx.history.mockImplementation(async () => { + ctx.invalidate(); + return []; + }); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.deliver).not.toHaveBeenCalled(); + }); + + test("generation change during post-gate history cannot deliver an approval", async () => { + const ctx = deferredResume(); + ctx.history.mockResolvedValueOnce([]).mockImplementationOnce(async () => { + ctx.invalidate(); + return []; + }); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.deliver).toHaveBeenCalledTimes(1); + const message = ctx.deliver.mock.calls[0]?.[0]; + if (message === undefined) + throw new Error("expected cancellation rejection"); + expect(decisionBody(message).outcome).toBe("rejected"); + }); + + for (const timedOutCallId of ["call-A", "call-B"]) { + test(`generation change during post-gate history with ${timedOutCallId} timeout preserves exact-call cancellation`, async () => { + const ctx = deferredResume(); + const reading = Promise.withResolvers(); + const history = Promise.withResolvers(); + ctx.history.mockResolvedValueOnce([]).mockImplementationOnce(() => { + reading.resolve(undefined); + return history.promise; + }); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.lookup.resolve("call-A"); + await reading.promise; + expect(ctx.resolveSuspended).toHaveBeenCalledTimes(1); + expect(ctx.registered()).toBeUndefined(); + ctx.invalidate(); + history.resolve([timeoutTurn(timedOutCallId)]); + expect(await handling).toBe(true); + expect(ctx.onDropped).toHaveBeenCalledTimes(1); + expect(ctx.onDropped).toHaveBeenCalledWith(APPROVAL_DROPPED_NOTICE); + expect(ctx.registered()).toBeUndefined(); + if (timedOutCallId === "call-A") { + expect(ctx.deliver).not.toHaveBeenCalled(); + } else { + expect(ctx.deliver).toHaveBeenCalledTimes(1); + const message = ctx.deliver.mock.calls[0]?.[0]; + if (message === undefined) + throw new Error("expected live-call cancellation rejection"); + expect(deliveredCorrelationId(message)).toBe("corr-A"); + expect(decisionBody(message)).toEqual({ + outcome: "rejected", + message: APPROVAL_DROPPED_NOTICE, + }); + } + }); + } + + test("uses the captured agent for both history reads and direct delivery", async () => { + const agent = { + deliver: mock((_message: InboundMessage): void => undefined), + history: mock(async () => []), + }; + const other = { + deliver: mock((_message: InboundMessage): void => undefined), + history: mock(async () => []), + }; + const getAgent = mock(() => agent) + .mockReturnValueOnce(agent) + .mockReturnValue(other); + const resume = createApprovalResume({ + getAgent, + resolveParkedCallId: () => "call-A", + gate: { + resolveSuspended: async () => ({ allow: true }), + } as unknown as PermissionGate, + }); + expect(await resume.handle(suspension("corr-A", "echo same"))).toBe(true); + expect(getAgent).toHaveBeenCalledTimes(1); + expect(agent.history).toHaveBeenCalledTimes(2); + expect(agent.deliver).toHaveBeenCalledTimes(1); + expect(other.history).not.toHaveBeenCalled(); + expect(other.deliver).not.toHaveBeenCalled(); + }); + + test("does not claim atomic expiry detection before a timeout result is observed", async () => { + const pending = new Map([["corr-A", "call-A"]]); + const { resume, delivered } = setup({ + preTurns: [], + resolveParkedCallId: (id) => pending.get(id), + onGate: () => { + pending.clear(); + }, + }); + // Reactor correlation removal precedes queued timeout publication. History + // alone cannot close this interval; atomic admission belongs to the reactor. + expect(await resume.handle(suspension("corr-A", "echo same"))).toBe(true); + expect(delivered).toHaveLength(1); + }); + + test("load errors propagate and clear registration without rejection delivery", async () => { + const ctx = deferredResume(); + const error = new Error("store unavailable"); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.lookup.reject(error); + await expect(handling).rejects.toBe(error); + expect(ctx.registered()).toBeUndefined(); + expect(ctx.deliver).not.toHaveBeenCalled(); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + await expect( + resolveParkedCallIdFromStore( + { + load: async () => { + throw error; + }, + }, + "corr-A", + ), + ).rejects.toBe(error); + }); + + for (const snapshot of [ + undefined, + { name: 42 }, + { name: "run_shell", arguments: { command: 42 } }, + ]) { + for (const identity of ["missing", "expired", "live"] as const) { + test(`invalid snapshot ${JSON.stringify(snapshot)} with ${identity} identity`, async () => { + const ctx = deferredResume(); + if (identity === "expired") + ctx.history.mockResolvedValue([timeoutTurn("call-A")]); + const result = { + type: "suspended", + correlationId: "corr-A", + approvalSnapshot: snapshot, + } as unknown as SendResult; + const handling = ctx.resume.handle(result); + ctx.lookup.resolve(identity === "missing" ? undefined : "call-A"); + expect(await handling).toBe(true); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.deliver).toHaveBeenCalledTimes(identity === "live" ? 1 : 0); + if (identity === "live") { + const message = ctx.deliver.mock.calls[0]?.[0]; + if (message === undefined) throw new Error("expected rejection"); + expect(decisionBody(message).outcome).toBe("rejected"); + } + }); + } + } + + test("observed exact timeout during lookup prevents the gate and delivery", async () => { + const ctx = deferredResume(); + const handling = ctx.resume.handle(suspension("corr-A", "echo same")); + ctx.history.mockResolvedValue([timeoutTurn("call-A")]); + ctx.lookup.resolve("call-A"); + expect(await handling).toBe(true); + expect(ctx.resolveSuspended).not.toHaveBeenCalled(); + expect(ctx.deliver).not.toHaveBeenCalled(); + }); +}); + +describe("approval resume retry re-await", () => { + function retryHarness(outcome: { allow: boolean; message?: string }) { + const delivered: InboundMessage[] = []; + const deliver = mock((message: InboundMessage): void => { + delivered.push(message); + }); + const resolveSuspended = mock(async () => outcome); + const resume = createApprovalResume({ + getAgent: () => ({ + deliver, + history: async () => [], + }), + resolveParkedCallId: () => "call-A", + gate: { resolveSuspended } as unknown as PermissionGate, + }); + return { resume, delivered, deliver, resolveSuspended }; + } + + function onlyDecision(delivered: InboundMessage[]): InboundMessage { + const message = delivered[0]; + if (message === undefined) throw new Error("expected a delivered decision"); + return message; + } + + test("retry after a delivered acceptance reuses it exactly once", async () => { + const { resume, delivered, resolveSuspended } = retryHarness({ + allow: true, + }); + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + expect(delivered).toHaveLength(1); + expect(decisionBody(onlyDecision(delivered)).outcome).toBe("approved"); + + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + expect(delivered).toHaveLength(1); + }); + + test("late duplicate acceptance after a rejected decision is a no-op", async () => { + const { resume, delivered, resolveSuspended } = retryHarness({ + allow: false, + message: "not today", + }); + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(delivered).toHaveLength(1); + expect(decisionBody(onlyDecision(delivered)).outcome).toBe("rejected"); + + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + expect(delivered).toHaveLength(1); + }); + + test("retry without acceptance still gates", async () => { + const delivered: InboundMessage[] = []; + const deliver = mock((message: InboundMessage): void => { + delivered.push(message); + }); + deliver.mockImplementationOnce(() => { + throw new Error("agent is done"); + }); + const resolveSuspended = mock(async () => ({ allow: true })); + const resume = createApprovalResume({ + getAgent: () => ({ + deliver, + history: async () => [], + }), + resolveParkedCallId: () => "call-A", + gate: { resolveSuspended } as unknown as PermissionGate, + }); + await expect( + resume.handle(suspension("corr-A", "echo alpha")), + ).rejects.toThrow("agent is done"); + expect(delivered).toHaveLength(0); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + + expect(await resume.handle(suspension("corr-A", "echo alpha"))).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(2); + expect(delivered).toHaveLength(1); + }); + + test("concurrent duplicate handles share one gate and one deliver", async () => { + const gate = Promise.withResolvers<{ allow: boolean }>(); + const delivered: InboundMessage[] = []; + const deliver = mock((message: InboundMessage): void => { + delivered.push(message); + }); + const resolveSuspended = mock(() => gate.promise); + const resume = createApprovalResume({ + getAgent: () => ({ + deliver, + history: async () => [], + }), + resolveParkedCallId: () => "call-A", + gate: { resolveSuspended } as unknown as PermissionGate, + }); + const first = resume.handle(suspension("corr-A", "echo alpha")); + const second = resume.handle(suspension("corr-A", "echo alpha")); + for (let i = 0; i < 10; i++) await Promise.resolve(); + expect(resolveSuspended).toHaveBeenCalledTimes(1); + gate.resolve({ allow: true }); + expect(await first).toBe(true); + expect(await second).toBe(true); + expect(resolveSuspended).toHaveBeenCalledTimes(1); expect(delivered).toHaveLength(1); }); }); diff --git a/src/session/approval-resume.ts b/src/session/approval-resume.ts index 73c090135..ab3c0a300 100644 --- a/src/session/approval-resume.ts +++ b/src/session/approval-resume.ts @@ -12,7 +12,11 @@ // a rejected one answers it with an error result. import type { Agent, SendResult } from "@intx/agent"; -import type { ApprovalSnapshot, InboundMessage } from "@intx/types/runtime"; +import type { + ApprovalSnapshot, + ContextStore, + InboundMessage, +} from "@intx/types/runtime"; import { type } from "arktype"; import { getLogger } from "@intx/log"; @@ -68,95 +72,41 @@ export function requestFromApprovalSnapshot( return anySecret ? { ...request, scopes: [] } : request; } -// The reactor's approval timeout answers the parked call with this exact -// upstream text (see permission/decline-markers.ts) before removing the -// correlation, so its presence after the suspension watermark marks the -// correlation as settled — but only when the result answers this very call. -// Parallel-parked calls each time out into their own tool result (callId is -// the original tool-call id, not the minted correlationId), so matching text -// alone abandons a still-valid sibling decision. The parked call id must come -// along and match block.callId; without it the text-only scan stays as the -// fallback so a genuinely late decision is still dropped. - -function timeoutResultAfterSuspend( - turns: Awaited>, - fromIndex: number, - parkedCallId: string | undefined, -): string | undefined { - for (const block of turns.slice(fromIndex).flatMap((turn) => turn.content)) { - if ( - block.type === "tool_result" && - (parkedCallId === undefined || block.callId === parkedCallId) && - block.content.some( - (part) => - part.type === "text" && part.text === APPROVAL_TIMEOUT_RESULT_TEXT, - ) - ) { - return block.callId; - } - } - return undefined; -} - -function stableArgs(value: unknown): string { - if (Array.isArray(value)) return `[${value.map(stableArgs).join(",")}]`; - if (typeof value === "object" && value !== null) { - const record = value as Record; - return `{${Object.keys(record) - .sort() - .map((key) => `${JSON.stringify(key)}:${stableArgs(record[key])}`) - .join(",")}}`; - } - return JSON.stringify(value) ?? "null"; +export async function resolveParkedCallIdFromStore( + storage: Pick, + correlationId: string, +): Promise { + const { pendingOperations } = await storage.load(); + const matches = pendingOperations.filter( + (operation) => + operation.kind === "approval" && + operation.correlationId === correlationId, + ); + return matches.length === 1 ? matches[0]?.suspendedCall?.id : undefined; } -// Derive this suspension's parked call id from history: the pre-watermark -// tool_call matching the snapshot's name and arguments that has no tool -// result anywhere yet. A parked call is neither run nor answered, so an -// unanswered match is normally this suspension's own call; a timed-out -// sibling is answered by its timeout result and drops out of the candidates. -// Returns undefined unless exactly one candidate matches. -// -// Known limitation: identical name+args twins are ambiguous. When this -// suspension's own timeout fires while an identical twin sits unanswered, -// the twin is the exact-one survivor, so the settled check misses and the -// late decision is delivered instead of dropped. The sibling-timeout mirror -// (identical args, sibling answered) still resolves to the live call and -// delivers. Telling identical twins apart needs the resolveParkedCallId -// lookup below; the history heuristic cannot do it. -function parkedCallIdFromHistory( +function timeoutResult( turns: Awaited>, - fromIndex: number, - snapshot: ApprovalSnapshot, -): string | undefined { - const answered = new Set(); - for (const turn of turns) { - for (const block of turn.content) { - if (block.type === "tool_result") answered.add(block.callId); - } - } - const wanted = stableArgs(snapshot.arguments ?? {}); - const candidates = new Set(); - for (const turn of turns.slice(0, fromIndex)) { - for (const block of turn.content) { - if ( - block.type === "tool_call" && - block.name === snapshot.name && - stableArgs(block.arguments) === wanted && - !answered.has(block.id) - ) { - candidates.add(block.id); - } - } - } - return candidates.size === 1 ? [...candidates][0] : undefined; + parkedCallId: string, +): boolean { + return turns.some((turn) => + turn.content.some( + (block) => + block.type === "tool_result" && + block.callId === parkedCallId && + block.content.some( + (part) => + part.type === "text" && part.text === APPROVAL_TIMEOUT_RESULT_TEXT, + ), + ), + ); } function decisionMessage( correlationId: string, outcome: "approved" | "rejected", message?: string, -) { +): InboundMessage { const body: { outcome: "approved" | "rejected"; message?: string } = { outcome, }; @@ -169,6 +119,8 @@ function decisionMessage( date: new Date().toISOString(), messageId: `approval-${correlationId}`, interchangeCorrelationId: correlationId, + interchangeType: + outcome === "approved" ? "approval.granted" : "approval.denied", }, flags: [], content: JSON.stringify(body), @@ -177,161 +129,144 @@ function decisionMessage( } export function createApprovalResume(args: { - // Live agent at history/deliver time. TUI occupancy holds this identity - // until the correlated resume is accepted; a generation bump aborts the - // gate rather than retargeting a rebuilt agent. getAgent: () => Pick | undefined; - // TUI session queue. When present, each decision is awaited through this - // seam; exec omits it and uses getAgent().deliver. deliver?: ( message: InboundMessage, stillCurrent: () => boolean, ) => void | Promise; - // TUI: capture at handle() start so interrupt, /clear, or /new during the - // overlay aborts the gate and drops the decision. Exec omits this. captureGeneration?: () => () => boolean; - // TUI: operator-visible notice when an overlay decision is dropped after - // a generation bump. onDropped?: (text: string) => void; - // TUI: interrupt/clear bump this to reject the parked call on the old - // agent before close/rebuild. Cleared when handle returns. registerParkedCancel?: (cancel: (() => void) | undefined) => void; - // Pending-operation lookup: map this suspension's correlationId to the - // parked tool-call id (PendingOperation.suspendedCall.id). The timeout - // result carries the original call id while the suspension carries only - // the minted correlationId, so this is what ties them together. Takes - // precedence over the history derivation below; absent callers fall back - // to it. - // - // Heuristic-only in production: neither the exec nor the TUI caller wires - // this, because the vendored reactor surface ({ start, deliver, abort }) - // exposes no correlationId-to-call lookup (see - // permission/decline-markers.ts), and the suspension snapshot carries - // name+arguments without the call id. Wire this if upstream ever exports - // the pending-operation lookup; until then the history derivation is the - // live path. - resolveParkedCallId?: (correlationId: string) => string | undefined; + resolveParkedCallId: ( + correlationId: string, + ) => string | undefined | Promise; gate: PermissionGate; }): ApprovalResume { - const { getAgent, gate } = args; + // Retry re-await wiring: correlation ids whose decision was handed to the + // reactor reuse that acceptance. A retry after an observed acceptance + // returns without opening the gate or delivering again, so the parked call + // resumes exactly once and a late duplicate acceptance is a no-op. Ids are + // recorded only when the decision is actually handed over — a failed send + // (deliver threw, so nothing reached the reactor) retries as before. + const handedOver = new Set(); + // Settlements currently gating-and-delivering, keyed by correlation id. A + // concurrent duplicate handle shares the one in-flight outcome instead of + // opening a second gate, so no waiter is lost and none double-resumes. + const inflight = new Map>(); - const requireAgent = (): Pick => { - const agent = getAgent(); - if (agent === undefined) { + const settleSuspended = async ( + result: Extract, + ): Promise => { + const generationCurrent = args.captureGeneration?.() ?? (() => true); + const parkedAgent = args.getAgent(); + if (parkedAgent === undefined) throw new Error("approval resume: no live agent"); - } - return agent; - }; - - return { - handle: async (result) => { - if (result.type !== "suspended") return false; - const stillCurrent = args.captureGeneration?.() ?? (() => true); - const parkedAgent = requireAgent(); - const { correlationId, approvalSnapshot } = result; - - let cancelled = false; - const cancelParked = (): void => { - if (cancelled) return; - cancelled = true; + const { correlationId, approvalSnapshot } = result; + let cancelled = false; + let canReject = false; + const stillCurrent = (): boolean => !cancelled && generationCurrent(); + const cancelParked = (): void => { + if (cancelled) return; + cancelled = true; + if (canReject) { parkedAgent.deliver( decisionMessage(correlationId, "rejected", APPROVAL_DROPPED_NOTICE), ); - }; - args.registerParkedCancel?.(cancelParked); - - const dropParked = (): void => { - args.onDropped?.(APPROVAL_DROPPED_NOTICE); - cancelParked(); - }; - - 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; - } - - 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 dropParked = (): void => { + args.onDropped?.(APPROVAL_DROPPED_NOTICE); + cancelParked(); + }; + args.registerParkedCancel?.(cancelParked); + try { + // The resolver captures the paired store synchronously before its first await. + const parkedCallId = await args.resolveParkedCallId(correlationId); + if (!stillCurrent()) { + dropParked(); + return true; + } + if (parkedCallId === undefined) return true; + const initialHistory = await parkedAgent.history(); + if (!stillCurrent()) { + dropParked(); + return true; + } + if (timeoutResult(initialHistory, parkedCallId)) return true; + canReject = true; - const outcome = await gate.resolveSuspended(request, stillCurrent); - if (!stillCurrent()) { - dropParked(); - return true; + const deliverDecision = async ( + message: InboundMessage, + ): Promise => { + if (!stillCurrent()) return false; + if (args.deliver !== undefined) { + await args.deliver(message, stillCurrent); + } else { + parkedAgent.deliver(message); } + handedOver.add(correlationId); + return true; + }; + const request = + approvalSnapshot === undefined + ? null + : requestFromApprovalSnapshot(approvalSnapshot, correlationId); + if (request === null) { args.registerParkedCancel?.(undefined); - const history = await requireAgent().history(); - const parkedCallId = - args.resolveParkedCallId?.(correlationId) ?? - parkedCallIdFromHistory(history, turnsAtSuspend, approvalSnapshot); - const matchedTimeoutCallId = timeoutResultAfterSuspend( - history, - turnsAtSuspend, - parkedCallId, + await deliverDecision( + decisionMessage( + correlationId, + "rejected", + "approval surface unavailable", + ), ); - if (matchedTimeoutCallId !== undefined) { - // The reactor already answered the parked call (its approval timeout - // fired while the surface was still up). Delivering now would append - // the raw decision JSON as an uncorrelated user turn — drop and log. - // The matched call id rides along so a fallback drop (parkedCallId - // undefined) can still be attributed to the timeout that caused it. - logger.warn`late approval decision dropped correlation=${correlationId} timeoutCall=${matchedTimeoutCallId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`; - 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); } + const outcome = await args.gate.resolveSuspended(request, stillCurrent); + if (!stillCurrent()) { + dropParked(); + return true; + } + args.registerParkedCancel?.(undefined); + const history = await parkedAgent.history(); + const timedOut = timeoutResult(history, parkedCallId); + if (timedOut) canReject = false; + if (!stillCurrent()) { + dropParked(); + return true; + } + if (timedOut) { + logger.warn`late approval decision dropped correlation=${correlationId} timeoutCall=${parkedCallId} outcome=${outcome?.allow === true ? "approved" : "rejected"}`; + return true; + } + await deliverDecision( + decisionMessage( + correlationId, + outcome?.allow === true ? "approved" : "rejected", + outcome?.allow === true ? undefined : outcome?.message, + ), + ); + return true; + } finally { + args.registerParkedCancel?.(undefined); + } + }; + + return { + handle: (result) => { + if (result.type !== "suspended") return Promise.resolve(false); + const { correlationId } = result; + if (handedOver.has(correlationId)) return Promise.resolve(true); + const ongoing = inflight.get(correlationId); + if (ongoing !== undefined) return ongoing; + const task = settleSuspended(result); + inflight.set(correlationId, task); + const forget = (): void => { + if (inflight.get(correlationId) === task) + inflight.delete(correlationId); + }; + task.then(forget, forget); + return task; }, }; } diff --git a/src/tui/deliver-agent-message.test.ts b/src/tui/deliver-agent-message.test.ts index 904182c64..7a6a88243 100644 --- a/src/tui/deliver-agent-message.test.ts +++ b/src/tui/deliver-agent-message.test.ts @@ -4,7 +4,7 @@ import { deliverAgentMessage, deliveryResultNotice, runGenerationGuardedDeliver, -} from "./deliver-agent-message.js"; +} from "./delivery-queue.js"; describe("deliverAgentMessage", () => { test("reports session-unavailable without calling deliver when rebuild failed", async () => { diff --git a/src/tui/deliver-agent-message.ts b/src/tui/deliver-agent-message.ts deleted file mode 100644 index 0a70b3f07..000000000 --- a/src/tui/deliver-agent-message.ts +++ /dev/null @@ -1,117 +0,0 @@ -/** - * Guards a queued/steer deliver against a mid-rebuild or closed agent. The shell - * paints the delivered row and pops the queue item before this runs, so the - * caller must settle ownership from the structured result — a swallowed failure - * here means the transcript claims delivery for a message that never reached - * the agent. - */ -import { AgentClosedError } from "@intx/agent"; - -export type AgentDeliveryNotDeliveredReason = - | "agent-closed" - | "session-unavailable" - | "superseded" - | "preparation-failed"; - -export type AgentDeliveryResult = - | { readonly status: "accepted" } - | { - readonly status: "not-delivered"; - readonly reason: AgentDeliveryNotDeliveredReason; - readonly detail: string; - } - | { - readonly status: "uncertain"; - readonly detail: string; - }; - -export interface DeliverAgentMessageDeps { - getFatalBuildError: () => Error | null; - deliverToLiveAgent: () => void; -} - -export async function deliverAgentMessage( - deps: DeliverAgentMessageDeps, -): Promise { - const fatal = deps.getFatalBuildError(); - if (fatal !== null) { - return { - status: "not-delivered", - reason: "session-unavailable", - detail: fatal.message, - }; - } - try { - deps.deliverToLiveAgent(); - return { status: "accepted" }; - } catch (err) { - if (err instanceof AgentClosedError) { - return { - status: "not-delivered", - reason: "agent-closed", - detail: err.message, - }; - } - return { - status: "uncertain", - detail: err instanceof Error ? err.message : String(err), - }; - } -} - -/** - * Settles a deliver that was enqueued on the serial operation queue against - * the shoot generation captured at enqueue time. The queue is FIFO with no - * preemption, so a deliver queued ahead of a reload still executes after the - * reload has replaced the agent — the generation must be re-checked when the - * queued closure runs, not just when it enqueues. A stale deliver takes the - * `onStale` path (the caller reports `not-delivered`); a current deliver runs - * the real settle. This is what closes the reload-vs-async-deliver race: a - * reload that lands while a continuation answer is queued wins, and the stale - * answer is dropped instead of reaching the replaced agent. - */ -export async function runGenerationGuardedDeliver(options: { - stillCurrent: () => boolean; - run: () => Promise; - onStale: () => AgentDeliveryResult; -}): Promise { - if (!options.stillCurrent()) { - return options.onStale(); - } - return options.run(); -} - -/** Operator-facing copy for a settled delivery that did not accept. */ -export function deliveryResultNotice( - result: Exclude, - disposition: "restored" | "deferred" | "none" = "none", -): string { - if (result.status === "uncertain") { - const base = `Delivery failed: ${result.detail}. Delivery status is uncertain; review the transcript before sending again.`; - return appendDisposition(base, disposition); - } - if (result.reason === "agent-closed") { - if (disposition === "restored") { - return "Message not delivered because the agent closed. It is back in the prompt; press Enter to send it."; - } - if (disposition === "deferred") { - return "Message not delivered because the agent closed. Your current draft is unchanged; the message will return to the prompt after you send it."; - } - return "Message not delivered because the agent closed."; - } - const base = `Message not delivered: ${result.detail}`; - return appendDisposition(base, disposition); -} - -function appendDisposition( - base: string, - disposition: "restored" | "deferred" | "none", -): string { - if (disposition === "restored") { - return `${base} It is back in the prompt; press Enter to send it.`; - } - if (disposition === "deferred") { - return `${base} Your current draft is unchanged; the message will return to the prompt after you send it.`; - } - return base; -} diff --git a/src/tui/delivery-queue.ts b/src/tui/delivery-queue.ts index 933a154cb..6e2f12114 100644 --- a/src/tui/delivery-queue.ts +++ b/src/tui/delivery-queue.ts @@ -17,12 +17,128 @@ * and no product chord wires it anymore — leave the path for tests/API only. */ +import { AgentClosedError } from "@intx/agent"; import type { PendingImageAttachment } from "./image-attachments.js"; -import type { AgentDeliveryResult } from "./deliver-agent-message.js"; import type { ProductHostDeliver } from "./product-host.js"; import { ASK_DIRECTOR_WAKE_PREFIX } from "../subagent/fleet-report.js"; import { MAILBOX_MAIL_WAKE_PREFIX } from "../subagent/mailbox-mail-drive.js"; +export type AgentDeliveryNotDeliveredReason = + | "agent-closed" + | "session-unavailable" + | "superseded" + | "preparation-failed"; + +export type AgentDeliveryResult = + | { readonly status: "accepted" } + | { + readonly status: "not-delivered"; + readonly reason: AgentDeliveryNotDeliveredReason; + readonly detail: string; + } + | { + readonly status: "uncertain"; + readonly detail: string; + }; + +export interface DeliverAgentMessageDeps { + getFatalBuildError: () => Error | null; + deliverToLiveAgent: () => void; +} + +/** + * Guards a queued/steer deliver against a mid-rebuild or closed agent. The + * shell paints the delivered row and pops the queue item before this runs, so + * the caller must settle ownership from the structured result — a swallowed + * failure here means the transcript claims delivery for a message that never + * reached the agent. + */ +export async function deliverAgentMessage( + deps: DeliverAgentMessageDeps, +): Promise { + const fatal = deps.getFatalBuildError(); + if (fatal !== null) { + return { + status: "not-delivered", + reason: "session-unavailable", + detail: fatal.message, + }; + } + try { + deps.deliverToLiveAgent(); + return { status: "accepted" }; + } catch (err) { + if (err instanceof AgentClosedError) { + return { + status: "not-delivered", + reason: "agent-closed", + detail: err.message, + }; + } + return { + status: "uncertain", + detail: err instanceof Error ? err.message : String(err), + }; + } +} + +/** + * Settles a deliver that was enqueued on the serial operation queue against + * the shoot generation captured at enqueue time. The queue is FIFO with no + * preemption, so a deliver queued ahead of a reload still executes after the + * reload has replaced the agent — the generation must be re-checked when the + * queued closure runs, not just when it enqueues. A stale deliver takes the + * `onStale` path (the caller reports `not-delivered`); a current deliver runs + * the real settle. This is what closes the reload-vs-async-deliver race: a + * reload that lands while a continuation answer is queued wins, and the stale + * answer is dropped instead of reaching the replaced agent. + */ +export async function runGenerationGuardedDeliver(options: { + stillCurrent: () => boolean; + run: () => Promise; + onStale: () => AgentDeliveryResult; +}): Promise { + if (!options.stillCurrent()) { + return options.onStale(); + } + return options.run(); +} + +/** Operator-facing copy for a settled delivery that did not accept. */ +export function deliveryResultNotice( + result: Exclude, + disposition: "restored" | "deferred" | "none" = "none", +): string { + if (result.status === "uncertain") { + const base = `Delivery failed: ${result.detail}. Delivery status is uncertain; review the transcript before sending again.`; + return appendDisposition(base, disposition); + } + if (result.reason === "agent-closed") { + if (disposition === "restored") { + return "Message not delivered because the agent closed. It is back in the prompt; press Enter to send it."; + } + if (disposition === "deferred") { + return "Message not delivered because the agent closed. Your current draft is unchanged; the message will return to the prompt after you send it."; + } + return "Message not delivered because the agent closed."; + } + const base = `Message not delivered: ${result.detail}`; + return appendDisposition(base, disposition); +} + +function appendDisposition( + base: string, + disposition: "restored" | "deferred" | "none", +): string { + if (disposition === "restored") { + return `${base} It is back in the prompt; press Enter to send it.`; + } + if (disposition === "deferred") { + return `${base} Your current draft is unchanged; the message will return to the prompt after you send it.`; + } + return base; +} + export type QueueKind = "queue" | "steer"; export interface QueueItem { diff --git a/src/tui/queued-delivery-hop.test.ts b/src/tui/queued-delivery-hop.test.ts index a21310cac..f3a0c9cc1 100644 --- a/src/tui/queued-delivery-hop.test.ts +++ b/src/tui/queued-delivery-hop.test.ts @@ -12,10 +12,10 @@ import { createLiveSteerDeliver, createSessionOperationQueue, routeQueuedDelivery, + type AgentDeliveryResult, type DeliverySettle, type QueueItem, } from "./delivery-queue.js"; -import type { AgentDeliveryResult } from "./deliver-agent-message.js"; function lastHopPort(bridgeRef: { current: SessionBridge | undefined }) { const sends: string[] = []; diff --git a/src/tui/queued-delivery.test.ts b/src/tui/queued-delivery.test.ts index de8e2d0b7..274afaab1 100644 --- a/src/tui/queued-delivery.test.ts +++ b/src/tui/queued-delivery.test.ts @@ -8,8 +8,8 @@ import { createSessionOperationQueue, routeQueuedDelivery, SESSION_IDENTITY_ABORT_REASON, + type AgentDeliveryResult, } from "./delivery-queue.js"; -import type { AgentDeliveryResult } from "./deliver-agent-message.js"; const image: PendingImageAttachment = { id: "img-1", diff --git a/src/tui/runner/session.ts b/src/tui/runner/session.ts index 083ad5558..cf45941c3 100644 --- a/src/tui/runner/session.ts +++ b/src/tui/runner/session.ts @@ -57,7 +57,10 @@ import { type LiveSessionSources, } from "../../session/assemble-runtime.js"; import type { CompactionArchive } from "../../session/compaction-archive.js"; -import { createApprovalResume } from "../../session/approval-resume.js"; +import { + createApprovalResume, + resolveParkedCallIdFromStore, +} from "../../session/approval-resume.js"; import { createReactorAuthorize } from "../../permission/reactor-authorize.js"; import { buildShellBackgroundMessage, @@ -76,6 +79,10 @@ import { createSessionCostAccumulator } from "../../cost/session-cost.js"; import { createDeliveryGeneration, createSessionOperationQueue, + deliverAgentMessage, + deliveryResultNotice, + runGenerationGuardedDeliver, + type AgentDeliveryResult, } from "../delivery-queue.js"; import { createCorrelationAcceptance } from "../correlation-acceptance.js"; import { createApprovalDeliverer } from "../approval-delivery.js"; @@ -89,12 +96,6 @@ import { detectLanguageServerAvailable } from "../../agent/lsp-availability.js"; import type { SessionMode } from "../../config/session-mode.js"; import { WorkflowHost, type WorkflowHostState } from "../../workflows/host.js"; import type { ToolWatchdogConfig } from "../tool-execution-watchdog.js"; -import { - deliverAgentMessage, - deliveryResultNotice, - runGenerationGuardedDeliver, - type AgentDeliveryResult, -} from "../deliver-agent-message.js"; import { createProviderFailureAttemptTracker } from "../provider/failure-attempt.js"; import { getTelemetry, liveTelemetry } from "../../telemetry/singleton.js"; import { @@ -502,8 +503,6 @@ export async function assembleTUISession( // Reload, interrupt, compaction continuation, and proxy deliver share one queue // so a rebuild never races an in-flight deliver. const sessionOps = createSessionOperationQueue(); - // No resolveParkedCallId: the vendored reactor exposes no - // correlationId-to-call lookup, so the history heuristic is the path. // The deliverer bounds the acceptance wait so one stuck delivery fails fast // with diagnostics instead of wedging the sessionOps tail for every later // approval (send_input answers, interrupt_agent releases, ask_operator). @@ -513,6 +512,12 @@ export async function assembleTUISession( }); const approvalResume = createApprovalResume({ getAgent: () => state.currentAgent, + resolveParkedCallId: (correlationId) => { + const storage = state.currentStorage; + if (storage === null) + throw new Error("approval resume: no context store"); + return resolveParkedCallIdFromStore(storage, correlationId); + }, captureGeneration: deliveryGeneration.capture, onDropped: (text) => state.systemNotice?.(text), registerParkedCancel: (cancel) => { diff --git a/src/tui/runner/state.ts b/src/tui/runner/state.ts index 42b0101b1..8f3dffa97 100644 --- a/src/tui/runner/state.ts +++ b/src/tui/runner/state.ts @@ -32,7 +32,7 @@ import type { ProviderFailureAttempt } from "../provider/failure-attempt.js"; import type { ScopedApproval } from "../../permission/admin.js"; import type { ConnectedMcpServer, RunState } from "../../session/state.js"; import type { PendingImageAttachment } from "../image-attachments.js"; -import type { AgentDeliveryResult } from "../deliver-agent-message.js"; +import type { AgentDeliveryResult } from "../delivery-queue.js"; import type { SubmitOutcome } from "./submit.js"; import type { mountRunnerHost } from "./host.js"; import { EventEmitter } from "node:events"; diff --git a/src/tui/runner/submit.ts b/src/tui/runner/submit.ts index f42c9b7f3..725d33782 100644 --- a/src/tui/runner/submit.ts +++ b/src/tui/runner/submit.ts @@ -35,8 +35,8 @@ import { createLeftoverSend, createLiveSteerDeliver, routeQueuedDelivery, + type AgentDeliveryResult, } from "../delivery-queue.js"; -import type { AgentDeliveryResult } from "../deliver-agent-message.js"; import type { InferenceAttemptIdentity } from "./state.js"; import { tuiSendFailureMessage } from "./send-failure-message.js"; import type { ProviderFailureAttempt } from "../provider/failure-attempt.js"; diff --git a/src/tui/runtime-bridge.ts b/src/tui/runtime-bridge.ts index a31276e39..d5ad1922f 100644 --- a/src/tui/runtime-bridge.ts +++ b/src/tui/runtime-bridge.ts @@ -72,7 +72,7 @@ import { import { deliveryResultNotice, type AgentDeliveryResult, -} from "./deliver-agent-message.js"; +} from "./delivery-queue.js"; import { toolCallRow } from "./diff.js"; import { toolResultRow } from "./mcp-view.js"; import { diff --git a/tests/integration/harness.ts b/tests/integration/harness.ts index 2701ee8d5..e3cc03970 100644 --- a/tests/integration/harness.ts +++ b/tests/integration/harness.ts @@ -74,6 +74,7 @@ export const INTEGRATION_SOURCE: InferenceSource = { const INTEGRATION_SECRET = "integration-test-key"; export interface IntegrationSession { + storage: ContextStore; harness: Harness; cwd: string; workdir: string; @@ -248,7 +249,7 @@ export async function openIntegrationSession( ? innerAgent : createPrimaryDeliveryAdmission(innerAgent, primaryArchive); - return { harness, cwd, workdir, agent, toolset }; + return { harness, cwd, workdir, agent, toolset, storage: storageForAgent }; } export async function closeIntegrationSession( diff --git a/tests/integration/reactor-approval-suspend.test.ts b/tests/integration/reactor-approval-suspend.test.ts index bfc04cf15..57d37a131 100644 --- a/tests/integration/reactor-approval-suspend.test.ts +++ b/tests/integration/reactor-approval-suspend.test.ts @@ -6,6 +6,7 @@ import { createPermissionGate } from "../../src/permission/gate.js"; import { createReactorAuthorize } from "../../src/permission/reactor-authorize.js"; import { createApprovalResume, + resolveParkedCallIdFromStore, requestFromApprovalSnapshot, } from "../../src/session/approval-resume.js"; import { @@ -84,6 +85,16 @@ describe("integration — reactor approval suspend/resume", () => { expect(result.type).toBe("suspended"); if (result.type !== "suspended") return; + const parkedCallId = await resolveParkedCallIdFromStore( + session.storage, + result.correlationId, + ); + const parkedCalls = (await session.agent.history()) + .flatMap((historyTurn) => historyTurn.content) + .filter((block) => block.type === "tool_call"); + expect(parkedCalls).toHaveLength(1); + expect(parkedCallId).toBe(defined(parkedCalls[0]).id); + // The parked call carried an approver-facing snapshot and parked on a // gate keyed by the correlationId (reactor.gate.blocked). expect(result.approvalSnapshot?.name).toBe("run_shell"); @@ -112,6 +123,8 @@ describe("integration — reactor approval suspend/resume", () => { expect(request?.subject).toBe("curl -sS https://example.com"); const resume = createApprovalResume({ getAgent: () => session.agent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(session.storage, correlationId), gate: ctx.gate, }); const handling = resume.handle(result); @@ -171,6 +184,8 @@ describe("integration — reactor approval suspend/resume", () => { const resume = createApprovalResume({ getAgent: () => session.agent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(session.storage, correlationId), gate: ctx.gate, }); const handling = resume.handle(result); @@ -221,6 +236,8 @@ describe("integration — reactor approval suspend/resume", () => { const resume = createApprovalResume({ getAgent: () => session.agent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(session.storage, correlationId), gate: ctx.gate, }); const handling = resume.handle(result); @@ -334,6 +351,8 @@ describe("integration — reactor approval suspend/resume", () => { const resume = createApprovalResume({ getAgent: () => session.agent, + resolveParkedCallId: (correlationId) => + resolveParkedCallIdFromStore(session.storage, correlationId), gate: ctx.gate, }); const handling = resume.handle(result); diff --git a/tests/unit/approval-resume.test.ts b/tests/unit/approval-resume.test.ts index fc8ce1d92..d2ae50c9b 100644 --- a/tests/unit/approval-resume.test.ts +++ b/tests/unit/approval-resume.test.ts @@ -81,7 +81,11 @@ describe("approval resume late-decision guard", () => { test("a decision after the reactor settled the correlation is dropped", async () => { const turns = [userTurn()]; const { agent, gate, delivered } = harness(turns, true); - const resume = createApprovalResume({ getAgent: () => agent, gate }); + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => agent, + gate, + }); expect(await resume.handle(SUSPENDED)).toBe(true); expect(delivered).toEqual([]); @@ -90,7 +94,11 @@ describe("approval resume late-decision guard", () => { test("a live decision is still delivered", async () => { const turns = [userTurn()]; const { agent, gate, delivered } = harness(turns, false); - const resume = createApprovalResume({ getAgent: () => agent, gate }); + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => agent, + gate, + }); expect(await resume.handle(SUSPENDED)).toBe(true); expect(delivered).toHaveLength(1); @@ -105,13 +113,17 @@ describe("approval resume late-decision guard", () => { expect(correlationHeaders(delivered[0]).messageId).toBe("approval-corr-1"); }); - test("an approval timeout from before the suspension does not suppress delivery", async () => { + test("an exact approval timeout already present before lookup suppresses delivery", async () => { const turns = [userTurn(), approvalTimedOutTurn()]; const { agent, gate, delivered } = harness(turns, false); - const resume = createApprovalResume({ getAgent: () => agent, gate }); + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => agent, + gate, + }); expect(await resume.handle(SUSPENDED)).toBe(true); - expect(delivered).toHaveLength(1); + expect(delivered).toHaveLength(0); }); }); @@ -125,6 +137,7 @@ describe("approval resume delivery", () => { }; let customResolved = false; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, deliver: async (message) => { await Promise.resolve(); @@ -147,6 +160,7 @@ describe("approval resume delivery", () => { test("undefined agent throws instead of returning true", async () => { const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => undefined, gate: { resolveSuspended: async () => ({ allow: true }), @@ -163,6 +177,7 @@ describe("approval resume delivery", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, gate: { resolveSuspended: async () => ({ allow: true }), @@ -184,7 +199,11 @@ describe("approval resume delivery", () => { return { allow: true }; }, } as unknown as PermissionGate; - const resume = createApprovalResume({ getAgent: () => agent, gate }); + const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", + getAgent: () => agent, + gate, + }); expect(await resume.handle(SUSPENDED)).toBe(true); expect(delivered).toHaveLength(1); @@ -206,6 +225,7 @@ describe("approval resume generation capture", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, captureGeneration: generation.capture, deliver: (message, stillCurrent) => { @@ -236,6 +256,7 @@ describe("approval resume generation capture", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, captureGeneration: generation.capture, deliver: (message, stillCurrent) => { @@ -271,6 +292,7 @@ describe("approval resume generation capture", () => { }; let current: typeof agentA | typeof agentB = agentA; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => current, captureGeneration: generation.capture, deliver: (message, stillCurrent) => { @@ -325,6 +347,7 @@ describe("approval resume generation capture", () => { }); let finishOverlay: ((outcome: { allow: boolean }) => void) | undefined; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, captureGeneration: generation.capture, registerParkedCancel: (cancel) => { @@ -453,6 +476,7 @@ describe("approval resume persist Allow after interrupt", () => { const delivered: unknown[] = []; const notices: string[] = []; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => ({ deliver: (message: unknown) => delivered.push(message), history: async () => [userTurn()], @@ -488,6 +512,7 @@ describe("approval resume stillCurrent at resolve", () => { const generation = createDeliveryGeneration(); let atResolve: boolean | undefined; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => ({ deliver: () => undefined, history: async () => [userTurn()], @@ -563,6 +588,7 @@ describe("approval resume occupancy until correlation", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, deliver: (message) => enqueue(async () => { @@ -627,6 +653,7 @@ describe("approval resume occupancy until correlation", () => { history: async () => [userTurn()], }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => agent, deliver: (message) => enqueue(async () => { @@ -697,6 +724,7 @@ describe("approval resume occupancy until correlation", () => { reloadIfIdle: () => undefined, }; const resume = createApprovalResume({ + resolveParkedCallId: () => "call-ask", getAgent: () => ({ deliver: () => undefined, history: async () => [userTurn()], diff --git a/tests/unit/reactor-approval-acceptance.test.ts b/tests/unit/reactor-approval-acceptance.test.ts new file mode 100644 index 000000000..655eba097 --- /dev/null +++ b/tests/unit/reactor-approval-acceptance.test.ts @@ -0,0 +1,521 @@ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; +import { createInboundMessage } from "@intx/mime"; +import type { + ContextStore, + InboundMessage, + PendingOperation, + ReactorInboundEvent, + ReactorState, +} from "@intx/types/runtime"; +import { + createReactor, + type ReactorConfig, + type ReactorEmittedEvent, +} from "../../vendor/intx-inference/src/reactor.js"; +import { createDefaultDependencies } from "../../vendor/intx-inference/src/providers/index.js"; +import { createSessionOperationQueue } from "../../src/tui/delivery-queue.js"; + +const call = { id: "parked-call", name: "test_tool", arguments: {} }; +const epoch = 100_000; +function pending( + timeoutAt: number | undefined = epoch + 1000, +): PendingOperation { + return { + correlationId: "approval", + kind: "approval", + gateId: "gate", + registeredAt: epoch, + suspendedCall: call, + ...(timeoutAt === undefined ? {} : { timeoutAt }), + }; +} +function decision( + outcome: "approved" | "rejected" = "approved", +): InboundMessage { + const message = createInboundMessage({ + from: "signal@local", + to: "agent@local", + correlationId: "approval", + content: JSON.stringify({ outcome }), + }); + message.headers.interchangeType = + outcome === "approved" ? "approval.granted" : "approval.denied"; + return message; +} +function deferred() { + let resolve!: (value: T) => void; + const promise = new Promise((yes) => { + resolve = yes; + }); + return { promise, resolve }; +} +async function drain() { + for (let i = 0; i < 100; i++) await Promise.resolve(undefined); +} + +const cleanups: (() => Promise)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +async function harness( + options: { + op?: PendingOperation; + validator?: ReactorConfig["correlationValidator"]; + grant?: () => void; + event?: (event: ReactorEmittedEvent) => void; + } = {}, +) { + let now = epoch; + const clock = spyOn(Date, "now").mockImplementation(() => now); + const timers: (() => void)[] = []; + const timer = spyOn(globalThis, "setTimeout").mockImplementation((( + callback: () => void, + ) => { + timers.push(callback); + return timers.length; + }) as typeof setTimeout); + const clear = spyOn(globalThis, "clearTimeout").mockImplementation( + (): void => undefined, + ); + const events: ReactorEmittedEvent[] = []; + const directed: ReactorInboundEvent[] = []; + const snapshots: ReactorState[] = []; + const grants: string[] = []; + const tools: string[] = []; + let hold: Promise | undefined; + const closing = deferred(); + const store: ContextStore = { + load: async () => ({ + turns: [ + { + role: "assistant", + content: [{ type: "tool_call", ...call }], + timestamp: epoch, + }, + ], + pendingOperations: [options.op ?? pending()], + tokenUsage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + thinking: 0, + }, + connectorState: null, + }), + setConnectorState: () => undefined, + commit: async ({ message }) => ({ hash: "test", message, timestamp: now }), + branch: async () => undefined, + log: async () => [], + readAt: async () => [], + writeBlob: async () => undefined, + readBlob: async () => { + throw new Error("unused"); + }, + writePrompt: async () => undefined, + writeResponse: async () => undefined, + writeManifest: async () => undefined, + writeTurns: async () => undefined, + writeMetadata: async () => undefined, + readManifestHistory: async () => [], + }; + const reactor = createReactor({ + sessionId: "atomic-approval", + source: { + id: "test", + provider: "anthropic", + model: "test", + baseURL: "https://example.com", + credentialId: "test", + }, + deps: createDefaultDependencies(), + contextStore: store, + gateTimeout: 1000, + ...(options.validator === undefined + ? {} + : { correlationValidator: options.validator }), + beforeToolExtensions: [ + { + beforeTool: async () => ({ type: "allow" }), + grantOneShot(id) { + grants.push(id); + options.grant?.(); + }, + }, + ], + toolRunner: { + async run(tool) { + tools.push(tool.id); + return { callId: tool.id, content: "executed" }; + }, + }, + onEvent(event) { + events.push(event); + options.event?.(event); + }, + director: { + async decide(event, state, caps) { + directed.push(event); + snapshots.push(state); + if (hold !== undefined) await Promise.race([hold, closing.promise]); + if (event.type === "resume.execute_tools") + return caps.executeTools(event.calls, false, true); + return caps.wait(); + }, + }, + }); + cleanups.push(async () => { + closing.resolve(undefined); + hold = undefined; + reactor.abort("admin_kill"); + await drain(); + timer.mockRestore(); + clear.mockRestore(); + clock.mockRestore(); + }); + reactor.start(); + await drain(); + expect(events.some((e) => e.type === "reactor.start")).toBe(true); + return { + reactor, + events, + directed, + snapshots, + grants, + tools, + advance(value: number) { + now = value; + }, + expire() { + now = epoch + 1000; + const fire = timers[0]; + if (!fire) throw new Error("gate timer missing"); + fire(); + }, + hold(value: Promise | undefined) { + hold = value; + }, + }; +} +function accepted(h: Awaited>) { + return h.events.filter((e) => e.type === "message.correlated"); +} +function received(h: Awaited>) { + return h.events.filter((e) => e.type === "message.received"); +} +function resumes(h: Awaited>) { + return h.directed.filter( + (e) => e.type === "resume.execute_tools" || e.type === "resume.tool_result", + ); +} +function expectDiscard(h: Awaited>) { + expect(accepted(h)).toHaveLength(0); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(0); + expect(h.tools).toHaveLength(0); +} + +describe("atomic typed approval acceptance", () => { + test("expiry before timeout result publication cannot become ordinary input", async () => { + const h = await harness(); + const hold = deferred(); + h.hold(hold.promise); + h.reactor.deliver( + createInboundMessage({ + from: "user@local", + to: "agent@local", + content: "hold director", + }), + ); + await drain(); + h.expire(); + const stale = decision(); + stale.content = "not JSON"; + h.reactor.deliver(stale); + await drain(); + expect(accepted(h)).toHaveLength(0); + expect(received(h)).toHaveLength(1); + expect(h.grants).toHaveLength(0); + expect(h.tools).toHaveLength(0); + expect(resumes(h)).toHaveLength(0); + hold.resolve(undefined); + h.hold(undefined); + await drain(); + expect(resumes(h)).toHaveLength(1); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).toContain( + "approval timed out", + ); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).not.toContain("not JSON"); + }); + test("actual session operation queue drains an expired approval to Reactor.deliver", async () => { + const h = await harness(); + const queue = createSessionOperationQueue(); + const held = deferred(); + queue.enqueue(() => held.promise); + queue.enqueue(async () => { + h.reactor.deliver(decision()); + }); + await drain(); + h.expire(); + held.resolve(undefined); + await queue.awaitTail(); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(1); + }); + test("expiry during validator await discards the claimed decision", async () => { + const valid = deferred(); + const h = await harness({ validator: { validate: () => valid.promise } }); + h.reactor.deliver(decision()); + await drain(); + h.expire(); + valid.resolve(true); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(1); + }); + for (const outcome of ["approved", "rejected"] as const) + test(`live ${outcome} and concurrent/postconsumption duplicates resume once`, async () => { + const valid = deferred(); + const h = await harness({ validator: { validate: () => valid.promise } }); + h.reactor.deliver(decision(outcome)); + h.reactor.deliver(decision(outcome)); + valid.resolve(true); + await drain(); + h.reactor.deliver(decision(outcome)); + await drain(); + expect(accepted(h)).toHaveLength(1); + expect(received(h)).toHaveLength(0); + expect(resumes(h)).toHaveLength(1); + expect(h.grants).toHaveLength(outcome === "approved" ? 1 : 0); + expect(h.tools).toHaveLength(outcome === "approved" ? 1 : 0); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).toContain( + outcome === "approved" ? "executed" : "denied by approver", + ); + }); + for (const deadline of [epoch - 1, epoch]) + test(`persisted deadline ${deadline} is not revived by the clamped timer`, async () => { + const h = await harness({ op: pending(deadline) }); + h.reactor.deliver(decision()); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + }); + test("deadline equality wins before actual timer callback", async () => { + const h = await harness(); + h.advance(epoch + 1000); + h.reactor.deliver(decision()); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + }); + test("termination during validation prevents successful acceptance", async () => { + const valid = deferred(); + const h = await harness({ validator: { validate: () => valid.promise } }); + h.reactor.deliver(decision()); + h.reactor.abort("admin_kill"); + await drain(); + valid.resolve(true); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + }); + for (const failure of ["false", "throw"] as const) + test(`validator ${failure} releases claim for valid retry`, async () => { + let first = true; + const h = await harness({ + validator: { + async validate() { + if (!first) return true; + first = false; + if (failure === "throw") throw new Error("validator failed"); + return false; + }, + }, + }); + h.reactor.deliver(decision()); + await drain(); + expectDiscard(h); + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(1); + expect(resumes(h)).toHaveLength(1); + expect(h.tools).toHaveLength(1); + }); + for (const suspended of [true, false]) + test(`contradictory header/body with suspendedCall=${suspended} cannot resume`, async () => { + const op = pending(); + if (!suspended) delete op.suspendedCall; + const h = await harness({ op }); + const message = decision(); + message.content = JSON.stringify({ outcome: "rejected" }); + h.reactor.deliver(message); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + if (suspended) + expect(h.events.some((e) => e.type === "reactor.error")).toBe(true); + }); + for (const callback of ["grant", "event"] as const) + test(`synchronous duplicate from ${callback} does not resume twice`, async () => { + let deliver = (): void => undefined; + const duplicate = () => deliver(); + const h = await harness( + callback === "grant" + ? { grant: duplicate } + : { + event: (event) => { + if (event.type === "message.correlated") duplicate(); + }, + }, + ); + deliver = () => h.reactor.deliver(decision()); + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(1); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(1); + expect(h.tools).toHaveLength(1); + expect(resumes(h)).toHaveLength(1); + }); + for (const action of ["throw", "abort"] as const) + test(`grant hook ${action} cannot publish successful acceptance`, async () => { + let abort = (): void => undefined; + const h = await harness({ + grant() { + if (action === "throw") throw new Error("grant failed"); + abort(); + }, + }); + abort = () => h.reactor.abort("admin_kill"); + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(0); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(1); + expect(h.tools).toHaveLength(0); + expect(resumes(h)).toHaveLength(0); + }); + for (const action of ["throw", "abort"] as const) + test(`acceptance observer ${action} happens after consumption, without a second resume`, async () => { + let reenter = (): void => undefined; + const h = await harness({ + event(event) { + if (event.type !== "message.correlated") return; + reenter(); + if (action === "throw") throw new Error("observer failed"); + }, + }); + reenter = () => { + h.reactor.deliver(decision()); + if (action === "abort") h.reactor.abort("admin_kill"); + }; + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(1); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(1); + expect(resumes(h).length).toBeLessThanOrEqual(1); + expect(h.tools.length).toBeLessThanOrEqual(1); + expect(h.events.some((e) => e.type === "reactor.done")).toBe(true); + }); + test("gate deadline applies when the operation has no persisted deadline", async () => { + const op = pending(); + delete op.timeoutAt; + const h = await harness({ op }); + h.advance(epoch + 1000); + h.reactor.deliver(decision()); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + }); + test("grant callback advancing the clock cannot accept an expired operation", async () => { + let expire = (): void => undefined; + const h = await harness({ grant: () => expire() }); + expire = () => h.advance(epoch + 1000); + h.reactor.deliver(decision()); + await drain(); + expect(accepted(h)).toHaveLength(0); + expect(received(h)).toHaveLength(0); + expect(h.grants).toHaveLength(1); + expect(h.tools).toHaveLength(0); + expect(resumes(h)).toHaveLength(0); + }); + for (const correlation of [undefined, "unknown"]) + test(`typed unknown correlation ${correlation} discards before parsing`, async () => { + const h = await harness(); + const message = decision(); + message.content = "malformed"; + if (correlation === undefined) + delete message.headers.interchangeCorrelationId; + else message.headers.interchangeCorrelationId = correlation; + h.reactor.deliver(message); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + expect(h.events.some((e) => e.type === "reactor.error")).toBe(false); + }); + test("live malformed typed decisions retain the fatal error policy", async () => { + const h = await harness(); + const message = decision(); + message.content = "malformed"; + h.reactor.deliver(message); + await drain(); + expectDiscard(h); + expect(resumes(h)).toHaveLength(0); + expect(h.events.some((e) => e.type === "reactor.error")).toBe(true); + expect(h.events.some((e) => e.type === "reactor.done")).toBe(true); + }); + for (const failure of ["false", "throw"] as const) + test(`nonapproval validator ${failure} retains ordinary fallback`, async () => { + const h = await harness({ + validator: { + async validate() { + if (failure === "throw") throw new Error("invalid sender"); + return false; + }, + }, + }); + const message = decision(); + message.headers.interchangeType = "offering.response"; + h.reactor.deliver(message); + await drain(); + expect(received(h)).toHaveLength(1); + expect(accepted(h)).toHaveLength(0); + expect(h.grants).toHaveLength(0); + expect(resumes(h)).toHaveLength(0); + }); + test("ordinary unknown correlations, approval-shaped text and nonapproval responses retain routing", async () => { + const op = pending(); + delete op.suspendedCall; + const h = await harness({ op }); + for (const correlationId of [undefined, "unknown"]) { + h.reactor.deliver( + createInboundMessage({ + from: "user@local", + to: "agent@local", + content: JSON.stringify({ outcome: "approved" }), + ...(correlationId === undefined ? {} : { correlationId }), + }), + ); + } + h.reactor.deliver( + createInboundMessage({ + from: "worker@local", + to: "agent@local", + content: "async response", + correlationId: "approval", + }), + ); + await drain(); + expect(received(h)).toHaveLength(2); + expect(accepted(h)).toHaveLength(1); + expect(h.grants).toHaveLength(0); + expect(h.tools).toHaveLength(0); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).toContain( + "async response", + ); + expect(JSON.stringify(h.snapshots.at(-1)?.turns)).toContain("approved"); + }); +}); diff --git a/tests/unit/tui/approval-reload-during-suspend.test.ts b/tests/unit/tui/approval-reload-during-suspend.test.ts index 93b253c6e..f6125baa8 100644 --- a/tests/unit/tui/approval-reload-during-suspend.test.ts +++ b/tests/unit/tui/approval-reload-during-suspend.test.ts @@ -117,6 +117,7 @@ describe("pendingReload during resolveSuspended vs deliver enqueue", () => { }; const resume = createApprovalResume({ getAgent: () => agent, + resolveParkedCallId: () => "call-ask", deliver: (message) => enqueue(async () => { agent.deliver(message); diff --git a/vendor/intx-inference/PATCHES.md b/vendor/intx-inference/PATCHES.md index 0461fcd3f..7b100bc47 100644 --- a/vendor/intx-inference/PATCHES.md +++ b/vendor/intx-inference/PATCHES.md @@ -26,6 +26,32 @@ Re-carryable, "Kill candidate" = Droppable, and "Companion" entries are Re-carryable but ride their primary patch's disposition (they ship out or die with it). +## reactor-ts-atomic-approval-acceptance + +**Promotion candidate.** `reactor.ts` classifies native `approval.granted` / +`approval.denied` headers as conditional decision intent, not authentication. +Its private correlation disposition distinguishes ordinary input from discarded +approvals; stale, ineligible and already-claimed decisions never enter history +or emit `message.received`. `correlation.ts` documents the typed-only exception +to validator false/throw fallback, without changing the validator signature. + +The claim spans validation, parsing, dispatch, consumption and publication in +one `try/finally`. Before parsing and after synchronous grant hooks, acceptance +requires the same pending operation and live gate, an accepting lifecycle, and +strictly unexpired gate and persisted deadlines. A clamped rehydration timer +cannot revive an expired persisted operation. Typed targets without a suspended +call are ineligible; eligible bodies must agree with the header. Consumption +and continuation enqueue run without callbacks before `message.correlated`. +Grant hooks may reenter or abort; a failed transition publishes no acceptance. +Arbitrary extension side effects are not rolled back, and observers throwing +after the transition retain the existing fatal-error policy. + +**Re-carry:** preserve all acceptance checks and the ordinary-message path when +merging upstream correlation changes. Deterministic real-reactor coverage lives +in `tests/unit/reactor-approval-acceptance.test.ts`. A never-settling validator +retains its claim until settlement; there are no tombstones or discard receipts. +This is a local patch, not a change to the upstream pin or retrieval metadata. + ### 2026-09-07 re-sync (upstream `0205b07b`) Every entry below was re-carried against the new pin; none was dropped as diff --git a/vendor/intx-inference/src/correlation.ts b/vendor/intx-inference/src/correlation.ts index 974a03d03..9ab5d95a5 100644 --- a/vendor/intx-inference/src/correlation.ts +++ b/vendor/intx-inference/src/correlation.ts @@ -15,7 +15,11 @@ import type { InboundMessage, PendingOperation } from "@intx/types/runtime"; export interface CorrelationValidator { /** * Return true if `message` is a valid resolution for `pending`. - * False causes the message to be delivered as a regular uncorrelated event. + * False (or a thrown error) falls back to a regular uncorrelated event, + * except explicit approval.granted/approval.denied messages: those are + * discarded, leaving a still-live pending operation available for retry. + * These headers classify intent; they do not authenticate the sender. + * Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance */ validate( pending: PendingOperation, diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts index 379adc5cc..08f18e732 100644 --- a/vendor/intx-inference/src/reactor.ts +++ b/vendor/intx-inference/src/reactor.ts @@ -495,9 +495,10 @@ export function createReactor(config: ReactorConfig): Reactor { | { mode: "gate-cleared" } | { mode: "error_result"; result: ToolResult }; - // Decide how a correlated approval-kind pending operation resumes, granting - // any one-shot bypass synchronously so no delivery can interleave between the - // grant and the re-dispatch enqueued by the caller. An operation that carries + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + // Decide how a correlated approval-kind pending operation resumes. Grant + // callbacks can reenter; the caller owns the claim and final live checks. + // An operation that carries // a `suspendedCall` is an ask-flow suspension: the approver's decision routes // it down the re-dispatch rail. An operation without one is an async-tool // pending marker, which resumes on the normal gate-cleared rail. @@ -509,6 +510,8 @@ export function createReactor(config: ReactorConfig): Reactor { function resumePendingOperation( op: PendingOperation, message: InboundMessage, + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + expectedOutcome: "approved" | "rejected" | undefined, ): ResumeDispatch { if (op.suspendedCall === undefined) { return { mode: "gate-cleared" }; @@ -536,6 +539,13 @@ export function createReactor(config: ReactorConfig): Reactor { ); } + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + if (expectedOutcome !== undefined && decision.outcome !== expectedOutcome) { + throw new Error( + `Correlated approval decision for ${op.correlationId} contradicts its header`, + ); + } + switch (op.kind) { case "approval": switch (decision.outcome) { @@ -569,44 +579,63 @@ export function createReactor(config: ReactorConfig): Reactor { } } - async function tryCorrelate(message: InboundMessage): Promise { - const correlationId = message.headers.interchangeCorrelationId; - if (correlationId === undefined) return false; - - if (correlatingIds.has(correlationId)) return false; - const pending = correlations.lookup(correlationId); - if (pending === undefined) return false; - - correlatingIds.add(correlationId); + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + type CorrelationDisposition = "correlated" | "ordinary" | "discarded"; - if (correlationValidator !== undefined) { - let valid: boolean; - try { - valid = await correlationValidator.validate(pending, message); - } catch (cause) { - logger.warn`Correlation validator threw for ${correlationId}: ${cause}`; - correlatingIds.delete(correlationId); - return false; - } - if (!valid) { - correlatingIds.delete(correlationId); - return false; - } + async function tryCorrelate( + message: InboundMessage, + ): Promise { + const expectedOutcome = + message.headers.interchangeType === "approval.granted" + ? "approved" + : message.headers.interchangeType === "approval.denied" + ? "rejected" + : undefined; + const typedApproval = expectedOutcome !== undefined; + const unmatched = typedApproval ? "discarded" : "ordinary"; + const correlationId = message.headers.interchangeCorrelationId; + if (correlationId === undefined || correlatingIds.has(correlationId)) { + return unmatched; } + const pending = correlations.lookup(correlationId); + if (pending === undefined) return unmatched; + const claimedGate = gates.findByCorrelationId(correlationId); + + const isLiveApproval = (): boolean => { + const now = Date.now(); + return ( + running && !done && !shutdownStarted && + !queue.some((event) => event.type === "abort") && + correlations.lookup(correlationId) === pending && + pending.kind === "approval" && + pending.suspendedCall !== undefined && + claimedGate !== undefined && + claimedGate.gateId === pending.gateId && + gates.findByCorrelationId(correlationId) === claimedGate && + now < claimedGate.timeoutAt && + (pending.timeoutAt === undefined || now < pending.timeoutAt) + ); + }; - // Capture the operation before removal so the resume dispatch can read its - // kind and suspended call. Removal happens only after the dispatch is - // decided, all inside this correlatingIds-guarded critical section so a - // double-deliver early-returns rather than double-dispatching. - const op = pending; - - // A finally clears the in-flight marker on every exit — success included. - // Without it the success path leaves the id in the set forever, leaking - // one entry per correlated message for the life of the session. - // // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-correlating-ids-leak + correlatingIds.add(correlationId); try { - const dispatch = resumePendingOperation(op, message); + if (typedApproval && !isLiveApproval()) return "discarded"; + if (correlationValidator !== undefined) { + let valid: boolean; + try { + valid = await correlationValidator.validate(pending, message); + } catch (cause) { + logger.warn`Correlation validator threw for ${correlationId}: ${cause}`; + return unmatched; + } + if (!valid) return unmatched; + } + if (typedApproval && !isLiveApproval()) return "discarded"; + const dispatch = resumePendingOperation(pending, message, expectedOutcome); + // Extension grants are synchronous but can reenter, abort or advance the + // clock. Only the internal consume-and-enqueue below is callback-free. + if (typedApproval && !isLiveApproval()) return "discarded"; const gate = gates.findByCorrelationId(correlationId); switch (dispatch.mode) { @@ -668,17 +697,18 @@ export function createReactor(config: ReactorConfig): Reactor { break; } } + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + // Observers may reenter: publish only after consumption and continuation + // enqueue, while the claim still excludes a duplicate delivery. + emit({ + type: "message.correlated", + seq: nextSeq(), + data: { message, correlationId }, + }); + return "correlated"; } finally { correlatingIds.delete(correlationId); } - - emit({ - type: "message.correlated", - seq: nextSeq(), - data: { message, correlationId }, - }); - - return true; } // ------------------------------------------------------------------------- @@ -1809,9 +1839,10 @@ export function createReactor(config: ReactorConfig): Reactor { function processDelivery(message: InboundMessage): void { void (async () => { - let correlated: boolean; + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + let disposition: CorrelationDisposition; try { - correlated = await tryCorrelate(message); + disposition = await tryCorrelate(message); } catch (cause) { // A correlation-path invariant failed (e.g. a malformed approval // decision). Surface it as a fatal reactor error rather than a silent @@ -1830,7 +1861,8 @@ export function createReactor(config: ReactorConfig): Reactor { } return; } - if (!correlated) { + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance + if (disposition === "ordinary") { emit({ type: "message.received", seq: nextSeq(),