Merge pull request #1492 from corbitsdev/cl-10765-never-inject-a-timeout-result-for-a-parked-call-that-already

fix(reactor): never re-answer a parked call that already has a result

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