perf(session): cut per-turn store walks, re-parses and stringifies

scripts/context-store-perf.ts

@@ -0,0 +1,170 @@
+#!/usr/bin/env bun
+/**
+ * CL-10666 before/after measurement: session-store per-turn I/O.
+ *
+ * One machine, one scenario: a fresh session dir plus N simulated reactor
+ * cycles (load, writePrompt, writeTurns, commit) through the optimized
+ * context store. Every node fs.promises call the store stack makes is
+ * counted (split by whether the path is under .git), along with wall time,
+ * so the per-turn walks, re-parses, and re-stringified comparisons become an
+ * honest delta between two builds on the same host.
+ *
+ * Usage:
+ *   bun scripts/context-store-perf.ts [turnCount]
+ *   bun scripts/context-store-perf.ts [turnCount] --json
+ */
+
+import fs from "node:fs";
+import path from "node:path";
+import type { ConversationTurn } from "@intx/types/runtime";
+import { createOptimizedContextStore } from "../src/session/optimized-context-store.js";
+
+const COUNTED_OPS = [
+  "access",
+  "copyFile",
+  "lstat",
+  "mkdir",
+  "open",
+  "opendir",
+  "readFile",
+  "readdir",
+  "readlink",
+  "rename",
+  "rmdir",
+  "stat",
+  "truncate",
+  "unlink",
+  "writeFile",
+] as const;
+
+const counts: Record<string, number> = {};
+const gitPaths: Record<string, number> = {};
+
+function bump(table: Record<string, number>, op: string): void {
+  table[op] = (table[op] ?? 0) + 1;
+}
+
+function installCounters(): void {
+  const target = fs.promises as unknown as Record<
+    string,
+    (...args: unknown[]) => unknown
+  >;
+  for (const op of COUNTED_OPS) {
+    const original = target[op];
+    if (typeof original !== "function") continue;
+    target[op] = function counted(this: unknown, ...args: unknown[]) {
+      const first = args[0];
+      if (typeof first === "string" && first.includes(`/.git/`)) {
+        bump(gitPaths, op);
+      }
+      bump(counts, op);
+      return original.apply(this, args);
+    };
+  }
+}
+
+function turn(text: string): ConversationTurn {
+  return { role: "user", content: [{ type: "text", text }], timestamp: 1 };
+}
+
+function assistantTurn(text: string): ConversationTurn {
+  return {
+    role: "assistant",
+    content: [{ type: "text", text }],
+    timestamp: 2,
+  };
+}
+
+function turnBody(index: number): string {
+  // ~1.5 KB of filler plus a marker line, close to a small unit of context.
+  return (
+    `turn ${index}: ` +
+    "boundary-check padding. ".repeat(60) +
+    `end-of-turn ${index}`
+  );
+}
+
+async function runScenario(dir: string, turnCount: number): Promise<number> {
+  const store = await createOptimizedContextStore(dir);
+  const history: ConversationTurn[] = [];
+  for (let index = 0; index < turnCount; index++) {
+    history.push(turn(turnBody(index)));
+    if (index % 4 === 1) history.push(assistantTurn(`reply to ${index}`));
+    await store.load();
+    await store.writePrompt(history);
+    await store.writeTurns(history);
+    await store.commit({ message: `checkpoint ${index}` });
+  }
+  const fresh = await createOptimizedContextStore(dir);
+  return (await fresh.load()).turns.length;
+}
+
+function sortedEntries(table: Record<string, number>): [string, number][] {
+  return Object.entries(table).sort((a, b) => b[1] - a[1]);
+}
+
+async function main(argv: string[]): Promise<number> {
+  const arg = argv[2];
+  const turnCount =
+    arg !== undefined && !arg.startsWith("-") ? Number(arg) : 40;
+  if (!Number.isFinite(turnCount) || turnCount < 1) {
+    console.error(
+      "usage: bun scripts/context-store-perf.ts [turnCount] [--json]",
+    );
+    return 1;
+  }
+  installCounters();
+
+  const dir = await fs.promises.mkdtemp(
+    path.join(process.cwd(), "context-store-perf-"),
+  );
+  const started = performance.now();
+  const loadedTurns = await runScenario(dir, turnCount);
+  const wallMs = Math.round(performance.now() - started);
+
+  const totalOps = Object.values(counts).reduce((sum, n) => sum + n, 0);
+  const gitOps = Object.values(gitPaths).reduce((sum, n) => sum + n, 0);
+
+  const asJson = argv.includes("--json");
+  if (asJson) {
+    console.log(
+      JSON.stringify(
+        {
+          turnCount,
+          wallMs,
+          loadedTurns,
+          totalFsOps: totalOps,
+          gitPathFsOps: gitOps,
+          fsOps: counts,
+          gitPathFsOpsByName: gitPaths,
+          perTurn: {
+            wallMs: +(wallMs / turnCount).toFixed(2),
+            fsOps: +(totalOps / turnCount).toFixed(2),
+          },
+        },
+        null,
+        2,
+      ),
+    );
+  } else {
+    console.log(
+      `scenario: ${turnCount} turns, one store instance (fresh dir); verified ${loadedTurns} turns at the end`,
+    );
+    console.log(
+      `wall: ${wallMs} ms (${(wallMs / turnCount).toFixed(2)} ms/turn)`,
+    );
+    console.log(
+      `fs ops: ${totalOps} total (${(totalOps / turnCount).toFixed(2)}/turn), of which ${gitOps} under .git`,
+    );
+    for (const [op, n] of sortedEntries(counts)) {
+      console.log(`  ${op}: ${n}`);
+    }
+    for (const [op, n] of sortedEntries(gitPaths)) {
+      console.log(`  (.git) ${op}: ${n}`);
+    }
+  }
+  await fs.promises.rm(dir, { recursive: true, force: true });
+  return 0;
+}
+
+process.exit(await main(process.argv));

src/session/anthropic-cache-prompt.ts

@@ -42,42 +42,37 @@ function passthrough(
   };
 }
 
-function newestAssistantTextIndex(turns: readonly ConversationTurn[]): number {
-  for (let i = turns.length - 1; i >= 0; i--) {
-    const turn = turns[i];
-    if (turn === undefined || turn.role !== "assistant") continue;
-    if (
-      turn.content.some(
-        (block) => block.type === "text" && block.text.length > 0,
-      )
-    ) {
-      return i;
-    }
-  }
-  return -1;
-}
-
-// Tool results before the newest assistant text are already consumed.
-// When the model has not written that text yet, the newest result turn is
-// the unconsumed suffix and stays whole.
-function stubBoundary(turns: readonly ConversationTurn[]): number {
-  const assistantText = newestAssistantTextIndex(turns);
-  if (assistantText >= 0) return assistantText;
-  for (let i = turns.length - 1; i >= 0; i--) {
-    const turn = turns[i];
-    if (turn?.content.some((block) => block.type === "tool_result")) return i;
-  }
-  return turns.length;
-}
-
-function toolNames(turns: readonly ConversationTurn[]): Map<string, string> {
+// One forward pass collects everything shrinkPrompt needs: every tool_call
+// id→name (a result's call turn can sit anywhere, so names span the whole
+// transcript), the newest assistant text index, and the last tool_result
+// index. The boundary is that text index when one exists, else the last
+// result turn (the unconsumed suffix stays whole when the model has not
+// written text yet), else the whole history.
+function stubSetup(turns: readonly ConversationTurn[]): {
+  names: Map<string, string>;
+  boundary: number;
+} {
   const names = new Map<string, string>();
-  for (const turn of turns) {
+  let newestAssistantText = -1;
+  let lastToolResult = -1;
+  for (let i = 0; i < turns.length; i++) {
+    const turn = turns[i];
+    if (turn === undefined) continue;
+    let hasText = false;
     for (const block of turn.content) {
       if (block.type === "tool_call") names.set(block.id, block.name);
+      else if (block.type === "tool_result") lastToolResult = i;
+      else if (block.type === "text" && block.text.length > 0) hasText = true;
     }
+    if (hasText && turn.role === "assistant") newestAssistantText = i;
   }
-  return names;
+  const boundary =
+    newestAssistantText >= 0
+      ? newestAssistantText
+      : lastToolResult >= 0
+        ? lastToolResult
+        : turns.length;
+  return { names, boundary };
 }
 
 function stubText(name: string | undefined): string {
@@ -124,8 +119,7 @@ function shrinkPrompt(turns: ConversationTurn[]): {
   output: ConversationTurn[];
   stubbed: number;
 } {
-  const boundary = stubBoundary(turns);
-  const names = toolNames(turns);
+  const { names, boundary } = stubSetup(turns);
   let stubbed = 0;
   let changed = false;
   const output = turns.map((turn, index) => {

src/session/attachment-store.test.ts

@@ -146,4 +146,61 @@ describe("ageImageBlocks / rehydrateAttachmentImages", () => {
     expect(aged.blobs).toHaveLength(1);
     expect(recorded).toEqual([]);
   });
+
+  test("identical markers across turns read the blob once (CL-10666)", async () => {
+    const text = formatAgedImageMarker({
+      uri: attachmentUri("img-shared"),
+      mimeType: "image/png",
+    });
+    const markerTurn: ConversationTurn = {
+      role: "user",
+      content: [{ type: "text", text }],
+      timestamp: 1,
+    };
+    const blob = new TextEncoder().encode(PNG_B64);
+    const reads: string[] = [];
+    const readBlob = async (key: string) => {
+      reads.push(key);
+      return blob;
+    };
+
+    const out = await rehydrateAttachmentImages(
+      [markerTurn, markerTurn, markerTurn],
+      readBlob,
+    );
+    expect(reads).toEqual(["img-shared"]);
+    for (const turn of out) {
+      expect(turn.content).toEqual([
+        {
+          type: "image",
+          source: { kind: "base64", mimeType: "image/png", data: PNG_B64 },
+        },
+      ]);
+    }
+    expect(reads).toHaveLength(1);
+  });
+
+  test("a missing shared blob keeps every copy of its marker from one read", async () => {
+    const text = formatAgedImageMarker({
+      uri: attachmentUri("img-gone"),
+      mimeType: "image/png",
+    });
+    const markerTurn: ConversationTurn = {
+      role: "user",
+      content: [{ type: "text", text }],
+      timestamp: 1,
+    };
+    const reads: string[] = [];
+    const out = await rehydrateAttachmentImages(
+      [markerTurn, markerTurn],
+      async (key) => {
+        reads.push(key);
+        throw new Error(`Blob not found: ${key}`);
+      },
+    );
+    expect(reads).toHaveLength(1);
+    for (const turn of out) {
+      expect(turn.content).toEqual([{ type: "text", text }]);
+    }
+  });
 });

src/session/attachment-store.ts

@@ -85,11 +85,33 @@ export async function ageImageBlocks(
   return { turn: { ...turn, content }, blobs };
 }
 
-/** Restore aged markers to image blocks via a blob reader; missing blobs leave the marker text. */
+/** Restore aged markers to image blocks via a blob reader; missing blobs leave
+ * the marker text. The same attachment id only reads its blob once per
+ * restore: compaction copies (excerpted spines, folded handoffs) repeat the
+ * identical marker, and each copy then deserializes from the one read. */
 export async function rehydrateAttachmentImages(
   turns: readonly ConversationTurn[],
   readBlob: (key: string) => Promise<Uint8Array>,
 ): Promise<ConversationTurn[]> {
+  const blobReads = new Map<
+    string,
+    Promise<{ bytes: Uint8Array } | { error: unknown }>
+  >();
+  const blobFor = (
+    key: string,
+  ): Promise<{ bytes: Uint8Array } | { error: unknown }> => {
+    const shared = blobReads.get(key);
+    if (shared !== undefined) return shared;
+    // One failure is memoized for the id too: missing blobs keep their
+    // markers, and re-serving that outcome avoids a retry storm per copy.
+    // Each marker still warns; the shared read does not hide the failure.
+    const sharedRead = readBlob(key).then(
+      (bytes) => ({ bytes }),
+      (error: unknown) => ({ error }),
+    );
+    blobReads.set(key, sharedRead);
+    return sharedRead;
+  };
   const out: ConversationTurn[] = [];
   for (const turn of turns) {
     let changed = false;
@@ -104,22 +126,23 @@ export async function rehydrateAttachmentImages(
         content.push(block);
         continue;
       }
-      try {
-        const bytes = await readBlob(marker.id);
-        const data = new TextDecoder().decode(bytes);
-        content.push({
-          type: "image",
-          source: {
-            kind: "base64",
-            mimeType: marker.mimeType,
-            data,
-          },
-        });
-        changed = true;
-      } catch (err) {
+      const read = await blobFor(marker.id);
+      if ("error" in read) {
+        const err = read.error;
         logger.warn`aged image marker stays dead in context image=${marker.id} error=${err instanceof Error ? err.message : String(err)}`;
         content.push(block);
+        continue;
       }
+      const data = new TextDecoder().decode(read.bytes);
+      content.push({
+        type: "image",
+        source: {
+          kind: "base64",
+          mimeType: marker.mimeType,
+          data,
+        },
+      });
+      changed = true;
     }
     out.push(changed ? { ...turn, content } : turn);
   }

src/session/attachment-uri.ts

@@ -36,6 +36,12 @@ export interface AgedImageMarker {
 export function parseAgedImageMarker(
   text: string,
 ): AgedImageMarker | undefined {
+  // Prefilter: markers are the only shape that can start this way, and text
+  // blocks here include whole tool outputs, so skip the regex on anything
+  // that cannot possibly match.
+  if (!text.startsWith("[image attachment aged: attachment:///")) {
+    return undefined;
+  }
   // [image attachment aged: attachment:///ID mimeType=TYPE — ...]
   const match = text.match(
     /^\[image attachment aged: (attachment:\/\/\/[^\s]+) mimeType=([^\s]+) —/,