diff --git a/e2e/reactor-approval-reload.test.ts b/e2e/reactor-approval-reload.test.ts index 816330907..6a7fc8e38 100644 --- a/e2e/reactor-approval-reload.test.ts +++ b/e2e/reactor-approval-reload.test.ts @@ -181,6 +181,31 @@ function toolResultsFor( ); } +interface RequestMessage { + content: string | { type: string; id?: string; tool_use_id?: string }[]; +} + +/** The block types naming `callId` in the last model request, in order. */ +async function lastRequestBlocksFor( + session: IntegrationSession, + callId: string, +): Promise { + const request = session.harness.scenario.matchedRequests().at(-1); + if (request === undefined) throw new Error("no routed requests"); + const body = JSON.parse( + await (request.clone() as unknown as Request).text(), + ) as { messages: RequestMessage[] }; + return body.messages.flatMap((message) => + typeof message.content === "string" + ? [] + : message.content.flatMap((block) => + block.id === callId || block.tool_use_id === callId + ? [block.type] + : [], + ), + ); +} + function collect(session: IntegrationSession): ReactorEmittedEvent[] { const events: ReactorEmittedEvent[] = []; void (async () => { @@ -495,10 +520,166 @@ describe("integration: approving an ask, then reloading the agent", () => { ); }); -// Reloads at the run's end, where the TUI's deferred reload fires. A reload -// after the decision but before the cycle commits is not covered here: the -// rejected or timed-out result is not yet persisted, so the rebuilt agent -// loads the parked call with no result at all. +// Reloads once the model's next turn has started, before the cycle commits: +// where an interrupt or session rotation lands mid-run. +describe("integration: an ask settled without running, then reloading the agent before the cycle commits", () => { + const continuationStarted = (session: IntegrationSession): boolean => + session.harness.scenario.matchedRequests().length === 2; + + test.serial( + "after a rejection: the rebuilt agent sends the denial once, and the old deadline adds no result", + async () => { + const reloaded = await decideThenReload( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + continuationStarted, + { decision: "reject" }, + ); + const { session, callId } = reloaded; + expect((await session.storage.load()).pendingOperations).toEqual([]); + const settled = toolResultsFor(await session.agent.history(), callId); + expect(settled).toHaveLength(1); + expect(settled.join("")).toContain("denied by approver"); + + await passDeadline(reloaded); + expect(toolResultsFor(await session.agent.history(), callId)).toEqual( + settled, + ); + + session.harness.scenario.replyOnce("anthropic", { text: "Done." }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Done."); + expect(await lastRequestBlocksFor(session, callId)).toEqual([ + "tool_use", + "tool_result", + ]); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + }, + 20_000, + ); + + test.serial( + "after the deadline answers it: the rebuilt agent loads the timeout result and no parked call", + async () => { + const reloaded = await decideThenReload( + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + continuationStarted, + { decision: "expire" }, + ); + const { session, callId } = reloaded; + // Read before the rebuilt agent runs: a parked call it loaded would be + // re-armed at the already-past deadline and answered by the rebuilt + // reactor itself, which would hide the lost result. + expect((await session.storage.load()).pendingOperations).toEqual([]); + const settled = toolResultsFor(await session.agent.history(), callId); + expect(settled).toHaveLength(1); + expect(settled.join("")).toContain("approval timed out"); + + await passDeadline(reloaded, reloaded.deadline + FRESH_WINDOW_PASSED_MS); + expect(toolResultsFor(await session.agent.history(), callId)).toEqual( + settled, + ); + + session.harness.scenario.replyOnce("anthropic", { text: "Done." }); + const followUp = await runUntilDone(session, "Continue."); + expect(followUp.reply).toBe("Done."); + expect(await lastRequestBlocksFor(session, callId)).toEqual([ + "tool_use", + "tool_result", + ]); + }, + 20_000, + ); +}); + +// The loop is busy on a second message's inference when the rejection is +// accepted, so the close's abort is served first and only the shutdown flush +// writes the rejection's result. The reactor leaves the approval record +// behind; the session store must close it on load. +describe("integration: a rejection the reactor accepted but never ran, then reopening the agent", () => { + test.serial( + "the reopened agent loads the denial once and no parked call, and the old deadline adds no result", + async () => { + const clock = holdLongTimers(); + cleanups.push(() => clock.restore()); + const ctx = deferredApprovalGate(); + const sessionOpts = { + permissionGate: ctx.gate, + authorize: createReactorAuthorize(ctx.gate), + pinToolSetAtBuild: true, + }; + const live = { session: await openIntegrationSession(sessionOpts) }; + cleanups.push(() => closeIntegrationSession(live.session)); + live.session.harness.scenario.replyOnce("anthropic", { + toolCalls: [ + { name: "run_shell", args: { command: CLOSED_PORT_PROBE } }, + ], + }); + live.session.harness.scenario.stall(); + + const turn = await runUntilSuspended( + live.session, + "Probe the local port.", + ); + const suspended = turn.result; + if (suspended.type !== "suspended") throw new Error("did not park"); + const parked = (await live.session.storage.load()).pendingOperations.find( + (op) => op.correlationId === suspended.correlationId, + ); + const callId = parked?.suspendedCall?.id; + const deadline = parked?.timeoutAt; + if (callId === undefined || deadline === undefined) { + throw new Error("parked call has no persisted call id or deadline"); + } + + void live.session.agent.send("Still there?").catch(() => undefined); + await waitFor( + () => live.session.harness.scenario.matchedRequests().length === 2, + "second message's inference", + ); + const resume = createApprovalResume({ + getAgent: () => live.session.agent, + gate: ctx.gate, + resolveParkedCallId: (id) => + resolveParkedCallIdFromStore(live.session.storage, id), + }); + const handling = resume.handle(suspended); + await waitFor(ctx.asked, "approval prompt"); + ctx.reject(); + expect(await handling).toBe(true); + await waitFor( + () => turn.events.some((e) => e.type === "message.correlated"), + "rejection accepted", + ); + + live.session = await rebuildIntegrationSession(live.session, sessionOpts); + const { session } = live; + expect((await session.storage.load()).pendingOperations).toEqual([]); + const settled = toolResultsFor(await session.agent.history(), callId); + expect(settled).toHaveLength(1); + expect(settled.join("")).toContain("denied by approver"); + + const reloaded = { + session, + events: collect(session), + callId, + deadline, + clock, + }; + await passDeadline(reloaded, deadline + FRESH_WINDOW_PASSED_MS); + expect(reloaded.events.filter((e) => e.type === "reactor.error")).toEqual( + [], + ); + expect(toolResultsFor(await session.agent.history(), callId)).toEqual( + settled, + ); + }, + 20_000, + ); +}); + +// Reloads at the run's end, where the TUI's deferred reload fires. describe("integration: an ask settled without running, then reloading the agent at the run's end", () => { const runEnded = ( _session: IntegrationSession, diff --git a/vendor/intx-inference/PATCHES.md b/vendor/intx-inference/PATCHES.md index 08030733b..9f2d1bac6 100644 --- a/vendor/intx-inference/PATCHES.md +++ b/vendor/intx-inference/PATCHES.md @@ -508,9 +508,37 @@ 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. +The loop's `resume.tool_result` step (a parked call rejected, timed out or +dropped without running) writes turns the same way, through the shared +`writeTurnsIfChanged` helper `executeTools` also calls, so a rebuild before +the next terminal loads the result. That closes the parked approval only +because Corbits' optimized context store drops, on load, a pending operation +whose call has a `tool_result` in the turn log (`openPendingOperations` in +`src/session/optimized-context-store.ts`). The reactor does not clear the +record itself: a store that returns pending operations from metadata as last +committed, such as `@intx/storage-isogit`, reloads both the result and the +parked approval, and the re-armed gate's deadline adds a second result. +Upstream needs no equivalent while it commits every tool cycle. A staged +compact skips the write in both places; a parked approval cannot coincide +with one, since the suspend terminal commits and clears it. + +`initiateShutdown` appends any `resume.tool_result` still queued behind the +abort (a decision the reactor accepted and removed from its pending set, but +never processed, since the loop serves `abort` first), before +`gates.shutdown()` clears the gate timers. It then writes turns once, only +when it appended something, inside a `try/catch` that reports a non-fatal +`reactor.error`, so a failing store cannot skip `reactor.done` or leak the +workdir lock. Approved calls (`resume.execute_tools`) are not flushed: one +that never ran stays parked on disk rather than gaining a result for a +tool that did not run. A flushed result closes the approval under the same +store dependency as the `resume.tool_result` write. + **Disposition:** Promotion candidate. Performance coalescing — collapses K git commits per agentic turn to one deferred commit while preserving -rebuild-from-disk durability. +rebuild-from-disk durability. The `resume.tool_result` write and the +shutdown flush rely on the store closing an answered call's approval, which +upstream stores do not do, so promoting them also needs the reactor to clear +that pending operation when it writes the result. **Removal path:** Upstream PR replacing the `executeTools` checkpoint with a writeTurns-only segments touch and deferring the cycle commit to cycle terminals. @@ -750,7 +778,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` and the `resume.tool_result` step write turns to store segments only and defer the cycle commit to terminals | 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..d02349893 100644 --- a/vendor/intx-inference/src/reactor.test.ts +++ b/vendor/intx-inference/src/reactor.test.ts @@ -5374,6 +5374,229 @@ describe("createReactor — approval resume error result", () => { await waitFor("reactor.done"); expect(cleared).toBe(true); }); + + // Every inference after the first holds until released, so the run cannot + // reach a terminal (and its cycle commit) while the test reads the store. + function heldContinuationRunner() { + let call = 0; + let markStarted = (): void => {}; + const continuationStarted = new Promise((resolve) => { + markStarted = resolve; + }); + let release = (): void => {}; + const released = new Promise((resolve) => { + release = resolve; + }); + async function* runner( + opts: InferenceHarnessOptions, + ): AsyncGenerator { + call += 1; + if (call > 1) { + markStarted(); + await released; + } + const turn: AssistantTurn = + call === 1 + ? suspendToolCallTurn + : { + role: "assistant", + content: [{ type: "text", text: "acknowledged" }], + model: "test-model", + timestamp: 2000, + }; + yield { + type: "inference.done", + seq: opts.nextSeq(), + data: { turn, usage: emptyUsage(), source: TEST_SOURCE }, + }; + } + return { runner, continuationStarted, release }; + } + + function parkedCallResultTexts(turns: ConversationTurn[]): string[] { + return turns + .flatMap((t) => t.content) + .flatMap((b) => + b.type === "tool_result" && b.callId === "call-ask" + ? [ + b.content + .filter((c) => c.type === "text") + .map((c) => c.text) + .join(""), + ] + : [], + ); + } + + function rejection(correlationId: string): InboundMessage { + return createInboundMessage({ + from: "signal@local", + to: "agent@example.com", + content: JSON.stringify({ outcome: "rejected", message: "too risky" }), + correlationId, + }); + } + + test("a rejected parked call's result reaches the store before the cycle commits", async () => { + const cell: PersistedCell = { + turns: [], + pendingOperations: [], + tokenUsage: emptyUsage(), + }; + const held = heldContinuationRunner(); + const { reactor, events, waitFor } = createTestReactor({ + contextStore: makePersistingContextStore(cell), + director: createDefaultDirector("test agent", []), + inferenceRunner: held.runner, + beforeToolExtensions: [askExtension()], + }); + + reactor.start(); + reactor.deliver(makeInboundMessage()); + const blocked = await 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"); + + reactor.deliver(rejection(correlationId)); + await held.continuationStarted; + + expect(parkedCallResultTexts(cell.turns)).toEqual([ + "denied by approver: too risky", + ]); + + held.release(); + await waitForEvent(events, (e) => e.type === "connector.reply"); + expect(parkedCallResultTexts(cell.turns)).toHaveLength(1); + }); + + test("a timed-out parked call's result reaches the store before the cycle commits", async () => { + const cell: PersistedCell = { + turns: [], + pendingOperations: [], + tokenUsage: emptyUsage(), + }; + const held = heldContinuationRunner(); + const { reactor, events } = createTestReactor({ + contextStore: makePersistingContextStore(cell), + director: createDefaultDirector("test agent", []), + inferenceRunner: held.runner, + beforeToolExtensions: [askExtension(80)], + }); + + reactor.start(); + reactor.deliver(makeInboundMessage()); + await held.continuationStarted; + + expect(parkedCallResultTexts(cell.turns)).toEqual(["approval timed out"]); + + held.release(); + await waitForEvent(events, (e) => e.type === "connector.reply"); + expect(parkedCallResultTexts(cell.turns)).toHaveLength(1); + }); + + // The loop must be busy when the decision is accepted, or it acts on it + // on the next turn of the loop before an abort can be queued. A second + // message's held inference keeps it busy; the abort queued behind that + // inference is dequeued ahead of the accepted decision. + async function decideThenAbortBehindHeldInference( + contextStore: ContextStore, + beforeAbort: () => void = () => {}, + decision: (correlationId: string) => InboundMessage = rejection, + ) { + const held = heldContinuationRunner(); + const handle = createTestReactor({ + contextStore, + director: createDefaultDirector("test agent", []), + inferenceRunner: held.runner, + beforeToolExtensions: [askExtension()], + }); + const { reactor, events, waitFor } = handle; + + reactor.start(); + reactor.deliver(makeInboundMessage()); + const blocked = await 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"); + + reactor.deliver(makeInboundMessage()); + await held.continuationStarted; + reactor.deliver(decision(correlationId)); + await waitForEvent(events, (e) => e.type === "message.correlated"); + + beforeAbort(); + reactor.abort("user_disconnect"); + held.release(); + await waitFor("reactor.done"); + return handle; + } + + test("an accepted rejection still queued at shutdown reaches the store", async () => { + const cell: PersistedCell = { + turns: [], + pendingOperations: [], + tokenUsage: emptyUsage(), + }; + + await decideThenAbortBehindHeldInference(makePersistingContextStore(cell)); + + expect(parkedCallResultTexts(cell.turns)).toEqual([ + "denied by approver: too risky", + ]); + }); + + test("an accepted approval still queued at shutdown stays parked with no result", async () => { + const cell: PersistedCell = { + turns: [], + pendingOperations: [], + tokenUsage: emptyUsage(), + }; + + await decideThenAbortBehindHeldInference( + makePersistingContextStore(cell), + undefined, + (correlationId) => makeApprovalMessage(correlationId), + ); + + // The host handed the approval over and the reactor accepted it, but the + // call never ran: only a rebuilt agent's host can settle it now. + expect(parkedCallResultTexts(cell.turns)).toEqual([]); + expect(cell.pendingOperations.map((op) => op.suspendedCall?.id)).toEqual([ + "call-ask", + ]); + }); + + test("a store that fails the shutdown write is reported without blocking shutdown", async () => { + const cell: PersistedCell = { + turns: [], + pendingOperations: [], + tokenUsage: emptyUsage(), + }; + const persisting = makePersistingContextStore(cell); + let failWrites = false; + const contextStore: ContextStore = { + ...persisting, + async writeTurns(turns) { + if (failWrites) throw new Error("disk full"); + await persisting.writeTurns(turns); + }, + }; + + const { events } = await decideThenAbortBehindHeldInference( + contextStore, + () => { + failWrites = true; + }, + ); + + const errors = events.flatMap((e) => + e.type === "reactor.error" ? [e.data] : [], + ); + expect(errors).toHaveLength(1); + expect(errors[0]?.fatal).toBe(false); + expect(errors[0]?.error).toContain("disk full"); + }); }); // --------------------------------------------------------------------------- diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts index 39f40737a..09410110e 100644 --- a/vendor/intx-inference/src/reactor.ts +++ b/vendor/intx-inference/src/reactor.ts @@ -1224,17 +1224,7 @@ 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 writeTurnsIfChanged(); } async function executeCompact( @@ -1645,11 +1635,20 @@ export function createReactor(config: ReactorConfig): Reactor { // Land it in history before the director decides so the tool_result turn // closes the dangling tool_use and the re-inference the director returns // sees a well-formed sequence. No tool ran, so no tool.done and no - // counter change accompany it. + // counter change accompany it. Write it to disk now, as an approved + // call's tool cycle does: held only in memory until the cycle commits, + // a rebuild in between would reload the call as still parked with no + // result. The written result closes the approval only in a store that + // drops a pending operation whose call is answered on disk, as Corbits' + // does; one that keeps pending operations until the commit reloads the + // approval beside the result. + // + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only if (event.type === "resume.tool_result") { if (stateManager !== null) { stateManager.appendTurn(createToolResultTurn([event.result])); } + await writeTurnsIfChanged(); } let actions; @@ -1869,17 +1868,63 @@ export function createReactor(config: ReactorConfig): Reactor { // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-last-written-turns-revision let lastWrittenTurnsRevision = 0; + // Touches the store's on-disk turn segments only; the full cycle commit + // waits for a terminal. A staged compact already wrote the new generation, + // and memory still holds the old turns, so it is never overwritten here. + // + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + async function writeTurnsIfChanged(): Promise { + if (stateManager === null || pendingCompactOutput !== null) return; + const revision = stateManager.getTurnsRevision(); + if (revision === lastWrittenTurnsRevision) return; + await contextStore.writeTurns(stateManager.getTurns()); + lastWrittenTurnsRevision = revision; + } + async function initiateShutdown(): Promise { if (shutdownStarted) return; shutdownStarted = true; abortOperations(); + + // A rejection or timeout the reactor accepted but had not yet processed + // when the abort overtook it has already left the pending set, so its + // result line is the only trace of the decision, and closes the approval + // on disk only in a store that drops an answered call's pending + // operation, as the resume.tool_result write above relies on. Taken before + // gates.shutdown(), which clears every gate timer, so no timeout can + // enqueue another after it; isLiveApproval refuses decisions from here. + // + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only + const acceptedResults = queue.flatMap((event) => + event.type === "resume.tool_result" ? [event.result] : [], + ); + if (stateManager !== null) { + for (const result of acceptedResults) { + stateManager.appendTurn(createToolResultTurn([result])); + } + } + gates.shutdown(); if (stateManager !== null) { stateManager.setGatesSnapshot([]); } + // Shutdown also runs after fatal errors, where the store may be what + // failed; a throw here must not skip reactor.done or the lock release. + if (acceptedResults.length > 0) { + try { + await writeTurnsIfChanged(); + } catch (cause) { + logger.error`shutdown turn write failed: ${cause}`; + emitError( + `shutdown turn write failed: ${cause instanceof Error ? cause.message : String(cause)}`, + false, + ); + } + } + if (inFlight.size > 0) { const deadline = new Promise((resolve) => setTimeout(resolve, shutdownTimeoutMs),