From 0278479e26dab55f58c3d2b3047741e93f4547ce Mon Sep 17 00:00:00 2001 From: Ben Wachman Date: Fri, 16 Oct 2026 23:16:20 -0400 Subject: [PATCH 1/2] 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 2/2] 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;