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.

Author: bwachman

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 |

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
 // ---------------------------------------------------------------------------

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;