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.

Author: bwachman

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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | 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 <alexander.guy@pm.me> | This ledger (#reactor-ts-after-checkpoint-director-only) | Next vendored sync |

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<InferenceEvent> {
+      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<void>((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<number> {
+    const continuation = Promise.withResolvers<void>();
+    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
 // ---------------------------------------------------------------------------

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<void> {
+    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<void> {
     if (shutdownStarted) return;
     shutdownStarted = true;