From aad53cf3f410eed16d6d8d5f004a19490a942b88 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 16 Oct 2026 22:54:00 -0400 Subject: [PATCH 1/4] fix(reactor): persist resolved approval removal with its result turn Tool cycles write turns only and defer metadata to the next terminal commit. An approved or rejected call drops its pending operation from memory, but a teardown before that terminal left it in metadata.json; a reactor rebuilt from the store re-armed the answered call's gate and, at the original deadline, appended a second tool_result for the same call id, which the prompt tripwire rejects. The turns-only write now also writes metadata when a resolved approval changed the pending operations, so the removal lands with the turn that answers the call. Writing metadata in tryCorrelate instead would drop the op before the re-dispatched result is on disk, leaving a dangling tool_use with no gate to answer it. Cycles that leave pending operations untouched still skip the metadata write. --- vendor/intx-inference/PATCHES.md | 19 ++- vendor/intx-inference/src/reactor.test.ts | 180 +++++++++++++++++++++- vendor/intx-inference/src/reactor.ts | 47 ++++-- 3 files changed, 234 insertions(+), 12 deletions(-) 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; From 19ce192cfb03b1b2c739bc52bf3e063d93773ab1 Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 16 Oct 2026 22:54:56 -0400 Subject: [PATCH 2/4] test(e2e): cover reloading the agent after an approved ask --- e2e/integration-harness.ts | 28 ++- e2e/reactor-approval-reload.test.ts | 280 ++++++++++++++++++++++++++++ 2 files changed, 306 insertions(+), 2 deletions(-) create mode 100644 e2e/reactor-approval-reload.test.ts 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, + ); +}); From 0278479e26dab55f58c3d2b3047741e93f4547ce Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 16 Oct 2026 23:16:20 -0400 Subject: [PATCH 3/4] fix(reactor): never re-answer a parked call that already has a result The pending-operation removal reaches metadata in a second await after the result turn, so a crash between the two leaves a stale op for a call history already answers. A reactor rebuilt from that store re-armed its gate and, at the original deadline, appended a second tool_result for the same call id, which the prompt tripwire rejects. Startup now drops and persists the removal of any loaded op whose parked call already has a result, and the approval timeout injects no synthetic result for such a call. --- vendor/intx-inference/PATCHES.md | 36 ++++++- vendor/intx-inference/src/reactor.test.ts | 121 +++++++++++++++++++++- vendor/intx-inference/src/reactor.ts | 46 +++++++- 3 files changed, 199 insertions(+), 4 deletions(-) diff --git a/vendor/intx-inference/PATCHES.md b/vendor/intx-inference/PATCHES.md index c4a069082..adbbfc664 100644 --- a/vendor/intx-inference/PATCHES.md +++ b/vendor/intx-inference/PATCHES.md @@ -484,7 +484,8 @@ 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. +that narrower window is #reactor-ts-skip-answered-parked-call's concern, not +this patch's. **Disposition:** Promotion candidate. Performance coalescing — collapses K git commits per agentic turn to one deferred commit while preserving @@ -498,6 +499,38 @@ batch committed); the turns still reach disk segments, so only the cycle's manifest records and metadata staging stay uncommitted. Intentionally not committed; re-verify after upstream doom-loop accounting changes. +## reactor-ts-skip-answered-parked-call + +`reactor.ts` — A parked call that history already answers is never answered +again. `start()` drops every loaded pending operation whose +`suspendedCall.id` already has a `tool_result` in the loaded turns: the +state manager never holds it, `rehydrateGates` never re-arms it, and the +removal is written to metadata before `reactor.start`. `onGateCleared`'s +timeout path still removes the op but, when the parked call already has a +result, enqueues no synthetic "approval timed out" result (and no +`reactor.gate.cleared`) and logs a warning instead. Both checks scan the +turns through the module-level `answeredCallIds` helper. + +Defense in depth behind the pending-operation durability contract in +#reactor-ts-tool-cycle-turns-only: that patch writes the result turn and then +metadata in two awaits, so a crash between them leaves a stale op for an +answered call. Re-armed, it fired at the original deadline, appended a second +`tool_result` for the same call id, and the next inference failed the +well-formedness tripwire (`Malformed tool sequence: duplicate tool_result`), +killing the session. The timeout guard leaves `pendingOperationsChanged` set +rather than writing metadata from the timer callback; the next turns-only +write or terminal persists the removal, and a crash before then is caught by +the startup check. + +**Disposition:** Promotion candidate. Upstream's rehydration and timeout +paths trust persisted pending operations without consulting history. +**Removal path:** Upstream PR making rehydration and the approval timeout +skip calls that already have a result. +**Re-carry:** new at this patch; sits in `start()` around +`createStateManager`/`rehydrateGates` and inside the `onGateCleared` timeout +branch. Re-verify after upstream changes to rehydration or to the ask-rail +timeout result. + ## reactor-ts-skip-unchanged-history `reactor.ts` — Skip re-serializing unchanged history on checkpoint. @@ -728,6 +761,7 @@ revisit point is the next vendored sync (see `docs/VENDORING.md`). | 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; 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-answered-parked-call | Drop pending operations whose parked call already has a result at rehydration, and inject no approval-timeout result for such a call | Alexander Guy | This ledger (#reactor-ts-skip-answered-parked-call) | 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 e72eb7c50..fc43f0e86 100644 --- a/vendor/intx-inference/src/reactor.test.ts +++ b/vendor/intx-inference/src/reactor.test.ts @@ -7,7 +7,7 @@ import { createCorrelationRegistry } from "./correlation"; import { createReactor, mergeInferenceOptions } from "./reactor"; import { createDefaultDependencies } from "./providers"; import { createDefaultDirector } from "./default-director"; -import { assertWellFormedToolSequence } from "./turns"; +import { assertWellFormedToolSequence, createToolResultTurn } from "./turns"; import { createInboundMessage } from "@intx/mime"; import { waitUntil } from "@intx/types/testing"; import { PerCallInferenceOptions } from "@intx/types/runtime"; @@ -5554,6 +5554,125 @@ describe("createReactor — resolved approval persists pending-operation removal }); }); +// --------------------------------------------------------------------------- +// A parked call that already has a result is never answered again +// --------------------------------------------------------------------------- + +describe("createReactor — answered parked call is not re-answered", () => { + function parkedOp(correlationId: string, timeoutAt: number): PendingOperation { + return { + correlationId, + kind: "approval", + registeredAt: Date.now(), + gateId: `pending-${correlationId}`, + timeoutAt, + suspendedCall: { id: "call-ask", name: "charge_card", arguments: {} }, + }; + } + + function resultsFor(turns: ConversationTurn[], callId: string): number { + return turns + .flatMap((t) => t.content) + .filter((b) => b.type === "tool_result" && b.callId === callId).length; + } + + test("startup drops and persists the removal of a pending operation whose call already has a result", async () => { + // The state a crash between the result-turn write and the metadata write + // leaves behind: the parked call is answered on disk, but its pending + // operation was never removed. + const deadline = Date.now() + 200; + const cell: PersistedCell = { + turns: [ + suspendToolCallTurn, + createToolResultTurn([{ callId: "call-ask", content: "charged" }]), + ], + pendingOperations: [parkedOp("stale-corr", deadline)], + tokenUsage: emptyUsage(), + }; + + const { reactor, events, waitFor } = createTestReactor({ + contextStore: makePersistingContextStore(cell), + director: createDefaultDirector("test agent", []), + inferenceRunner: makeInferenceRunner({ + type: "done", + turn: makeAssistantTurn("resumed"), + usage: emptyUsage(), + }), + }); + reactor.start(); + await waitFor("reactor.start"); + + expect(cell.pendingOperations).toEqual([]); + + // A re-armed gate would fire at the deadline and append "approval timed + // out" for the answered call; the next inference would then fail the + // well-formedness tripwire. + await waitUntil(() => Date.now() > deadline + 50); + reactor.deliver(makeInboundMessage()); + const settled = await waitForEvent( + events, + (e) => e.type === "connector.reply" || e.type === "reactor.done", + ); + + expect(events.filter((e) => e.type === "reactor.error")).toEqual([]); + expect(settled.type).toBe("connector.reply"); + expect(resultsFor(cell.turns, "call-ask")).toBe(1); + + reactor.abort("admin_kill"); + await waitFor("reactor.done"); + }); + + test("a gate timeout for a parked call that already has a result injects no second result", async () => { + // The op is live at startup (its call is unanswered), then the call is + // answered by a path other than its approval, so only the timeout guard + // stands between the deadline and a duplicate tool_result. + const deadline = Date.now() + 200; + const cell: PersistedCell = { + turns: [suspendToolCallTurn], + pendingOperations: [parkedOp("answered-corr", deadline)], + tokenUsage: emptyUsage(), + }; + + let messages = 0; + const { reactor, events, waitFor } = createTestReactor({ + contextStore: makePersistingContextStore(cell), + director: directorFromTable( + { + "message.received": (_e, _s, caps) => { + messages += 1; + return messages === 1 + ? caps.executeTools([ + { id: "call-ask", name: "charge_card", arguments: {} }, + ]) + : caps.reply("ok"); + }, + }, + "wait", + ), + toolRunner: makeToolRunner(async (call) => ({ + callId: call.id, + content: "charged", + })), + }); + reactor.start(); + await waitFor("reactor.start"); + + reactor.deliver(makeInboundMessage()); + await waitFor("tool.done"); + expect(resultsFor(cell.turns, "call-ask")).toBe(1); + + await waitUntil(() => Date.now() > deadline + 50); + reactor.deliver(makeInboundMessage()); + await waitFor("connector.reply"); + + expect(resultsFor(cell.turns, "call-ask")).toBe(1); + expect(() => assertWellFormedToolSequence(cell.turns)).not.toThrow(); + + reactor.abort("admin_kill"); + await waitFor("reactor.done"); + }); +}); + // --------------------------------------------------------------------------- // Corrupt persisted state surfaces loud at startup // --------------------------------------------------------------------------- diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts index 0387505d2..13042206e 100644 --- a/vendor/intx-inference/src/reactor.ts +++ b/vendor/intx-inference/src/reactor.ts @@ -329,6 +329,17 @@ function isTrivialNoopShellCommand(command: unknown): boolean { return TRIVIAL_NOOP_SHELL_PROGRAMS.has(unquoted); } +// Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-skip-answered-parked-call +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; +} + /** * Creates a reactor instance bound to the given configuration. * Call `start()` to begin the event loop. @@ -1462,6 +1473,17 @@ export function createReactor(config: ReactorConfig): Reactor { // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only pendingOperationsChanged = true; } + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-skip-answered-parked-call + // A call that already has a result was resumed some other way; a + // second result for the same id is a malformed tool sequence that + // fails the next inference, so the timeout drives no continuation. + if ( + stateManager !== null && + answeredCallIds(stateManager.getTurns()).has(op.suspendedCall.id) + ) { + logger.warn`Skipped approval timeout result for ${op.suspendedCall.id}: the parked call already has a result`; + return; + } enqueue({ type: "resume.tool_result", result: { @@ -1964,10 +1986,25 @@ export function createReactor(config: ReactorConfig): Reactor { return; } + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-skip-answered-parked-call + // A crash between the result-turn write and the metadata write leaves a + // pending operation for a call history already answers. Re-arming it + // would answer the call again at its deadline, so it is dropped here + // and its removal persisted before the reactor starts. + const answered = answeredCallIds(initialTurns); + const answeredOps = initialOps.filter( + (op) => + op.suspendedCall !== undefined && answered.has(op.suspendedCall.id), + ); + for (const op of answeredOps) { + logger.warn`Dropped pending operation ${op.correlationId}: its parked call ${op.suspendedCall?.id ?? ""} already has a result`; + } + const liveOps = initialOps.filter((op) => !answeredOps.includes(op)); + stateManager = createStateManager( sessionId, initialTurns, - initialOps, + liveOps, initialUsage, ); @@ -1986,10 +2023,15 @@ export function createReactor(config: ReactorConfig): Reactor { // correlationId or gateId. A throw must surface as reactor.error plus // reactor.done (matching the load-failure path), not brick the reactor // as a silent unhandled rejection. - rehydrateGates(initialOps); + rehydrateGates(liveOps); stateManager.setGatesSnapshot(gates.snapshot()); + // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-skip-answered-parked-call + if (answeredOps.length > 0) { + await writeMetadata(); + } + emit({ type: "reactor.start", seq: nextSeq(), data: {} }); const bufferedDeliveries = startupDeliveries; From 0bffe4fc10bbf9c51390c703c1c3d0e9af605206 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sat, 17 Oct 2026 09:11:16 -0700 Subject: [PATCH 4/4] fix(session): drop answered pending operations on load Crash recovery could reload a pending operation for a call the turns already answer, re-arming its gate and injecting a second tool result at the deadline. load() now drops any pending operation whose suspended call id already has a tool result, warning per drop; marker operations without a suspended call pass through. The removal converges to disk on the next metadata write. --- src/session/optimized-context-store.test.ts | 219 ++++++++++++++++++++ src/session/optimized-context-store.ts | 46 +++- vendor/intx-inference/PATCHES.md | 36 +--- vendor/intx-inference/src/reactor.test.ts | 121 +---------- vendor/intx-inference/src/reactor.ts | 46 +--- 5 files changed, 268 insertions(+), 200 deletions(-) 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 adbbfc664..c4a069082 100644 --- a/vendor/intx-inference/PATCHES.md +++ b/vendor/intx-inference/PATCHES.md @@ -484,8 +484,7 @@ 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 #reactor-ts-skip-answered-parked-call's concern, not -this patch's. +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 @@ -499,38 +498,6 @@ batch committed); the turns still reach disk segments, so only the cycle's manifest records and metadata staging stay uncommitted. Intentionally not committed; re-verify after upstream doom-loop accounting changes. -## reactor-ts-skip-answered-parked-call - -`reactor.ts` — A parked call that history already answers is never answered -again. `start()` drops every loaded pending operation whose -`suspendedCall.id` already has a `tool_result` in the loaded turns: the -state manager never holds it, `rehydrateGates` never re-arms it, and the -removal is written to metadata before `reactor.start`. `onGateCleared`'s -timeout path still removes the op but, when the parked call already has a -result, enqueues no synthetic "approval timed out" result (and no -`reactor.gate.cleared`) and logs a warning instead. Both checks scan the -turns through the module-level `answeredCallIds` helper. - -Defense in depth behind the pending-operation durability contract in -#reactor-ts-tool-cycle-turns-only: that patch writes the result turn and then -metadata in two awaits, so a crash between them leaves a stale op for an -answered call. Re-armed, it fired at the original deadline, appended a second -`tool_result` for the same call id, and the next inference failed the -well-formedness tripwire (`Malformed tool sequence: duplicate tool_result`), -killing the session. The timeout guard leaves `pendingOperationsChanged` set -rather than writing metadata from the timer callback; the next turns-only -write or terminal persists the removal, and a crash before then is caught by -the startup check. - -**Disposition:** Promotion candidate. Upstream's rehydration and timeout -paths trust persisted pending operations without consulting history. -**Removal path:** Upstream PR making rehydration and the approval timeout -skip calls that already have a result. -**Re-carry:** new at this patch; sits in `start()` around -`createStateManager`/`rehydrateGates` and inside the `onGateCleared` timeout -branch. Re-verify after upstream changes to rehydration or to the ask-rail -timeout result. - ## reactor-ts-skip-unchanged-history `reactor.ts` — Skip re-serializing unchanged history on checkpoint. @@ -761,7 +728,6 @@ revisit point is the next vendored sync (see `docs/VENDORING.md`). | 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; 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-answered-parked-call | Drop pending operations whose parked call already has a result at rehydration, and inject no approval-timeout result for such a call | Alexander Guy | This ledger (#reactor-ts-skip-answered-parked-call) | 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 fc43f0e86..e72eb7c50 100644 --- a/vendor/intx-inference/src/reactor.test.ts +++ b/vendor/intx-inference/src/reactor.test.ts @@ -7,7 +7,7 @@ import { createCorrelationRegistry } from "./correlation"; import { createReactor, mergeInferenceOptions } from "./reactor"; import { createDefaultDependencies } from "./providers"; import { createDefaultDirector } from "./default-director"; -import { assertWellFormedToolSequence, createToolResultTurn } from "./turns"; +import { assertWellFormedToolSequence } from "./turns"; import { createInboundMessage } from "@intx/mime"; import { waitUntil } from "@intx/types/testing"; import { PerCallInferenceOptions } from "@intx/types/runtime"; @@ -5554,125 +5554,6 @@ describe("createReactor — resolved approval persists pending-operation removal }); }); -// --------------------------------------------------------------------------- -// A parked call that already has a result is never answered again -// --------------------------------------------------------------------------- - -describe("createReactor — answered parked call is not re-answered", () => { - function parkedOp(correlationId: string, timeoutAt: number): PendingOperation { - return { - correlationId, - kind: "approval", - registeredAt: Date.now(), - gateId: `pending-${correlationId}`, - timeoutAt, - suspendedCall: { id: "call-ask", name: "charge_card", arguments: {} }, - }; - } - - function resultsFor(turns: ConversationTurn[], callId: string): number { - return turns - .flatMap((t) => t.content) - .filter((b) => b.type === "tool_result" && b.callId === callId).length; - } - - test("startup drops and persists the removal of a pending operation whose call already has a result", async () => { - // The state a crash between the result-turn write and the metadata write - // leaves behind: the parked call is answered on disk, but its pending - // operation was never removed. - const deadline = Date.now() + 200; - const cell: PersistedCell = { - turns: [ - suspendToolCallTurn, - createToolResultTurn([{ callId: "call-ask", content: "charged" }]), - ], - pendingOperations: [parkedOp("stale-corr", deadline)], - tokenUsage: emptyUsage(), - }; - - const { reactor, events, waitFor } = createTestReactor({ - contextStore: makePersistingContextStore(cell), - director: createDefaultDirector("test agent", []), - inferenceRunner: makeInferenceRunner({ - type: "done", - turn: makeAssistantTurn("resumed"), - usage: emptyUsage(), - }), - }); - reactor.start(); - await waitFor("reactor.start"); - - expect(cell.pendingOperations).toEqual([]); - - // A re-armed gate would fire at the deadline and append "approval timed - // out" for the answered call; the next inference would then fail the - // well-formedness tripwire. - await waitUntil(() => Date.now() > deadline + 50); - reactor.deliver(makeInboundMessage()); - const settled = await waitForEvent( - events, - (e) => e.type === "connector.reply" || e.type === "reactor.done", - ); - - expect(events.filter((e) => e.type === "reactor.error")).toEqual([]); - expect(settled.type).toBe("connector.reply"); - expect(resultsFor(cell.turns, "call-ask")).toBe(1); - - reactor.abort("admin_kill"); - await waitFor("reactor.done"); - }); - - test("a gate timeout for a parked call that already has a result injects no second result", async () => { - // The op is live at startup (its call is unanswered), then the call is - // answered by a path other than its approval, so only the timeout guard - // stands between the deadline and a duplicate tool_result. - const deadline = Date.now() + 200; - const cell: PersistedCell = { - turns: [suspendToolCallTurn], - pendingOperations: [parkedOp("answered-corr", deadline)], - tokenUsage: emptyUsage(), - }; - - let messages = 0; - const { reactor, events, waitFor } = createTestReactor({ - contextStore: makePersistingContextStore(cell), - director: directorFromTable( - { - "message.received": (_e, _s, caps) => { - messages += 1; - return messages === 1 - ? caps.executeTools([ - { id: "call-ask", name: "charge_card", arguments: {} }, - ]) - : caps.reply("ok"); - }, - }, - "wait", - ), - toolRunner: makeToolRunner(async (call) => ({ - callId: call.id, - content: "charged", - })), - }); - reactor.start(); - await waitFor("reactor.start"); - - reactor.deliver(makeInboundMessage()); - await waitFor("tool.done"); - expect(resultsFor(cell.turns, "call-ask")).toBe(1); - - await waitUntil(() => Date.now() > deadline + 50); - reactor.deliver(makeInboundMessage()); - await waitFor("connector.reply"); - - expect(resultsFor(cell.turns, "call-ask")).toBe(1); - expect(() => assertWellFormedToolSequence(cell.turns)).not.toThrow(); - - reactor.abort("admin_kill"); - await waitFor("reactor.done"); - }); -}); - // --------------------------------------------------------------------------- // Corrupt persisted state surfaces loud at startup // --------------------------------------------------------------------------- diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts index 13042206e..0387505d2 100644 --- a/vendor/intx-inference/src/reactor.ts +++ b/vendor/intx-inference/src/reactor.ts @@ -329,17 +329,6 @@ function isTrivialNoopShellCommand(command: unknown): boolean { return TRIVIAL_NOOP_SHELL_PROGRAMS.has(unquoted); } -// Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-skip-answered-parked-call -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; -} - /** * Creates a reactor instance bound to the given configuration. * Call `start()` to begin the event loop. @@ -1473,17 +1462,6 @@ export function createReactor(config: ReactorConfig): Reactor { // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-tool-cycle-turns-only pendingOperationsChanged = true; } - // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-skip-answered-parked-call - // A call that already has a result was resumed some other way; a - // second result for the same id is a malformed tool sequence that - // fails the next inference, so the timeout drives no continuation. - if ( - stateManager !== null && - answeredCallIds(stateManager.getTurns()).has(op.suspendedCall.id) - ) { - logger.warn`Skipped approval timeout result for ${op.suspendedCall.id}: the parked call already has a result`; - return; - } enqueue({ type: "resume.tool_result", result: { @@ -1986,25 +1964,10 @@ export function createReactor(config: ReactorConfig): Reactor { return; } - // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-skip-answered-parked-call - // A crash between the result-turn write and the metadata write leaves a - // pending operation for a call history already answers. Re-arming it - // would answer the call again at its deadline, so it is dropped here - // and its removal persisted before the reactor starts. - const answered = answeredCallIds(initialTurns); - const answeredOps = initialOps.filter( - (op) => - op.suspendedCall !== undefined && answered.has(op.suspendedCall.id), - ); - for (const op of answeredOps) { - logger.warn`Dropped pending operation ${op.correlationId}: its parked call ${op.suspendedCall?.id ?? ""} already has a result`; - } - const liveOps = initialOps.filter((op) => !answeredOps.includes(op)); - stateManager = createStateManager( sessionId, initialTurns, - liveOps, + initialOps, initialUsage, ); @@ -2023,15 +1986,10 @@ export function createReactor(config: ReactorConfig): Reactor { // correlationId or gateId. A throw must surface as reactor.error plus // reactor.done (matching the load-failure path), not brick the reactor // as a silent unhandled rejection. - rehydrateGates(liveOps); + rehydrateGates(initialOps); stateManager.setGatesSnapshot(gates.snapshot()); - // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-skip-answered-parked-call - if (answeredOps.length > 0) { - await writeMetadata(); - } - emit({ type: "reactor.start", seq: nextSeq(), data: {} }); const bufferedDeliveries = startupDeliveries;