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.

Author: TheGreatAxios

src/session/optimized-context-store.test.ts

@@ -2268,3 +2268,222 @@ describe("createOptimizedContextStore metadata ownership (CL-10667)", () => {
     expect((await resumed.load()).tokenUsage).toEqual(USAGE_TWO);
   });
 });
+
+describe("createOptimizedContextStore load drops pending operations for answered calls", () => {
+  function callTurn(callId: string, timestamp: number): ConversationTurn {
+    return {
+      role: "assistant",
+      content: [
+        { type: "tool_call", id: callId, name: "run_shell", arguments: {} },
+      ],
+      timestamp,
+    };
+  }
+
+  function resultTurn(
+    callId: string,
+    text: string,
+    timestamp: number,
+  ): ConversationTurn {
+    return {
+      role: "user",
+      content: [
+        { type: "tool_result", callId, content: [{ type: "text", text }] },
+      ],
+      timestamp,
+    };
+  }
+
+  function approvalOp(correlationId: string, callId: string) {
+    return {
+      correlationId,
+      kind: "approval" as const,
+      registeredAt: 1_700_000_000_000,
+      gateId: `pending-${correlationId}`,
+      timeoutAt: Date.now() + 60_000,
+      suspendedCall: { id: callId, name: "run_shell", arguments: {} },
+    };
+  }
+
+  function markerOp(correlationId: string) {
+    return {
+      correlationId,
+      kind: "approval" as const,
+      registeredAt: 1_700_000_000_000,
+      gateId: `pending-${correlationId}`,
+    };
+  }
+
+  function placeCrashState(
+    dir: string,
+    turns: ConversationTurn[],
+    pendingOperations: unknown[],
+  ): void {
+    fs.writeFileSync(path.join(dir, TURNS_FILE), jsonl(turns));
+    fs.writeFileSync(
+      path.join(dir, "metadata.json"),
+      JSON.stringify({ pendingOperations, connectorState: null }, null, 2),
+    );
+  }
+
+  // The crash window between the result-turn write and the metadata op-removal
+  // write leaves turns answering a parked call while metadata.json still lists
+  // its pending operation. load() is the only path into the rebuilt
+  // reactor's rehydrateGates, so the stale operation must die here.
+  test("drops an answered operation while keeping unanswered and marker operations", async () => {
+    const dir = tempDir();
+    placeCrashState(
+      dir,
+      [callTurn("call-answered", 1), resultTurn("call-answered", "ok", 2)],
+      [
+        approvalOp("stale-corr", "call-answered"),
+        approvalOp("live-corr", "call-live"),
+        markerOp("marker-corr"),
+      ],
+    );
+
+    const store = await createOptimizedContextStore(dir);
+    const loaded = await store.load();
+    expect(
+      loaded.pendingOperations.map((op) => op.correlationId).sort(),
+    ).toEqual(["live-corr", "marker-corr"]);
+  });
+
+  test("repeated loads are stable and emit no metadata write", async () => {
+    const dir = tempDir();
+    placeCrashState(
+      dir,
+      [callTurn("call-answered", 1), resultTurn("call-answered", "ok", 2)],
+      [approvalOp("stale-corr", "call-answered"), markerOp("marker-corr")],
+    );
+
+    const store = await createOptimizedContextStore(dir);
+    const before = fs.readFileSync(path.join(dir, "metadata.json"), "utf8");
+    const first = await store.load();
+    const second = await store.load();
+    expect(second.pendingOperations).toEqual(first.pendingOperations);
+    expect(first.pendingOperations.map((op) => op.correlationId)).toEqual([
+      "marker-corr",
+    ]);
+    expect(fs.readFileSync(path.join(dir, "metadata.json"), "utf8")).toBe(
+      before,
+    );
+  });
+
+  test("the next metadata write persists the removal", async () => {
+    const dir = tempDir();
+    placeCrashState(
+      dir,
+      [callTurn("call-answered", 1), resultTurn("call-answered", "ok", 2)],
+      [
+        approvalOp("stale-corr", "call-answered"),
+        approvalOp("live-corr", "call-live"),
+      ],
+    );
+
+    const store = await createOptimizedContextStore(dir);
+    const loaded = await store.load();
+    await store.writeMetadata({
+      pendingOperations: loaded.pendingOperations,
+      tokenUsage: loaded.tokenUsage,
+    });
+
+    const persisted = JSON.parse(
+      fs.readFileSync(path.join(dir, "metadata.json"), "utf8"),
+    ) as { pendingOperations: { correlationId: string }[] };
+    expect(persisted.pendingOperations.map((op) => op.correlationId)).toEqual([
+      "live-corr",
+    ]);
+
+    const resumed = await createOptimizedContextStore(dir);
+    expect(
+      (await resumed.load()).pendingOperations.map((op) => op.correlationId),
+    ).toEqual(["live-corr"]);
+  });
+
+  test("crash-window replay across segments yields no re-armable gate", async () => {
+    const dir = tempDir();
+    fs.writeFileSync(
+      path.join(dir, TURNS_FILE),
+      jsonl([callTurn("call-answered", 1)]),
+    );
+    fs.writeFileSync(
+      path.join(dir, segmentFileName(TURNS_FILE, 1)),
+      jsonl([resultTurn("call-answered", "ok", 2)]),
+    );
+    fs.writeFileSync(
+      path.join(dir, "metadata.json"),
+      JSON.stringify({
+        pendingOperations: [
+          approvalOp("stale-corr", "call-answered"),
+          approvalOp("live-corr", "call-live"),
+          markerOp("marker-corr"),
+        ],
+        connectorState: null,
+      }),
+    );
+
+    const store = await createOptimizedContextStore(dir);
+    const loaded = await store.load();
+    const answered = new Set(
+      loaded.turns.flatMap((turn) =>
+        turn.content.flatMap((block) =>
+          block.type === "tool_result" ? [block.callId] : [],
+        ),
+      ),
+    );
+    // Every operation load() hands the rebuilt reactor is safe to rehydrate:
+    // none parks a call the turns already answer.
+    for (const op of loaded.pendingOperations) {
+      if (op.suspendedCall !== undefined) {
+        expect(answered.has(op.suspendedCall.id)).toBe(false);
+      }
+    }
+    expect(
+      loaded.pendingOperations.map((op) => op.correlationId).sort(),
+    ).toEqual(["live-corr", "marker-corr"]);
+  });
+
+  test("matches a scrubbed on-disk operation by call id", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    const secret = { api_key: "hunter2" };
+    await store.writeMetadata({
+      pendingOperations: [
+        {
+          ...approvalOp("scrubbed-corr", "call-scrubbed"),
+          suspendedCall: {
+            id: "call-scrubbed",
+            name: "run_shell",
+            arguments: secret,
+          },
+          approvalSnapshot: {
+            name: "run_shell",
+            description: "runs a shell command",
+            inputSchema: {},
+            arguments: secret,
+          },
+        },
+      ],
+      tokenUsage: {
+        input: 0,
+        output: 0,
+        cacheRead: 0,
+        cacheWrite: 0,
+        thinking: 0,
+      },
+    });
+    // The on-disk approval snapshot is redacted while suspendedCall keeps
+    // the raw call; matching must work by call id alone.
+    const raw = fs.readFileSync(path.join(dir, "metadata.json"), "utf8");
+    expect(raw).toContain("[redacted");
+    expect(raw).toContain("call-scrubbed");
+    await store.writeTurns([
+      callTurn("call-scrubbed", 1),
+      resultTurn("call-scrubbed", "ok", 2),
+    ]);
+
+    const resumed = await createOptimizedContextStore(dir);
+    expect((await resumed.load()).pendingOperations).toEqual([]);
+  });
+});

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
@@ -1674,7 +1690,35 @@ export async function createSessionStores(
         );
         metadata = emptyMetadata();
       }
-      return { turns, ...metadata };
+      // Crash-window filter: load() is the only path into the rebuilt
+      // reactor's gate rehydration, so a pending operation whose parked call
+      // the turns already answer dies here instead of re-arming its gate.
+      // Marker operations carry no suspendedCall and pass through. Disk is
+      // untouched (no write-in-read): the removal converges on the next
+      // normal metadata write, which publishes because the filtered payload
+      // differs from the seeded disk bytes.
+      const answered = answeredCallIds(turns);
+      const stale = metadata.pendingOperations.filter(
+        (op) =>
+          op.suspendedCall !== undefined && answered.has(op.suspendedCall.id),
+      );
+      for (const op of stale) {
+        log.warn(
+          "dropped pending operation {correlationId}: its parked call {callId} already has a result",
+          {
+            correlationId: op.correlationId,
+            callId: op.suspendedCall?.id ?? "",
+          },
+        );
+      }
+      if (stale.length === 0) return { turns, ...metadata };
+      return {
+        turns,
+        ...metadata,
+        pendingOperations: metadata.pendingOperations.filter(
+          (op) => !stale.includes(op),
+        ),
+      };
     },
     // Buffer is carried only by this store: base.writeMetadata is never
     // called through the wrapper, so base's pendingConnectorState would

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 |