From f7fc15f280e76e0e6e274ef219fdda5615dea252 Mon Sep 17 00:00:00 2001
From: Ben Wachman <ben@abklabs.com>
Date: Fri, 16 Oct 2026 22:54:00 -0400
Subject: [PATCH 1/5] 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 <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 |
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<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
 // ---------------------------------------------------------------------------
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<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;

From 878315281e10eb0e35cf585246e133eb28652140 Mon Sep 17 00:00:00 2001
From: Ben Wachman <ben@abklabs.com>
Date: Fri, 16 Oct 2026 22:54:56 -0400
Subject: [PATCH 2/5] 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 1d171aea4..4f88e4c8e 100644
--- a/e2e/integration-harness.ts
+++ b/e2e/integration-harness.ts
@@ -157,6 +157,8 @@ export interface OpenIntegrationSessionOpts {
   retryPolicy?: RetryPolicy;
   /** Fields merged over harness.deps, e.g. a real scheduler when retry delays must elapse. */
   depsOverrides?: Partial<Dependencies>;
+  /** Reuse an existing session cwd, so the agent loads that session's store. */
+  cwd?: string;
 }
 
 export async function openIntegrationSession(
@@ -178,7 +180,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,
@@ -412,7 +415,7 @@ export async function openIntegrationSession(
   return session;
 }
 
-export async function closeIntegrationSession(
+async function releaseIntegrationAgent(
   session: IntegrationSession,
 ): Promise<void> {
   try {
@@ -420,10 +423,31 @@ export async function closeIntegrationSession(
   } finally {
     await session.toolset.dispose();
     session.harness.dispose();
+  }
+}
+
+export async function closeIntegrationSession(
+  session: IntegrationSession,
+): Promise<void> {
+  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<IntegrationSession> {
+  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<object, HeldTimer>();
+  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<typeof clearTimeout>[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<void> {
+  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> | 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<typeof holdLongTimers>;
+}
+
+/**
+ * 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<Reloaded> {
+  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<void> {
+  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 4a76bd0fdcb975aa0d16cb326a56510d72664de1 Mon Sep 17 00:00:00 2001
From: Ben Wachman <ben@abklabs.com>
Date: Fri, 16 Oct 2026 23:16:20 -0400
Subject: [PATCH 3/5] 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 <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; 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-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 <alexander.guy@pm.me> | 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 <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 |
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<string> {
+  const answered = new Set<string>();
+  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 667035ac7a77326ab32c43b391b99f08b1e4caac Mon Sep 17 00:00:00 2001
From: Sawyer Cutler <sawyer@dirtroad.dev>
Date: Sat, 17 Oct 2026 09:11:16 -0700
Subject: [PATCH 4/5] 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 b31afd8c8..c8169974e 100644
--- a/src/session/optimized-context-store.test.ts
+++ b/src/session/optimized-context-store.test.ts
@@ -2104,3 +2104,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 283ddca46..ee8f775c4 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<string> {
+  const answered = new Set<string>();
+  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
@@ -1562,7 +1578,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 <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; 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-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 <alexander.guy@pm.me> | 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 <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 |
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<string> {
-  const answered = new Set<string>();
-  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;

From bd224f1a058ae7c3f795b435bd3a85adb9715c93 Mon Sep 17 00:00:00 2001
From: Ben Wachman <ben@abklabs.com>
Date: Fri, 16 Oct 2026 23:19:30 -0400
Subject: [PATCH 5/5] fix(reactor): open a message run when an approval resumes
 a parked call

A run opens only when the loop dequeues message.received, and a
correlated approval never becomes one. A parked call rehydrated by a
new reactor has no run from the process that parked it, so its
approval re-dispatched the call and re-inferred with no run open, and
consumers that treat no open run as idle could rebuild the agent under
the running continuation.

The approval now opens a run keyed on its own message id when none is
open. A resume inside the still-open parking run is unchanged.
---
 vendor/intx-inference/PATCHES.md          |  33 ++++
 vendor/intx-inference/src/reactor.test.ts | 193 ++++++++++++++++++++++
 vendor/intx-inference/src/reactor.ts      |   8 +
 3 files changed, 234 insertions(+)

diff --git a/vendor/intx-inference/PATCHES.md b/vendor/intx-inference/PATCHES.md
index c4a069082..72d40b947 100644
--- a/vendor/intx-inference/PATCHES.md
+++ b/vendor/intx-inference/PATCHES.md
@@ -498,6 +498,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-resume-opens-message-run
+
+`reactor.ts` — A correlated approval that resumes a parked call opens a
+message run when none is open. `tryCorrelate`, after the `redispatch` or
+`error_result` dispatch has consumed the pending operation and enqueued the
+continuation (and before `message.correlated`), calls `openMessageRun` keyed
+on the approval message's `messageId`. The re-dispatched call and its
+continuation then run inside a normal `message.run.started` /
+`message.run.ended` bracket, closed by the usual reply/wait/done terminal or
+a failure. When a run is already open (the live case, where the run that
+parked the call is still open) nothing changes: the resume continues that
+run and no second bracket opens. The `gate-cleared` mode (an async-marker
+operation with no suspended call) is untouched.
+
+Upstream opens a run only when the loop dequeues `message.received`, and a
+correlated approval never becomes one. A pending operation rehydrated by a
+new reactor (restart, `corbits resume`, an agent rebuild) has no run from
+the parking process, so its approval drove tool execution and inference
+with no run open. Consumers that treat "no open run" as idle (the TUI defers
+agent reloads to `message.run.ended`) then saw the session as idle mid-call
+and could rebuild the agent under the running continuation. The
+`message.run.started` doc comment in `@intx/types` still describes the
+dequeue-only trigger; it is left verbatim.
+
+**Disposition:** Promotion candidate. Upstream's run bracket has no owner
+for a resume that outlives the run that parked it.
+**Removal path:** Upstream PR opening a message run for a correlated resume
+when none is open.
+**Re-carry:** new at this patch; sits between the `tryCorrelate` dispatch
+switch and the `message.correlated` emit, inside
+#reactor-ts-atomic-approval-acceptance's claim. Re-verify after upstream
+changes to correlation dispatch or the run-bracket helpers.
 ## reactor-ts-skip-unchanged-history
 
 `reactor.ts` — Skip re-serializing unchanged history on checkpoint.
@@ -728,6 +760,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 <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; 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-resume-opens-message-run | Open a message run when a correlated approval resumes a parked call and no run is open | Alexander Guy <alexander.guy@pm.me> | This ledger (#reactor-ts-resume-opens-message-run) | 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 |
diff --git a/vendor/intx-inference/src/reactor.test.ts b/vendor/intx-inference/src/reactor.test.ts
index e72eb7c50..840262d0a 100644
--- a/vendor/intx-inference/src/reactor.test.ts
+++ b/vendor/intx-inference/src/reactor.test.ts
@@ -6678,6 +6678,199 @@ describe("createReactor — message.run bracket emission", () => {
   });
 });
 
+describe("createReactor — message.run bracket for a resumed parked call", () => {
+  function askExtension() {
+    return createAuthzExtension({
+      authorize: async () => ({
+        effect: "ask" as const,
+        matchingGrants: [],
+        resolvedBy: null,
+      }),
+      approvalTimeoutMs: 60_000,
+    });
+  }
+
+  // What a restarted reactor loads after the original run parked on the ask
+  // gate: the call is in history, its pending operation is on disk, and no
+  // run from the parking process survives.
+  function parkedStore(correlationId: string): PersistedCell {
+    return {
+      turns: [suspendToolCallTurn],
+      pendingOperations: [
+        {
+          correlationId,
+          kind: "approval",
+          registeredAt: Date.now(),
+          gateId: `pending-${correlationId}`,
+          timeoutAt: Date.now() + 60_000,
+          suspendedCall: { id: "call-ask", name: "charge_card", arguments: {} },
+        },
+      ],
+      tokenUsage: emptyUsage(),
+    };
+  }
+
+  function resumeDirector(): ReactorDirector {
+    return directorFromTable(
+      {
+        "resume.execute_tools": (e, _s, caps) =>
+          caps.executeTools(e.calls, false, true),
+        "resume.tool_result": (_e, _s, caps) => caps.infer(),
+        "tool.done": (_e, _s, caps) => caps.infer(),
+        "inference.done": (_e, _s, caps) => caps.reply("resumed"),
+      },
+      "wait",
+    );
+  }
+
+  function runEvents(events: ReactorEmittedEvent[]) {
+    const started = events.filter((e) => e.type === "message.run.started");
+    const ended = events.filter((e) => e.type === "message.run.ended");
+    return { started, ended };
+  }
+
+  test("an approval for a rehydrated call opens a run around the re-dispatch and its continuation", async () => {
+    const correlationId = "rehydrated-run-approved";
+    const { reactor, events, waitFor } = createTestReactor({
+      contextStore: makePersistingContextStore(parkedStore(correlationId)),
+      director: resumeDirector(),
+      inferenceRunner: makeInferenceRunner({
+        type: "done",
+        turn: makeAssistantTurn("resumed"),
+        usage: emptyUsage(),
+      }),
+      toolRunner: makeToolRunner(async (call) => ({
+        callId: call.id,
+        content: "charged",
+      })),
+      beforeToolExtensions: [askExtension()],
+    });
+    reactor.start();
+    await waitFor("reactor.start");
+
+    const approval = makeApprovalMessage(correlationId);
+    reactor.deliver(approval);
+    await waitFor("connector.reply");
+
+    const { started, ended } = runEvents(events);
+    expect(started).toHaveLength(1);
+    expect(ended).toHaveLength(1);
+    const run = getEvent(events, "message.run.started");
+    const end = getEvent(events, "message.run.ended");
+    expect(run.data.messageId).toBe(approval.headers.messageId);
+    expect(end.data.messageRunId).toBe(run.data.messageRunId);
+    expect(end.data.status).toBe("completed");
+
+    const types = events.map((e) => e.type);
+    expect(types.indexOf("message.run.started")).toBeLessThan(
+      types.indexOf("tool.start"),
+    );
+    expect(types.indexOf("message.run.ended")).toBeGreaterThan(
+      types.indexOf("connector.reply"),
+    );
+
+    reactor.abort("admin_kill");
+    await waitFor("reactor.done");
+  });
+
+  test("a rejection for a rehydrated call opens a run around its continuation", async () => {
+    const correlationId = "rehydrated-run-rejected";
+    const { reactor, events, waitFor } = createTestReactor({
+      contextStore: makePersistingContextStore(parkedStore(correlationId)),
+      director: resumeDirector(),
+      inferenceRunner: makeInferenceRunner({
+        type: "done",
+        turn: makeAssistantTurn("resumed"),
+        usage: emptyUsage(),
+      }),
+      beforeToolExtensions: [askExtension()],
+    });
+    reactor.start();
+    await waitFor("reactor.start");
+
+    reactor.deliver(makeApprovalMessage(correlationId, "rejected"));
+    await waitFor("connector.reply");
+
+    const { started, ended } = runEvents(events);
+    expect(started).toHaveLength(1);
+    expect(ended).toHaveLength(1);
+    const types = events.map((e) => e.type);
+    expect(types.indexOf("message.run.started")).toBeLessThan(
+      types.indexOf("inference.done"),
+    );
+    expect(types.indexOf("message.run.ended")).toBeGreaterThan(
+      types.indexOf("connector.reply"),
+    );
+
+    reactor.abort("admin_kill");
+    await waitFor("reactor.done");
+  });
+
+  test("an approval for a call parked by the open run opens no second run", async () => {
+    let inferences = 0;
+    const inferenceRunner = async function* (
+      opts: InferenceHarnessOptions,
+    ): AsyncGenerator<InferenceEvent> {
+      inferences += 1;
+      yield {
+        type: "inference.done",
+        seq: opts.nextSeq(),
+        data: {
+          turn: inferences === 1 ? suspendToolCallTurn : makeAssistantTurn("ok"),
+          usage: emptyUsage(),
+          source: TEST_SOURCE,
+        },
+      };
+    };
+    const { reactor, events, waitFor } = createTestReactor({
+      director: directorFromTable(
+        {
+          "message.received": (_e, _s, caps) => caps.infer(),
+          "inference.done": (_e, _s, caps) =>
+            inferences === 1
+              ? caps.executeTools([
+                  { id: "call-ask", name: "charge_card", arguments: {} },
+                ])
+              : caps.reply("done"),
+          "resume.execute_tools": (e, _s, caps) =>
+            caps.executeTools(e.calls, false, true),
+          "tool.done": (_e, _s, caps) => caps.infer(),
+        },
+        "wait",
+      ),
+      inferenceRunner,
+      toolRunner: makeToolRunner(async (call) => ({
+        callId: call.id,
+        content: "charged",
+      })),
+      beforeToolExtensions: [askExtension()],
+    });
+    reactor.start();
+    const inbound = makeInboundMessage();
+    reactor.deliver(inbound);
+    const blocked = await waitFor("reactor.gate.blocked");
+    if (blocked.type !== "reactor.gate.blocked") throw new Error("unreachable");
+    const correlationId = blocked.data.correlationId;
+    if (correlationId === undefined) throw new Error("expected correlationId");
+
+    reactor.deliver(makeApprovalMessage(correlationId));
+    await waitFor("message.run.ended");
+
+    const { started, ended } = runEvents(events);
+    expect(started).toHaveLength(1);
+    expect(ended).toHaveLength(1);
+    const run = getEvent(events, "message.run.started");
+    expect(run.data.messageId).toBe(inbound.headers.messageId);
+    const types = events.map((e) => e.type);
+    expect(types.indexOf("message.run.ended")).toBeGreaterThan(
+      types.indexOf("connector.reply"),
+    );
+
+    reactor.abort("admin_kill");
+    await waitFor("reactor.done");
+  });
+});
+
 describe("createReactor — source failover", () => {
   // Simulates a priority-ordered source list the reactor fails over through.
   // `resultFor(id, attemptOnThisSource)` decides what the inference runner
diff --git a/vendor/intx-inference/src/reactor.ts b/vendor/intx-inference/src/reactor.ts
index 0387505d2..f59ec1a90 100644
--- a/vendor/intx-inference/src/reactor.ts
+++ b/vendor/intx-inference/src/reactor.ts
@@ -834,6 +834,14 @@ export function createReactor(config: ReactorConfig): Reactor {
           break;
         }
       }
+      // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-resume-opens-message-run
+      // A parked call normally resumes inside the run that parked it. One
+      // rehydrated from the store after a restart has no run open, so the
+      // approval opens one for the re-dispatch and its continuation; the
+      // usual reply/wait/done terminal or failure closes it.
+      if (dispatch.mode !== "gate-cleared" && currentMessageRunId === null) {
+        openMessageRun(message.headers.messageId);
+      }
       // Locally patched — see vendor/intx-inference/PATCHES.md#reactor-ts-atomic-approval-acceptance
       // Observers may reenter: publish only after consumption and continuation
       // enqueue, while the claim still excludes a duplicate delivery.