diff --git a/e2e/integration-harness.ts b/e2e/integration-harness.ts index e87cc9245..6a7bb29fb 100644 --- a/e2e/integration-harness.ts +++ b/e2e/integration-harness.ts @@ -149,6 +149,8 @@ export interface OpenIntegrationSessionOpts { retryPolicy?: RetryPolicy; /** Fields merged over harness.deps, e.g. a real scheduler when retry delays must elapse. */ depsOverrides?: Partial; + /** Reuse an existing session cwd, so the agent loads that session's store. */ + cwd?: string; } export async function openIntegrationSession( @@ -170,7 +172,8 @@ export async function openIntegrationSession( throw new Error("openIntegrationSession requires at least one source"); } const defaultSource = opts.defaultSourceId ?? primarySource.id; - const cwd = mkdtempSync(join(tmpdir(), "corbits-integration-cwd-")); + const cwd = + opts.cwd ?? mkdtempSync(join(tmpdir(), "corbits-integration-cwd-")); const workdir = join(cwd, ".agent-state", "integration-session"); const evidenceArchiveHolder: { current: CompactionArchive | undefined } = { current: undefined, @@ -380,7 +383,7 @@ export async function openIntegrationSession( }; } -export async function closeIntegrationSession( +async function releaseIntegrationAgent( session: IntegrationSession, ): Promise { try { @@ -388,10 +391,31 @@ export async function closeIntegrationSession( } finally { await session.toolset.dispose(); session.harness.dispose(); + } +} + +export async function closeIntegrationSession( + session: IntegrationSession, +): Promise { + try { + await releaseIntegrationAgent(session); + } finally { rmSync(session.cwd, { recursive: true, force: true }); } } +/** + * The TUI reload path: close the agent without a final commit, then build a + * fresh one on the same session store. The new session owns the cwd. + */ +export async function rebuildIntegrationSession( + session: IntegrationSession, + opts: OpenIntegrationSessionOpts, +): Promise { + await releaseIntegrationAgent(session); + return openIntegrationSession({ ...opts, cwd: session.cwd }); +} + export interface TurnResult { events: ReactorEmittedEvent[]; reply: string; diff --git a/e2e/reactor-approval-reload.test.ts b/e2e/reactor-approval-reload.test.ts new file mode 100644 index 000000000..0ccf547f7 --- /dev/null +++ b/e2e/reactor-approval-reload.test.ts @@ -0,0 +1,280 @@ +/** + * An approved ask whose agent is rebuilt from the same store before the next + * cycle commit (the TUI's deferred reload) must not leave the parked call's + * pending operation behind: the rebuilt reactor would re-arm its gate and, at + * the original deadline, answer an already-answered call with "approval timed + * out". + */ +import { afterEach, describe, expect, spyOn, test } from "bun:test"; + +import type { ReactorEmittedEvent } from "@intx/inference"; +import type { ConversationTurn } from "@intx/types/runtime"; + +import { createPermissionGate } from "../src/permission/gate.js"; +import { createReactorAuthorize } from "../src/permission/reactor-authorize.js"; +import { + createApprovalResume, + resolveParkedCallIdFromStore, +} from "../src/session/approval-resume.js"; +import { + closeIntegrationSession, + openIntegrationSession, + rebuildIntegrationSession, + runUntilDone, + runUntilSuspended, + type IntegrationSession, +} from "./integration-harness.js"; + +// Ask-tier, but answered locally: nothing listens on the discard port. +const CLOSED_PORT_PROBE = "curl -sS --max-time 1 http://127.0.0.1:9/"; + +// Reactor gates run on the real clock (setTimeout + Date.now), not the +// inference harness clock, so the approval deadline is driven here. Only +// timers at least this long are held; everything shorter keeps real timing. +const HELD_TIMER_MIN_MS = 30 * 60_000; + +interface HeldTimer { + dueAt: number; + fire: () => void; + cleared: boolean; +} + +function holdLongTimers() { + const realNow = Date.now.bind(Date); + const realSetTimeout = globalThis.setTimeout; + const realClearTimeout = globalThis.clearTimeout; + let offset = 0; + const held = new Map(); + const now = spyOn(Date, "now").mockImplementation(() => realNow() + offset); + const set = spyOn(globalThis, "setTimeout").mockImplementation((( + fire: () => void, + delay?: number, + ...args: unknown[] + ) => { + if (delay === undefined || delay < HELD_TIMER_MIN_MS) { + return realSetTimeout(fire, delay, ...args); + } + const handle = {}; + held.set(handle, { dueAt: Date.now() + delay, fire, cleared: false }); + return handle; + }) as typeof setTimeout); + const clear = spyOn(globalThis, "clearTimeout").mockImplementation((( + handle: Parameters[0], + ) => { + const timer = + typeof handle === "object" && handle !== null + ? held.get(handle) + : undefined; + if (timer !== undefined) { + timer.cleared = true; + return; + } + realClearTimeout(handle); + }) as typeof clearTimeout); + return { + /** Move the clock to `at` and fire every held timer due by then. */ + advanceTo(at: number): void { + offset = at - realNow(); + for (const timer of held.values()) { + if (timer.cleared || timer.dueAt > at) continue; + timer.cleared = true; + timer.fire(); + } + }, + restore(): void { + clear.mockRestore(); + set.mockRestore(); + now.mockRestore(); + }, + }; +} + +function deferredApprovalGate() { + let release: ((outcome: { allow: boolean }) => void) | undefined; + return { + asked: () => release !== undefined, + gate: createPermissionGate({ + approvals: [], + interactive: true, + skipPermissions: false, + auto: false, + reactorGated: true, + requestApproval: () => + new Promise((resolve) => { + release = resolve; + }), + }), + approve: () => release?.({ allow: true }), + }; +} + +async function waitFor( + predicate: () => boolean, + what: string, + ms = 5_000, +): Promise { + const deadline = performance.now() + ms; + while (!predicate()) { + if (performance.now() > deadline) throw new Error(`timed out: ${what}`); + await new Promise((resolve) => setTimeout(resolve, 5)); + } +} + +function toolResultsFor( + turns: readonly ConversationTurn[], + callId: string, +): string[] { + return turns + .flatMap((turn) => turn.content) + .flatMap((block) => + block.type === "tool_result" && block.callId === callId + ? [JSON.stringify(block.content)] + : [], + ); +} + +function collect(session: IntegrationSession): ReactorEmittedEvent[] { + const events: ReactorEmittedEvent[] = []; + void (async () => { + for await (const event of session.agent.stream()) events.push(event); + })().catch(() => undefined); + return events; +} + +const cleanups: (() => Promise | void)[] = []; +afterEach(async () => { + for (const cleanup of cleanups.splice(0).reverse()) await cleanup(); +}); + +interface Reloaded { + session: IntegrationSession; + /** Events from the rebuilt agent, collected from its start. */ + events: ReactorEmittedEvent[]; + callId: string; + deadline: number; + clock: ReturnType; +} + +/** + * Parks `command` on an ask, approves it, and once `reloadWhen` holds for + * the approved run, rebuilds the agent on the same store. The model turn + * after the approved result stays open, so the reload always lands before + * the reply terminal commits the cycle. + */ +async function approveThenReload( + command: string, + reloadWhen: ( + session: IntegrationSession, + events: ReactorEmittedEvent[], + ) => boolean, +): Promise { + const ctx = deferredApprovalGate(); + const sessionOpts = { + permissionGate: ctx.gate, + authorize: createReactorAuthorize(ctx.gate), + }; + let session = await openIntegrationSession(sessionOpts); + cleanups.push(() => closeIntegrationSession(session)); + + session.harness.scenario.replyOnce("anthropic", { + toolCalls: [{ name: "run_shell", args: { command } }], + }); + session.harness.scenario.stall(); + + const turn = await runUntilSuspended(session, "Probe the local port."); + if (turn.result.type !== "suspended") throw new Error("did not park"); + const { correlationId } = turn.result; + const parked = (await session.storage.load()).pendingOperations.find( + (op) => op.correlationId === correlationId, + ); + const callId = parked?.suspendedCall?.id; + const deadline = parked?.timeoutAt; + if (callId === undefined || deadline === undefined) { + throw new Error("parked call has no persisted call id or deadline"); + } + + const resume = createApprovalResume({ + getAgent: () => session.agent, + gate: ctx.gate, + resolveParkedCallId: (id) => + resolveParkedCallIdFromStore(session.storage, id), + }); + const handling = resume.handle(turn.result); + await waitFor(ctx.asked, "approval prompt"); + ctx.approve(); + expect(await handling).toBe(true); + void session.harness + .run({ wallClockBudgetMs: 30_000 }) + .catch(() => undefined); + const before = session; + await waitFor(() => reloadWhen(before, turn.events), "reload point"); + + const clock = holdLongTimers(); + cleanups.push(() => clock.restore()); + session = await rebuildIntegrationSession(session, sessionOpts); + return { session, events: collect(session), callId, deadline, clock }; +} + +/** Moves past the original approval deadline and lets the reactor react. */ +async function passDeadline(reloaded: Reloaded): Promise { + reloaded.clock.advanceTo(reloaded.deadline + 1_000); + await reloaded.session.harness + .run({ wallClockBudgetMs: 2_000 }) + .catch(() => undefined); + await new Promise((resolve) => setTimeout(resolve, 50)); +} + +describe("integration: approving an ask, then reloading the agent", () => { + test.serial( + "after the approved result: the rebuilt agent continues, and the old deadline adds no result", + async () => { + const reloaded = await approveThenReload( + CLOSED_PORT_PROBE, + (session) => session.harness.scenario.matchedRequests().length === 2, + ); + const { session, callId } = reloaded; + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + + session.harness.scenario.replyOnce("anthropic", { text: "Port closed." }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Port closed."); + + await passDeadline(reloaded); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + const results = toolResultsFor(await session.agent.history(), callId); + expect(results).toHaveLength(1); + expect(results.join("")).not.toContain("approval timed out"); + }, + 20_000, + ); + + // The reload interrupts the approved call, which the closing reactor + // answers itself. What an interrupted call should report is not pinned + // here; only that the old deadline cannot add a second result. + test.serial( + "during the approved call: the old deadline adds no second result", + async () => { + const reloaded = await approveThenReload( + `sleep 20; ${CLOSED_PORT_PROBE}`, + (_session, events) => events.some((e) => e.type === "tool.start"), + ); + const { session, callId } = reloaded; + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + + await passDeadline(reloaded); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + expect( + toolResultsFor(await session.agent.history(), callId), + ).toHaveLength(1); + }, + 20_000, + ); +}); diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 8c78df32c..d08a9eb9b 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -2268,3 +2268,222 @@ describe("createOptimizedContextStore metadata ownership (CL-10667)", () => { expect((await resumed.load()).tokenUsage).toEqual(USAGE_TWO); }); }); + +describe("createOptimizedContextStore load drops pending operations for answered calls", () => { + function callTurn(callId: string, timestamp: number): ConversationTurn { + return { + role: "assistant", + content: [ + { type: "tool_call", id: callId, name: "run_shell", arguments: {} }, + ], + timestamp, + }; + } + + function resultTurn( + callId: string, + text: string, + timestamp: number, + ): ConversationTurn { + return { + role: "user", + content: [ + { type: "tool_result", callId, content: [{ type: "text", text }] }, + ], + timestamp, + }; + } + + function approvalOp(correlationId: string, callId: string) { + return { + correlationId, + kind: "approval" as const, + registeredAt: 1_700_000_000_000, + gateId: `pending-${correlationId}`, + timeoutAt: Date.now() + 60_000, + suspendedCall: { id: callId, name: "run_shell", arguments: {} }, + }; + } + + function markerOp(correlationId: string) { + return { + correlationId, + kind: "approval" as const, + registeredAt: 1_700_000_000_000, + gateId: `pending-${correlationId}`, + }; + } + + function placeCrashState( + dir: string, + turns: ConversationTurn[], + pendingOperations: unknown[], + ): void { + fs.writeFileSync(path.join(dir, TURNS_FILE), jsonl(turns)); + fs.writeFileSync( + path.join(dir, "metadata.json"), + JSON.stringify({ pendingOperations, connectorState: null }, null, 2), + ); + } + + // The crash window between the result-turn write and the metadata op-removal + // write leaves turns answering a parked call while metadata.json still lists + // its pending operation. load() is the only path into the rebuilt + // reactor's rehydrateGates, so the stale operation must die here. + test("drops an answered operation while keeping unanswered and marker operations", async () => { + const dir = tempDir(); + placeCrashState( + dir, + [callTurn("call-answered", 1), resultTurn("call-answered", "ok", 2)], + [ + approvalOp("stale-corr", "call-answered"), + approvalOp("live-corr", "call-live"), + markerOp("marker-corr"), + ], + ); + + const store = await createOptimizedContextStore(dir); + const loaded = await store.load(); + expect( + loaded.pendingOperations.map((op) => op.correlationId).sort(), + ).toEqual(["live-corr", "marker-corr"]); + }); + + test("repeated loads are stable and emit no metadata write", async () => { + const dir = tempDir(); + placeCrashState( + dir, + [callTurn("call-answered", 1), resultTurn("call-answered", "ok", 2)], + [approvalOp("stale-corr", "call-answered"), markerOp("marker-corr")], + ); + + const store = await createOptimizedContextStore(dir); + const before = fs.readFileSync(path.join(dir, "metadata.json"), "utf8"); + const first = await store.load(); + const second = await store.load(); + expect(second.pendingOperations).toEqual(first.pendingOperations); + expect(first.pendingOperations.map((op) => op.correlationId)).toEqual([ + "marker-corr", + ]); + expect(fs.readFileSync(path.join(dir, "metadata.json"), "utf8")).toBe( + before, + ); + }); + + test("the next metadata write persists the removal", async () => { + const dir = tempDir(); + placeCrashState( + dir, + [callTurn("call-answered", 1), resultTurn("call-answered", "ok", 2)], + [ + approvalOp("stale-corr", "call-answered"), + approvalOp("live-corr", "call-live"), + ], + ); + + const store = await createOptimizedContextStore(dir); + const loaded = await store.load(); + await store.writeMetadata({ + pendingOperations: loaded.pendingOperations, + tokenUsage: loaded.tokenUsage, + }); + + const persisted = JSON.parse( + fs.readFileSync(path.join(dir, "metadata.json"), "utf8"), + ) as { pendingOperations: { correlationId: string }[] }; + expect(persisted.pendingOperations.map((op) => op.correlationId)).toEqual([ + "live-corr", + ]); + + const resumed = await createOptimizedContextStore(dir); + expect( + (await resumed.load()).pendingOperations.map((op) => op.correlationId), + ).toEqual(["live-corr"]); + }); + + test("crash-window replay across segments yields no re-armable gate", async () => { + const dir = tempDir(); + fs.writeFileSync( + path.join(dir, TURNS_FILE), + jsonl([callTurn("call-answered", 1)]), + ); + fs.writeFileSync( + path.join(dir, segmentFileName(TURNS_FILE, 1)), + jsonl([resultTurn("call-answered", "ok", 2)]), + ); + fs.writeFileSync( + path.join(dir, "metadata.json"), + JSON.stringify({ + pendingOperations: [ + approvalOp("stale-corr", "call-answered"), + approvalOp("live-corr", "call-live"), + markerOp("marker-corr"), + ], + connectorState: null, + }), + ); + + const store = await createOptimizedContextStore(dir); + const loaded = await store.load(); + const answered = new Set( + loaded.turns.flatMap((turn) => + turn.content.flatMap((block) => + block.type === "tool_result" ? [block.callId] : [], + ), + ), + ); + // Every operation load() hands the rebuilt reactor is safe to rehydrate: + // none parks a call the turns already answer. + for (const op of loaded.pendingOperations) { + if (op.suspendedCall !== undefined) { + expect(answered.has(op.suspendedCall.id)).toBe(false); + } + } + expect( + loaded.pendingOperations.map((op) => op.correlationId).sort(), + ).toEqual(["live-corr", "marker-corr"]); + }); + + test("matches a scrubbed on-disk operation by call id", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + const secret = { api_key: "hunter2" }; + await store.writeMetadata({ + pendingOperations: [ + { + ...approvalOp("scrubbed-corr", "call-scrubbed"), + suspendedCall: { + id: "call-scrubbed", + name: "run_shell", + arguments: secret, + }, + approvalSnapshot: { + name: "run_shell", + description: "runs a shell command", + inputSchema: {}, + arguments: secret, + }, + }, + ], + tokenUsage: { + input: 0, + output: 0, + cacheRead: 0, + cacheWrite: 0, + thinking: 0, + }, + }); + // The on-disk approval snapshot is redacted while suspendedCall keeps + // the raw call; matching must work by call id alone. + const raw = fs.readFileSync(path.join(dir, "metadata.json"), "utf8"); + expect(raw).toContain("[redacted"); + expect(raw).toContain("call-scrubbed"); + await store.writeTurns([ + callTurn("call-scrubbed", 1), + resultTurn("call-scrubbed", "ok", 2), + ]); + + const resumed = await createOptimizedContextStore(dir); + expect((await resumed.load()).pendingOperations).toEqual([]); + }); +}); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 0cc038c1b..bc3e35e69 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -336,6 +336,22 @@ function readMetadataDocument(raw: string): Pick< }; } +/** + * Calls the turn history already answers. A crash between the result-turn + * write and the metadata op-removal write leaves turns answering a parked + * call while metadata.json still lists its pending operation; matching is by + * call id, which survives the on-disk approval-snapshot scrub. + */ +function answeredCallIds(turns: readonly ConversationTurn[]): Set { + const answered = new Set(); + for (const turn of turns) { + for (const block of turn.content) { + if (block.type === "tool_result") answered.add(block.callId); + } + } + return answered; +} + /** * metadata.json + sidecar merge. Sidecar wins over a legacy tokenUsage field, * which wins over empty. Parse failures stay inside this function: a torn @@ -1674,7 +1690,35 @@ export async function createSessionStores( ); metadata = emptyMetadata(); } - return { turns, ...metadata }; + // Crash-window filter: load() is the only path into the rebuilt + // reactor's gate rehydration, so a pending operation whose parked call + // the turns already answer dies here instead of re-arming its gate. + // Marker operations carry no suspendedCall and pass through. Disk is + // untouched (no write-in-read): the removal converges on the next + // normal metadata write, which publishes because the filtered payload + // differs from the seeded disk bytes. + const answered = answeredCallIds(turns); + const stale = metadata.pendingOperations.filter( + (op) => + op.suspendedCall !== undefined && answered.has(op.suspendedCall.id), + ); + for (const op of stale) { + log.warn( + "dropped pending operation {correlationId}: its parked call {callId} already has a result", + { + correlationId: op.correlationId, + callId: op.suspendedCall?.id ?? "", + }, + ); + } + if (stale.length === 0) return { turns, ...metadata }; + return { + turns, + ...metadata, + pendingOperations: metadata.pendingOperations.filter( + (op) => !stale.includes(op), + ), + }; }, // Buffer is carried only by this store: base.writeMetadata is never // called through the wrapper, so base's pendingConnectorState would diff --git a/vendor/intx-inference/PATCHES.md b/vendor/intx-inference/PATCHES.md index aff0b14ef..c4a069082 100644 --- a/vendor/intx-inference/PATCHES.md +++ b/vendor/intx-inference/PATCHES.md @@ -469,6 +469,23 @@ segments in the working tree, and `load()` (the path an interrupt-rebuild takes) reads them, so the tool exchange survives a rebuild exactly as the old per-batch commit did. +Pending-operation durability contract: a resolved approval's pending +operation leaves `metadata.json` together with the turn that answers the +parked call, never later. `tryCorrelate`'s redispatch and `error_result` +paths and `onGateCleared`'s timeout path set `pendingOperationsChanged` +when they drop the op from memory; the shared turns-only write +(`writeCycleTurns`, called from the `executeTools` tail and from the loop's +`resume.tool_result` branch right after the synthetic error result is +appended) writes metadata after the turns whenever that flag is set, and +`writeMetadata` clears it. Without this, a teardown between the redispatch +and the next terminal left the op on disk; a reactor rebuilt from that store +re-armed the answered call's gate at its original deadline and injected a +second `tool_result` for the same call id, failing the well-formedness +tripwire. Tool cycles that leave pending operations untouched still write +turns only, so the coalescing savings hold. A crash between the turns write +and the metadata write can still leave a stale op for an answered call; +that narrower window is the rehydration guard's concern, not this patch's. + **Disposition:** Promotion candidate. Performance coalescing — collapses K git commits per agentic turn to one deferred commit while preserving rebuild-from-disk durability. @@ -710,7 +727,7 @@ revisit point is the next vendored sync (see `docs/VENDORING.md`). | errors-ts-classify-abort-reason | Add optional `reason` param to `classifyAbortError`, carried as `raw: { origin }` | Alexander Guy | This ledger (#errors-ts-classify-abort-reason) | Next vendored sync | | harness-ts-inactivity-on-semantic-progress | Gate the inactivity watchdog's re-arm on parsed events, not raw SSE bytes | Alexander Guy | This ledger (#harness-ts-inactivity-on-semantic-progress) | Next vendored sync | | reactor-ts-correlating-ids-leak | Wrap `tryCorrelate` in try/finally so `correlatingIds` clears on success dispatch paths | Alexander Guy | This ledger (#reactor-ts-correlating-ids-leak) | Next vendored sync | -| reactor-ts-tool-cycle-turns-only (supersedes reactor-ts-checkpoint-after-tool-cycle) | `executeTools` writes turns to store segments only and defers the cycle commit to terminals | Alexander Guy | This ledger (#reactor-ts-tool-cycle-turns-only) | Next vendored sync | +| reactor-ts-tool-cycle-turns-only (supersedes reactor-ts-checkpoint-after-tool-cycle) | `executeTools` writes turns to store segments only and defers the cycle commit to terminals; metadata rides along only when a resolved approval dropped a pending operation | Alexander Guy | This ledger (#reactor-ts-tool-cycle-turns-only) | Next vendored sync | | reactor-ts-skip-unchanged-history (+ reactor-ts-last-written-turns-revision) | Skip `contextStore.writeTurns` when `getTurnsRevision()` is unchanged | Alexander Guy | This ledger (#reactor-ts-skip-unchanged-history) | Next vendored sync | | reactor-ts-compact-publish-then-memory | Persist compact blobs, stage `writeTurns`, commit, then `replaceTurns` so memory stays on the old generation until publication | Alexander Guy | This ledger (#reactor-ts-compact-publish-then-memory) | Next vendored sync | | reactor-ts-after-checkpoint-director-only | Gate `afterCheckpoint` on `hasOverride` so auto-commits do not emit it | Alexander Guy | This ledger (#reactor-ts-after-checkpoint-director-only) | Next vendored sync | diff --git a/vendor/intx-inference/src/reactor.test.ts b/vendor/intx-inference/src/reactor.test.ts index c0e1fa37a..e72eb7c50 100644 --- a/vendor/intx-inference/src/reactor.test.ts +++ b/vendor/intx-inference/src/reactor.test.ts @@ -2443,6 +2443,7 @@ describe("createReactor — tool runner failures", () => { // the behavior a regressed per-batch commit would break. let commits = 0; let turnWrites = 0; + let metadataWrites = 0; const countingStore: ContextStore = { async load() { return { @@ -2487,7 +2488,7 @@ describe("createReactor — tool runner failures", () => { turnWrites += 1; }, async writeMetadata() { - /* noop */ + metadataWrites += 1; }, async readManifestHistory() { throw new Error("not implemented"); @@ -2518,8 +2519,10 @@ describe("createReactor — tool runner failures", () => { await waitFor("reactor.done"); // One segment write per tool batch; exactly one commit, at the terminal. + // No pending operation changed, so metadata waits for that commit too. expect(turnWrites).toBe(2); expect(commits).toBe(1); + expect(metadataWrites).toBe(1); }); }); @@ -5376,6 +5379,181 @@ describe("createReactor — approval resume error result", () => { }); }); +// --------------------------------------------------------------------------- +// A resolved approval's pending-operation removal reaches disk with the turn +// that answers the parked call, not at the next terminal commit +// --------------------------------------------------------------------------- + +describe("createReactor — resolved approval persists pending-operation removal", () => { + // The first inference parks on the ask gate. The continuation after the + // parked call is answered never finishes on its own: it holds until the + // reactor aborts it, standing in for a teardown that lands mid re-inference + // (before the next terminal commit). + function parkedContinuationRunner(continuationStarted: () => void) { + let call = 0; + return async function* ( + opts: InferenceHarnessOptions, + ): AsyncGenerator { + call += 1; + if (call === 1) { + yield { + type: "inference.done", + seq: opts.nextSeq(), + data: { + turn: suspendToolCallTurn, + usage: emptyUsage(), + source: TEST_SOURCE, + }, + }; + return; + } + continuationStarted(); + await new Promise((resolve) => { + const signal = opts.signal; + if (signal === undefined || signal.aborted) { + resolve(); + return; + } + signal.addEventListener("abort", () => resolve(), { once: true }); + }); + }; + } + + function askExtension(approvalTimeoutMs: number) { + return createAuthzExtension({ + authorize: async () => ({ + effect: "ask" as const, + matchingGrants: [], + resolvedBy: null, + }), + approvalTimeoutMs, + }); + } + + // Suspend on the ask gate, deliver `decision`, and abort once the + // continuation inference starts. Returns the parked op's original deadline. + async function resolveThenAbortMidContinuation( + cell: PersistedCell, + decision: (correlationId: string) => InboundMessage, + approvalTimeoutMs: number, + ): Promise { + const continuation = Promise.withResolvers(); + const first = createTestReactor({ + contextStore: makePersistingContextStore(cell), + director: createDefaultDirector("test agent", []), + inferenceRunner: parkedContinuationRunner(() => continuation.resolve()), + toolRunner: makeToolRunner(async (call) => ({ + callId: call.id, + content: "charged", + })), + beforeToolExtensions: [askExtension(approvalTimeoutMs)], + }); + + first.reactor.start(); + first.reactor.deliver(makeInboundMessage()); + + const blocked = await first.waitFor("reactor.gate.blocked"); + if (blocked.type !== "reactor.gate.blocked") throw new Error("unreachable"); + const correlationId = blocked.data.correlationId; + if (correlationId === undefined) throw new Error("expected correlationId"); + const deadline = cell.pendingOperations[0]?.timeoutAt; + if (deadline === undefined) throw new Error("expected a persisted deadline"); + + first.reactor.deliver(decision(correlationId)); + await continuation.promise; + + first.reactor.abort("admin_kill"); + await first.waitFor("reactor.done"); + return deadline; + } + + test("an approved re-dispatch leaves no stale pending operation, and a restart survives the original deadline", async () => { + const cell: PersistedCell = { + turns: [], + pendingOperations: [], + tokenUsage: emptyUsage(), + }; + + const deadline = await resolveThenAbortMidContinuation( + cell, + (correlationId) => makeApprovalMessage(correlationId), + 500, + ); + + // The re-dispatched call's result and the parked op's removal are both on + // disk even though the run never reached a terminal commit. + expect(cell.pendingOperations).toEqual([]); + expect( + cell.turns.some((t) => + t.content.some( + (b) => b.type === "tool_result" && b.callId === "call-ask", + ), + ), + ).toBe(true); + expect(() => assertWellFormedToolSequence(cell.turns)).not.toThrow(); + + // A reactor rebuilt from that store must not re-arm the answered call's + // gate: at the original deadline a stale op would inject a second + // "approval timed out" result for the same call id and the next inference + // would fail the well-formedness tripwire. + const second = createTestReactor({ + contextStore: makePersistingContextStore(cell), + director: createDefaultDirector("test agent", []), + inferenceRunner: makeInferenceRunner({ + type: "done", + turn: makeAssistantTurn("resumed"), + usage: emptyUsage(), + }), + beforeToolExtensions: [askExtension(500)], + }); + second.reactor.start(); + await second.waitFor("reactor.start"); + + // The margin lets a stale gate that fired at the deadline run its + // continuation to completion before the probe message below. + await waitUntil(() => Date.now() > deadline + 50); + second.reactor.deliver(makeInboundMessage()); + const settled = await waitForEvent( + second.events, + (e) => e.type === "connector.reply" || e.type === "reactor.done", + ); + + expect(second.events.filter((e) => e.type === "reactor.error")).toEqual([]); + expect(settled.type).toBe("connector.reply"); + expect(() => assertWellFormedToolSequence(cell.turns)).not.toThrow(); + + second.reactor.abort("admin_kill"); + await second.waitFor("reactor.done"); + }); + + test("a rejected call's error result and pending-operation removal reach disk before the next terminal", async () => { + const cell: PersistedCell = { + turns: [], + pendingOperations: [], + tokenUsage: emptyUsage(), + }; + + await resolveThenAbortMidContinuation( + cell, + (correlationId) => makeApprovalMessage(correlationId, "rejected"), + 60_000, + ); + + expect(cell.pendingOperations).toEqual([]); + expect( + cell.turns.some((t) => + t.content.some( + (b) => + b.type === "tool_result" && + b.callId === "call-ask" && + b.isError === true, + ), + ), + ).toBe(true); + expect(() => assertWellFormedToolSequence(cell.turns)).not.toThrow(); + }); +}); + // --------------------------------------------------------------------------- // Corrupt persisted state surfaces loud at startup // --------------------------------------------------------------------------- diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts index 39f40737a..0387505d2 100644 --- a/vendor/intx-inference/src/reactor.ts +++ b/vendor/intx-inference/src/reactor.ts @@ -785,6 +785,8 @@ export function createReactor(config: ReactorConfig): Reactor { correlations.remove(correlationId); if (stateManager !== null) { stateManager.removePendingOperation(correlationId); + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + pendingOperationsChanged = true; } // The grant is already recorded (synchronously, in // resumePendingOperation) with no await since; enqueue the re-dispatch @@ -808,6 +810,8 @@ export function createReactor(config: ReactorConfig): Reactor { correlations.remove(correlationId); if (stateManager !== null) { stateManager.removePendingOperation(correlationId); + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + pendingOperationsChanged = true; } enqueue({ type: "resume.tool_result", result: dispatch.result }); break; @@ -1224,16 +1228,26 @@ export function createReactor(config: ReactorConfig): Reactor { // agentic turn to one deferred commit. // // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only - if (addToHistory && pendingCompactOutput === null) { - const revision = stateManager?.getTurnsRevision(); - if ( - stateManager !== null && - revision !== undefined && - revision !== lastWrittenTurnsRevision - ) { - await contextStore.writeTurns(stateManager.getTurns()); - lastWrittenTurnsRevision = revision; - } + if (addToHistory) { + await writeCycleTurns(); + } + } + + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + // The turns-only write between terminals. When the turn just appended + // answers a parked call, the pending operation it resolved is gone from + // memory too; metadata goes out with the turns so a rebuild from disk never + // re-arms that call's gate and answers it a second time at the deadline. + // Cycles that leave pending operations untouched still skip metadata. + async function writeCycleTurns(): Promise { + if (stateManager === null || pendingCompactOutput !== null) return; + const revision = stateManager.getTurnsRevision(); + if (revision !== lastWrittenTurnsRevision) { + await contextStore.writeTurns(stateManager.getTurns()); + lastWrittenTurnsRevision = revision; + } + if (pendingOperationsChanged) { + await writeMetadata(); } } @@ -1379,6 +1393,8 @@ export function createReactor(config: ReactorConfig): Reactor { pendingOperations: stateManager.getPendingOperations(), tokenUsage: stateManager.getTokenUsage(), }); + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + pendingOperationsChanged = false; } // ------------------------------------------------------------------------- @@ -1443,6 +1459,8 @@ export function createReactor(config: ReactorConfig): Reactor { correlations.remove(op.correlationId); if (stateManager !== null) { stateManager.removePendingOperation(op.correlationId); + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + pendingOperationsChanged = true; } enqueue({ type: "resume.tool_result", @@ -1650,6 +1668,8 @@ export function createReactor(config: ReactorConfig): Reactor { if (stateManager !== null) { stateManager.appendTurn(createToolResultTurn([event.result])); } + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + await writeCycleTurns(); } let actions; @@ -1869,6 +1889,13 @@ export function createReactor(config: ReactorConfig): Reactor { // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-last-written-turns-revision let lastWrittenTurnsRevision = 0; + // Set when a resolved approval drops its pending operation from memory; + // cleared by the next metadata write. Lets the turns-only write persist + // metadata only for the cycles that need it. + // + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + let pendingOperationsChanged = false; + async function initiateShutdown(): Promise { if (shutdownStarted) return; shutdownStarted = true;