From 8d879e9448a87e815105562bfa895571ea60093e Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Wed, 14 Oct 2026 13:45:09 -0700 Subject: [PATCH] perf(session): cut per-turn store walks, re-parses and stringifies A 200-turn simulation dropped 139 to 124 fs ops per turn (access probes 2697 to 3) and wall median ~13%. commit() now surveys the worktree once instead of stat-probing every candidate segment; a per-store turns cache with per-segment stat fingerprints plus explicit mutation-site invalidation loads each segment at most once, so load/ readLiveTurns/prompt-write parse once per content instead of three times. contentPrefixLength/promptEqualsLiveTurns memoize turn JSON instead of stringifying full history; the well-formedness prefix scan is a single forward pass instead of O(E^2); readAt reads a commit-time segment manifest and parallel blob reads, degrading to the ref-walk when the manifest is absent or torn; attachment blob reads dedupe; aged-image markers prefilter before regex; anthropic cache prompt aggregates one forward scan; compaction identities are cached per tool_call block. Invalidation is strictly wider than the reads it replaced: mutation sites and failed-commit restores drop the caches; corrupt manifests degrade to the walk. --- scripts/context-store-perf.ts | 170 +++++ src/session/anthropic-cache-prompt.ts | 60 +- src/session/attachment-store.test.ts | 57 ++ src/session/attachment-store.ts | 44 +- src/session/attachment-uri.ts | 6 + src/session/compactor.ts | 46 +- src/session/optimized-context-store.test.ts | 172 +++++ src/session/optimized-context-store.ts | 681 +++++++++++++++----- 8 files changed, 1012 insertions(+), 224 deletions(-) create mode 100644 scripts/context-store-perf.ts diff --git a/scripts/context-store-perf.ts b/scripts/context-store-perf.ts new file mode 100644 index 000000000..b13a54dc3 --- /dev/null +++ b/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 = {}; +const gitPaths: Record = {}; + +function bump(table: Record, 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 { + 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][] { + return Object.entries(table).sort((a, b) => b[1] - a[1]); +} + +async function main(argv: string[]): Promise { + 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)); diff --git a/src/session/anthropic-cache-prompt.ts b/src/session/anthropic-cache-prompt.ts index 08e209773..fd85e2f4a 100644 --- a/src/session/anthropic-cache-prompt.ts +++ b/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 { +// 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; + boundary: number; +} { const names = new Map(); - 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) => { diff --git a/src/session/attachment-store.test.ts b/src/session/attachment-store.test.ts index 31cafa596..715767edd 100644 --- a/src/session/attachment-store.test.ts +++ b/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 }]); + } + }); }); diff --git a/src/session/attachment-store.ts b/src/session/attachment-store.ts index 600cf2a29..b9432baa8 100644 --- a/src/session/attachment-store.ts +++ b/src/session/attachment-store.ts @@ -81,11 +81,27 @@ 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, ): Promise { + const blobReads = new Map>(); + const blobFor = (key: string): Promise => { + 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. + const sharedRead = readBlob(key).then( + (bytes) => bytes, + () => null, + ); + blobReads.set(key, sharedRead); + return sharedRead; + }; const out: ConversationTurn[] = []; for (const turn of turns) { let changed = false; @@ -100,21 +116,21 @@ 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 { + const bytes = await blobFor(marker.id); + if (bytes === null) { content.push(block); + continue; } + const data = new TextDecoder().decode(bytes); + content.push({ + type: "image", + source: { + kind: "base64", + mimeType: marker.mimeType, + data, + }, + }); + changed = true; } out.push(changed ? { ...turn, content } : turn); } diff --git a/src/session/attachment-uri.ts b/src/session/attachment-uri.ts index 130e67911..5016228ac 100644 --- a/src/session/attachment-uri.ts +++ b/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]+) —/, diff --git a/src/session/compactor.ts b/src/session/compactor.ts index 364ac7640..01865ff98 100644 --- a/src/session/compactor.ts +++ b/src/session/compactor.ts @@ -209,6 +209,35 @@ function queryIdentityFromArguments( } // callId → tool name/path for readable stubs. Inverse of path-to-reads. +// Persisted tool_call blocks are immutable once appended, and re-parsed +// history rebuilds its own entries anyway, so identity work on a block pays +// once per object instead of once per compaction pass over the transcript. +const callIdentityMemo = new WeakMap< + Extract, + ToolCallInfo +>(); + +function toolCallInfoFor( + block: Extract, +): ToolCallInfo { + const memo = callIdentityMemo.get(block); + if (memo !== undefined) return memo; + // Persisted calls keep the wire name (read/glob/…); the read/query + // classification sets are engine-keyed. + const info: ToolCallInfo = { name: canonicalToolName(block.name) }; + const identity = readIdentityFromArguments(block.arguments); + if (identity !== undefined) { + info.pathArg = identity.path; + info.readKey = identity.readKey; + } + if (QUERY_TOOLS.has(block.name)) { + const queryKey = queryIdentityFromArguments(block.name, block.arguments); + if (queryKey !== undefined) info.readKey = queryKey; + } + callIdentityMemo.set(block, info); + return info; +} + function buildCallIndex( turns: readonly ConversationTurn[], ): Map { @@ -216,22 +245,7 @@ function buildCallIndex( for (const turn of turns) { for (const block of turn.content) { if (block.type !== "tool_call") continue; - // Persisted calls keep the wire name (read/glob/…); the read/query - // classification sets are engine-keyed. - const info: ToolCallInfo = { name: canonicalToolName(block.name) }; - const identity = readIdentityFromArguments(block.arguments); - if (identity !== undefined) { - info.pathArg = identity.path; - info.readKey = identity.readKey; - } - if (QUERY_TOOLS.has(block.name)) { - const queryKey = queryIdentityFromArguments( - block.name, - block.arguments, - ); - if (queryKey !== undefined) info.readKey = queryKey; - } - index.set(block.id, info); + index.set(block.id, toolCallInfoFor(block)); } } return index; diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index c6bfd9835..320448bb5 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -1727,3 +1727,175 @@ describe("createOptimizedContextStore prompt dedupe (CL-9026)", () => { expect(turnTexts((await store.load()).turns)).toEqual(turnTexts(live)); }); }); + +describe("createOptimizedContextStore per-turn perf (CL-10666)", () => { + const SEGMENT_MANIFEST_FILE = "segment-manifest.json"; + + test("an orphan extra in the middle drops it and everything after it", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + const callId = "call-middle-orphan"; + + const head: ConversationTurn[] = [ + { + role: "assistant", + content: [ + { type: "tool_call", id: callId, name: "grep", arguments: {} }, + ], + timestamp: 1, + }, + { + role: "user", + content: [ + { + type: "tool_result", + callId, + content: [{ type: "text", text: "ok" }], + }, + ], + timestamp: 2, + }, + ]; + // [well-formed, orphan, well-formed] — the incremental pass must stop at + // the orphan exactly where the old rescan-per-length loop did: keep 1 of + // 3 extras (segment 1), unlink segments 2 and 3. + const good: ConversationTurn[] = [ + { + role: "user", + content: [{ type: "text", text: "kept" }], + timestamp: 10, + }, + ]; + const orphan: ConversationTurn[] = [ + { + role: "user", + content: [ + { + type: "tool_result", + callId, + content: [{ type: "text", text: "again" }], + }, + ], + timestamp: 20, + }, + ]; + fs.writeFileSync( + path.join(dir, segmentFileName(TURNS_FILE, 1)), + jsonl(good), + ); + fs.writeFileSync( + path.join(dir, segmentFileName(TURNS_FILE, 2)), + jsonl(orphan), + ); + fs.writeFileSync( + path.join(dir, segmentFileName(TURNS_FILE, 3)), + jsonl([turn("last")]), + ); + fs.writeFileSync(path.join(dir, TURNS_FILE), jsonl(head)); + + const loaded = await store.load(); + expect(loaded.turns).toHaveLength(3); + expect(turnTexts(loaded.turns)).toEqual([...turnTexts(head), "kept"]); + expect(await listSegmentFiles(dir, TURNS_FILE)).toEqual([ + TURNS_FILE, + segmentFileName(TURNS_FILE, 1), + ]); + }); + + test("loads observe external segment mutations, byte-identical edits included", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + fs.writeFileSync(path.join(dir, TURNS_FILE), jsonl([turn("a"), turn("b")])); + fs.writeFileSync( + path.join(dir, segmentFileName(TURNS_FILE, 1)), + jsonl([turn("c")]), + ); + expect(turnTexts((await store.load()).turns)).toEqual(["a", "b", "c"]); + + // Crash-artifact appends land in the active segment between loads; the + // cached generation must invalidate (stat change) and merge them. + fs.appendFileSync( + path.join(dir, segmentFileName(TURNS_FILE, 1)), + jsonl([turn("d")]), + ); + expect(turnTexts((await store.load()).turns)).toEqual(["a", "b", "c", "d"]); + + // Same-length rewrite of the active segment ("c" -> "x") still invalidates. + fs.writeFileSync( + path.join(dir, segmentFileName(TURNS_FILE, 1)), + jsonl([turn("x"), turn("d")]), + ); + expect(turnTexts((await store.load()).turns)).toEqual(["a", "b", "x", "d"]); + + // Torn tail on the active segment keeps dropping it stably. + const active = path.join(dir, segmentFileName(TURNS_FILE, 1)); + fs.writeFileSync( + active, + jsonl([turn("x"), turn("d")]) + '{"role":"user","content":[{"type":"te', + ); + expect(turnTexts((await store.load()).turns)).toEqual(["a", "b", "x", "d"]); + }); + + test("commits a segment manifest only when turn extras exist and readAt consults it", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + const bigText = "x".repeat(300 * 1024); + const bigTurn: ConversationTurn = { + role: "user", + content: [{ type: "text", text: bigText }], + timestamp: 1, + }; + await store.writeTurns([bigTurn]); + await store.writeMetadata(EMPTY_CHECKPOINT_METADATA); + const first = await store.commit({ message: "base only" }); + expect(fs.existsSync(path.join(dir, SEGMENT_MANIFEST_FILE))).toBe(false); + + const last = turn("tail"); + await store.writeTurns([bigTurn, last]); + await store.writeMetadata(EMPTY_CHECKPOINT_METADATA); + const second = await store.commit({ message: "with extra" }); + const extraTurns = segmentFileName(TURNS_FILE, 1); + expect(fs.existsSync(path.join(dir, SEGMENT_MANIFEST_FILE))).toBe(true); + const tree = await gitLsTree(dir); + expect(tree).toContain(SEGMENT_MANIFEST_FILE); + expect( + fs.readFileSync(path.join(dir, SEGMENT_MANIFEST_FILE), "utf-8"), + ).toBe(JSON.stringify({ turnSegments: [extraTurns] })); + + // Manifest path and tree-walk path agree. + expect(turnTexts(await store.readAt(second.hash))).toEqual([ + bigText, + "tail", + ]); + expect(await store.readAt(first.hash)).toHaveLength(1); + }); + + test("a corrupt committed segment manifest falls back to the commit-tree walk", async () => { + const dir = tempDir(); + const store = await createOptimizedContextStore(dir); + const bigText = "x".repeat(300 * 1024); + const bigTurn: ConversationTurn = { + role: "user", + content: [{ type: "text", text: bigText }], + timestamp: 1, + }; + await store.writeTurns([bigTurn, turn("tail")]); + await store.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await store.commit({ message: "with extra" }); + + fs.writeFileSync(path.join(dir, SEGMENT_MANIFEST_FILE), "{torn"); + await store.writeMetadata(EMPTY_CHECKPOINT_METADATA); + const after = await store.commit({ message: "corrupt manifest" }); + + expect(turnTexts(await store.readAt(after.hash))).toEqual([ + bigText, + "tail", + ]); + await store.writeTurns([bigTurn, turn("tail"), turn("newer")]); + await store.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await store.commit({ message: "heal manifest" }); + expect( + fs.readFileSync(path.join(dir, SEGMENT_MANIFEST_FILE), "utf-8"), + ).toBe(JSON.stringify({ turnSegments: [segmentFileName(TURNS_FILE, 1)] })); + }); +}); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 85aacffd6..950a75d3e 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -19,7 +19,6 @@ import { createSegmentedJSONLWriter, highestSegmentIndex, listSegmentFiles, - readExtraSegmentTexts, segmentFileName, } from "./incremental-jsonl.js"; import type { ContextStore } from "@intx/types/runtime"; @@ -53,6 +52,24 @@ const BLOB_EXTENSIONS: Readonly> = { const UNSAFE_FILENAME_CHARS = /[^a-zA-Z0-9_-]/g; +/** + * Commit-time record of the contiguous turn segments tracked at that commit. + * `readAt` reads this blob instead of walking the commit tree, and reads the + * listed segments' blobs concurrently. Recomputed from disk truth every + * commit, so a worktree manifest left behind by a failed commit self-corrects + * on the next one. + */ +const SEGMENT_MANIFEST_FILE = "segment-manifest.json"; +const EMPTY_SEGMENT_MANIFEST = JSON.stringify({ turnSegments: [] }); + +const SegmentManifestSchema = type({ + turnSegments: "string[]", +}); + +// Only rolling-segment names are ever written here; anything else in a read +// manifest means it was not written by this store. +const TURN_SEGMENT_NAME_PATTERN = /^turns(-\d+)?\.jsonl$/; + const ConversationTurnSchema = type({ role: "'user' | 'assistant' | 'system'", content: ContentBlock.array(), @@ -60,6 +77,19 @@ const ConversationTurnSchema = type({ timestamp: "number", }); +// Persisted turns are immutable once appended; re-parsed history rebuilds its +// memo entries anyway, so this skips re-serializing an unchanged history on +// every comparison (contentPrefixLength, promptEqualsLiveTurns). +const turnJsonMemo = new WeakMap(); + +function turnJson(turn: ConversationTurn): string { + const memo = turnJsonMemo.get(turn); + if (memo !== undefined) return memo; + const json = JSON.stringify(turn); + turnJsonMemo.set(turn, json); + return json; +} + async function pathExists(fullPath: string): Promise { try { await fs.promises.access(fullPath); @@ -71,6 +101,19 @@ async function pathExists(fullPath: string): Promise { } } +async function readSegmentManifestText(dir: string): Promise { + try { + return await fs.promises.readFile( + path.join(dir, SEGMENT_MANIFEST_FILE), + "utf-8", + ); + } catch (cause) { + if (cause instanceof Error && "code" in cause && cause.code === "ENOENT") + return null; + throw cause; + } +} + function blobExtensionFor(contentType: string | undefined): string { if (contentType === undefined) return ""; return BLOB_EXTENSIONS[contentType] ?? ""; @@ -226,12 +269,19 @@ async function loadMetadataSoft( } } -// Non-throwing mirror of assertWellFormedToolSequence for picking the -// longest loadable segment prefix. Unpaired trailing tool_calls pass; dups -// and orphan results fail. -function toolSequenceIsWellFormed(turns: readonly ConversationTurn[]): boolean { - const calledIds = new Set(); - const answeredIds = new Set(); +/** + * Extends accumulated tool-sequence state with one segment's turns. Unpaired + * trailing tool_calls pass; dups and orphan results fail. Failure is monotone + * in the accumulated id sets: once a duplicate call or orphan result enters, + * no later segment can undo it, so a single forward pass per segment walks + * the same longest-well-formed prefix the previous re-scan-per-length loop + * picked, without rebuilding the merged history once per candidate length. + */ +function accumulateToolSequenceState( + turns: readonly ConversationTurn[], + calledIds: Set, + answeredIds: Set, +): boolean { for (const turn of turns) { for (const block of turn.content) { if (block.type === "tool_call") { @@ -254,17 +304,23 @@ function toolSequenceIsWellFormed(turns: readonly ConversationTurn[]): boolean { * session. */ function longestWellFormedExtraCount( - baseTurns: ConversationTurn[], - parsedExtras: ConversationTurn[][], + baseTurns: readonly ConversationTurn[], + parsedExtras: readonly ConversationTurn[][], ): number { - for (let keepExtras = parsedExtras.length; keepExtras >= 0; keepExtras--) { - const turns = - keepExtras === 0 - ? baseTurns - : [...baseTurns, ...parsedExtras.slice(0, keepExtras).flat()]; - if (toolSequenceIsWellFormed(turns)) return keepExtras; + const calledIds = new Set(); + const answeredIds = new Set(); + let wellFormed = accumulateToolSequenceState( + baseTurns, + calledIds, + answeredIds, + ); + let keep = 0; + for (const extra of parsedExtras) { + if (!wellFormed) break; + wellFormed = accumulateToolSequenceState(extra, calledIds, answeredIds); + if (wellFormed) keep += 1; } - return 0; + return keep; } async function unlinkExtraSegmentsFrom( @@ -652,6 +708,94 @@ async function listIndexPaths(dir: string): Promise> { return new Set(await git.listFiles({ fs, dir })); } +/** + * Segment names contiguous from the base name, derived from one readdir of + * the session dir. Equivalent to listSegmentFiles without its per-candidate + * stat; a readdir is shared by every staging check in one commit. + */ +function contiguousSegmentNamesFromListing( + names: ReadonlySet, + baseName: string, +): string[] { + if (!names.has(baseName)) return []; + const out = [baseName]; + for (let index = 1; ; index++) { + const name = segmentFileName(baseName, index); + if (!names.has(name)) break; + out.push(name); + } + return out; +} + +function highestSegmentIndexFromListing( + names: ReadonlySet, + baseName: string, +): number { + const dot = baseName.lastIndexOf("."); + const stem = dot === -1 ? baseName : baseName.slice(0, dot); + const ext = dot === -1 ? "" : baseName.slice(dot); + const pattern = new RegExp( + `^${stem.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}-(\\d+)${ext.replace(/[.*+?^${}()|[\]\\]/g, "\\$&")}$`, + ); + let highest = names.has(baseName) ? 0 : -1; + for (const name of names) { + const match = pattern.exec(name); + if (match === null) continue; + const index = Number(match[1]); + if (Number.isFinite(index) && index > highest) highest = index; + } + return highest; +} + +/** + * One readdir of the session dir answers every worktree-existence check in a + * single commit, including the lazy listing of the tool-output dir. Surveyed + * after the staged writes land, used before the lock releases. + */ +interface WorktreeListing { + top: ReadonlySet; + toolOutputFiles: () => Promise>; +} + +async function surveyWorktree(dir: string): Promise { + const top = new Set(await fs.promises.readdir(dir)); + let blobListing: Promise> | null = null; + return { + top, + toolOutputFiles() { + blobListing ??= fs.promises + .readdir(path.join(dir, TOOL_OUTPUT_DIR)) + .then((names) => new Set(names)) + .catch((cause: unknown) => { + if ( + cause instanceof Error && + "code" in cause && + cause.code === "ENOENT" + ) { + return new Set(); + } + throw cause; + }); + return blobListing; + }, + }; +} + +async function worktreeListingHas( + dir: string, + listing: WorktreeListing, + filepath: string, +): Promise { + const slash = filepath.indexOf("/"); + if (slash === -1) return listing.top.has(filepath); + if (filepath.startsWith(`${TOOL_OUTPUT_DIR}/`)) { + const nested = filepath.slice(TOOL_OUTPUT_DIR.length + 1); + return (await listing.toolOutputFiles()).has(nested); + } + // Unknown nested shape: fall back to the direct probe. + return pathExists(path.join(dir, filepath)); +} + async function extraSegmentNamesAtCommit( dir: string, hash: string, @@ -676,6 +820,60 @@ async function blobTextAtCommit( return new TextDecoder().decode(blob); } +/** + * Contiguous turn extras tracked at `hash`: the commit-time manifest when + * present, else the ref-tree walk for commits that predate the manifest. + * Manifest entries are validated because a torn or hand-corrupted blob must + * degrade to the tree walk, never fail the readAt. + */ +function isNotFoundError(cause: unknown): boolean { + return cause instanceof Error && cause.name === "NotFoundError"; +} + +async function turnSegmentNamesAtCommit( + dir: string, + hash: string, +): Promise { + let manifest: string | null = null; + try { + manifest = await blobTextAtCommit(dir, hash, SEGMENT_MANIFEST_FILE); + } catch (cause) { + // A commit without the manifest is the pre-manifest normal case; any + // other read error falls through to the tree walk as well. + if (!isNotFoundError(cause)) { + log.warn( + "unreadable segment manifest at commit {hash}; re-deriving from the commit tree", + { hash, cause: cause instanceof Error ? cause.message : String(cause) }, + ); + } + } + if (manifest !== null) { + try { + const parsed: unknown = JSON.parse(manifest); + const result = SegmentManifestSchema(parsed); + if ( + !(result instanceof type.errors) && + result.turnSegments.every( + (name) => name !== TURNS_FILE && TURN_SEGMENT_NAME_PATTERN.test(name), + ) + ) { + return result.turnSegments; + } + log.warn( + "malformed segment manifest at commit {hash}; re-deriving from the commit tree", + { hash }, + ); + } catch { + // Torn or non-JSON bytes degrade to the tree walk, same as a missing one. + log.warn( + "malformed segment manifest at commit {hash}; re-deriving from the commit tree", + { hash }, + ); + } + } + return extraSegmentNamesAtCommit(dir, hash); +} + async function resetIndexPaths( dir: string, filepaths: readonly string[], @@ -753,29 +951,31 @@ async function restoreTurnSegments( * Stage contiguous on-disk segments for `baseName`; unstage any * higher-numbered or gapped segment a rewrite deleted, even when the pending * set was lost (crash between heal unlink and commit). Gapped strays are - * unlinked, not re-added. + * unlinked, not re-added. Existence and index checks come from the shared + * worktree listing and the shared tracked-index listing, so one readdir and + * one tree walk serve every staged base name in the commit. */ async function reconcileSegmentStaging( dir: string, baseName: string, toAdd: string[], toRemove: string[], + listing: WorktreeListing, + tracked: ReadonlySet, ): Promise { - const contiguous = await listSegmentFiles(dir, baseName); + const contiguous = contiguousSegmentNamesFromListing(listing.top, baseName); for (const name of contiguous) toAdd.push(name); // Index 0 is the base name; numbered tails begin at 1. No base still // sweeps numbered files. const startIndex = Math.max(contiguous.length, 1); - const highestDisk = await highestSegmentIndex(dir, baseName); - const tracked = await listIndexPaths(dir); + const highestDisk = highestSegmentIndexFromListing(listing.top, baseName); for (let index = startIndex; ; index++) { const name = segmentFileName(baseName, index); - const full = path.join(dir, name); - if (await pathExists(full)) { + if (listing.top.has(name)) { // Gapped or post-contiguous stray — not part of the live history. - await fs.promises.unlink(full); + await fs.promises.unlink(path.join(dir, name)); toRemove.push(name); continue; } @@ -874,10 +1074,16 @@ export async function createSessionStores( ): number { const max = Math.min(prev.length, next.length); let prefix = 0; - while ( - prefix < max && - JSON.stringify(prev[prefix]) === JSON.stringify(next[prefix]) - ) { + while (prefix < max) { + const prevTurn = prev[prefix]; + const nextTurn = next[prefix]; + if ( + prevTurn === undefined || + nextTurn === undefined || + turnJson(prevTurn) !== turnJson(nextTurn) + ) { + break; + } prefix++; } return prefix; @@ -891,13 +1097,146 @@ export async function createSessionStores( for (const filepath of modifiedPaths) pendingSegmentPaths.add(filepath); } - async function readLiveTurns(): Promise { - const basePresent = await pathExists(path.join(dir, TURNS_FILE)); - const extraPresent = await pathExists( - path.join(dir, segmentFileName(TURNS_FILE, 1)), - ); - if (!basePresent && !extraPresent) return []; - return loadLiveTurnsFromDisk(); + /** + * Per-generation live-turns parse cache. `load()`, `resolveFirstWrite`, and + * the fresh-instance prompt path all read this instead of re-reading and + * re-parsing the full segment set, so one turn cycle performs one parse of + * the history (plus stats) instead of three. A stat change on any segment + * invalidates that segment only — parses are reused segment by segment — + * and every in-process mutation site invalidates explicitly, so anything + * cached dies on the triggers the unbounded disk read it replaced would + * have observed afresh. + */ + type SegmentFingerprint = { + name: string; + bytes: number; + mtimeMs: number; + ino: number; + }; + + type TurnsGenerationCache = { + segments: SegmentFingerprint[]; + baseTurns: ConversationTurn[]; + parsedExtras: ConversationTurn[][]; + merged: ConversationTurn[]; + }; + + let turnsCache: TurnsGenerationCache | null = null; + + function invalidateTurnsCache(): void { + turnsCache = null; + } + + async function turnSegmentStats(): Promise { + const listing = new Set(await fs.promises.readdir(dir)); + if (!listing.has(TURNS_FILE)) return []; + const out: SegmentFingerprint[] = []; + for (let index = 0; ; index++) { + const name = segmentFileName(TURNS_FILE, index); + if (!listing.has(name)) break; + const stats = await fs.promises.stat(path.join(dir, name)); + out.push({ + name, + bytes: stats.size, + mtimeMs: stats.mtimeMs, + ino: stats.ino, + }); + } + return out; + } + + /** Count of leading segments whose stat still matches the cached parse. */ + function sharedSegmentPrefix( + cached: SegmentFingerprint[], + observed: SegmentFingerprint[], + ): number { + let count = 0; + while (count < cached.length && count < observed.length) { + const before = cached[count]; + const after = observed[count]; + if ( + before === undefined || + after === undefined || + before.name !== after.name || + before.bytes !== after.bytes || + before.mtimeMs !== after.mtimeMs || + before.ino !== after.ino + ) { + break; + } + count++; + } + return count; + } + + async function readLiveTurnsCached( + signal?: AbortSignal, + ): Promise { + const segments = await turnSegmentStats(); + if ( + turnsCache !== null && + turnsCache.segments.length === segments.length && + sharedSegmentPrefix(turnsCache.segments, segments) === segments.length + ) { + return turnsCache.merged; + } + const reused = + turnsCache === null + ? 0 + : sharedSegmentPrefix(turnsCache.segments, segments); + // Base segment stat unchanged: its parse is still exactly what disk + // holds, so only the newer segments get re-read and re-parsed. + let baseTurns: ConversationTurn[]; + if (reused >= 1 && turnsCache !== null) { + baseTurns = turnsCache.baseTurns; + } else { + try { + baseTurns = (await base.load(signal)).turns; + } catch (cause) { + // A torn or poisoned base must not block the healing write; recover + // the usable turns the same way load() always has. Corrupt + // metadata.json also slides — callers here only need the turns. + log.warn( + "base context store load failed; recovering base segment from disk", + { cause: cause instanceof Error ? cause.message : String(cause) }, + ); + baseTurns = await readBaseTurnsFromDisk(dir); + } + } + const parsedExtras: ConversationTurn[][] = []; + for (let index = 1; index < segments.length; index++) { + const cached = + index < reused && turnsCache !== null + ? turnsCache.parsedExtras[index - 1] + : undefined; + if (cached !== undefined) { + parsedExtras.push(cached); + continue; + } + const segment = segments.at(index); + if (segment === undefined) continue; + const text = await fs.promises.readFile( + path.join(dir, segment.name), + "utf-8", + ); + parsedExtras.push( + parseSegmentTurns( + text, + index === segments.length - 1, + segment.name, + true, + ), + ); + } + const merged = await mergedLiveTurns(baseTurns, parsedExtras); + turnsCache = { segments, baseTurns, parsedExtras, merged }; + return merged; + } + + async function readLiveTurns( + signal?: AbortSignal, + ): Promise { + return readLiveTurnsCached(signal); } async function resolveFirstWrite( @@ -980,6 +1319,7 @@ export async function createSessionStores( } await writeSegmented(writeTurnsSegmented, turns); liveTurnRefs = [...turns]; + invalidateTurnsCache(); return; } await withResolvedDirLock(dir, async () => { @@ -994,6 +1334,7 @@ export async function createSessionStores( } await writeSegmented(writeTurnsSegmented, turns); liveTurnRefs = [...turns]; + invalidateTurnsCache(); return; } const decision = await resolveFirstWrite(turns); @@ -1013,28 +1354,37 @@ export async function createSessionStores( } await writeSegmented(writeTurnsSegmented, turns); liveTurnRefs = [...turns]; + invalidateTurnsCache(); await flushPendingPrompt(); }); } - async function loadLiveTurnsFromDisk(): Promise { - const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE); - let baseTurns: ConversationTurn[]; - try { - baseTurns = (await base.load()).turns; - } catch (cause) { - // A torn or poisoned base must not block the healing write; recover the - // usable turns the same way load() does. Corrupt metadata.json also - // slides — writeTurns only needs the turns. + // Prefer the longest loadable prefix of base + extras; orphan tails from a + // fresh-writer compaction rewrite are dropped and unlinked so the next load + // does not re-poison the session. + async function mergedLiveTurns( + baseTurns: ConversationTurn[], + parsedExtras: ConversationTurn[][], + ): Promise { + if (parsedExtras.length === 0) return baseTurns; + const keepExtras = longestWellFormedExtraCount(baseTurns, parsedExtras); + + if (keepExtras < parsedExtras.length) { + // Segment file index for the first extra is 1. + const removed = await unlinkExtraSegmentsFrom( + dir, + keepExtras + 1, + pendingSegmentPaths, + ); log.warn( - "base context store load failed during writeTurns; recovering base segment from disk", - { cause: cause instanceof Error ? cause.message : String(cause) }, + "dropped {removed} orphan turn segment(s) starting at index {fromIndex} (malformed tool sequence when concatenated)", + { removed, fromIndex: keepExtras + 1 }, ); - baseTurns = await readBaseTurnsFromDisk(dir); + invalidateTurnsCache(); } - return extraTexts.length === 0 - ? baseTurns - : await loadTurnsWithoutMalformedToolSequence(baseTurns, extraTexts); + + if (keepExtras === 0) return baseTurns; + return [...baseTurns, ...parsedExtras.slice(0, keepExtras).flat()]; } function promptEqualsLiveTurns( @@ -1043,9 +1393,11 @@ export async function createSessionStores( ): boolean { if (live.length !== prompt.length) return false; for (let index = 0; index < prompt.length; index++) { - if (live[index] === prompt[index]) continue; - if (JSON.stringify(live[index]) !== JSON.stringify(prompt[index])) - return false; + const liveTurn = live[index]; + const promptTurn = prompt[index]; + if (liveTurn === undefined || promptTurn === undefined) return false; + if (liveTurn === promptTurn) continue; + if (turnJson(liveTurn) !== turnJson(promptTurn)) return false; } return true; } @@ -1061,24 +1413,11 @@ export async function createSessionStores( if (unpublishedRewrite !== null) live = unpublishedRewrite; else if (liveTurnRefs !== null) live = liveTurnRefs; else { - // Fresh instance: recover live turns from disk like - // writeTurnsLiveOrStage; failure falls through to a normal write so an - // unreadable baseline cannot drop the snapshot. + // Fresh instance: recover live turns through the shared generation + // cache; failure falls through to a normal write so an unreadable + // baseline cannot drop the snapshot. try { - const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE); - let baseTurns: ConversationTurn[]; - try { - baseTurns = (await base.load()).turns; - } catch { - baseTurns = await readBaseTurnsFromDisk(dir); - } - live = - extraTexts.length === 0 - ? baseTurns - : await loadTurnsWithoutMalformedToolSequence( - baseTurns, - extraTexts, - ); + live = await readLiveTurnsCached(); } catch { live = null; } @@ -1108,107 +1447,74 @@ export async function createSessionStores( } } - // Prefer the longest loadable prefix of base + extras; orphan tails from a - // fresh-writer compaction rewrite are dropped and unlinked so the next load - // does not re-poison the session. - async function loadTurnsWithoutMalformedToolSequence( - baseTurns: ConversationTurn[], - extraTexts: string[], - ): Promise { - if (extraTexts.length === 0) return baseTurns; - - const parsedExtras = extraTexts.map((text, index) => - parseSegmentTurns( - text, - index === extraTexts.length - 1, - segmentFileName(TURNS_FILE, index + 1), - true, - ), - ); - const keepExtras = longestWellFormedExtraCount(baseTurns, parsedExtras); - - if (keepExtras < parsedExtras.length) { - // Segment file index for the first extra is 1. - const removed = await unlinkExtraSegmentsFrom( - dir, - keepExtras + 1, - pendingSegmentPaths, - ); - log.warn( - "dropped {removed} orphan turn segment(s) starting at index {fromIndex} (malformed tool sequence when concatenated)", - { removed, fromIndex: keepExtras + 1 }, - ); - } - - if (keepExtras === 0) return baseTurns; - return [...baseTurns, ...parsedExtras.slice(0, keepExtras).flat()]; - } - const store: ContextStore & AuditStore = { - // Full-history read: the reactor's initialization contract. When the base - // isogit store hard-fails (null-padded or mid-file garbage), recover - // usable turns via resilient segment parse and re-read metadata via the - // base schema (soft-empty only if that fails too), so resume does not die - // or wipe pending ops. + // Full-history read: the reactor's initialization contract. Turns come + // through the shared generation cache (strict base parse with resilient + // recovery, resilient extras, orphan-tail heal), so a repeated load within + // one store generation does not re-parse segments whose stat is unchanged. + // Metadata stays a fresh read: corrupt metadata.json soft-defaults so + // resume does not die or wipe pending ops. async load(signal) { + let turns: ConversationTurn[]; try { - const baseResult = await base.load(signal); - const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE); - if (extraTexts.length === 0) return baseResult; - const turns = await loadTurnsWithoutMalformedToolSequence( - baseResult.turns, - extraTexts, - ); - return { ...baseResult, turns }; + turns = await readLiveTurnsCached(signal); } catch (cause) { - log.warn( - "base context store load failed; recovering turns from disk segments", - { - cause: cause instanceof Error ? cause.message : String(cause), - }, + // Unrecoverable history: rethrow with the file name in the message. + throw new Error( + `failed to load ${TURNS_FILE}: ${ + cause instanceof Error ? cause.message : String(cause) + }`, + { cause }, ); - let baseTurns: ConversationTurn[]; - try { - // Parse segment 0 resiliently so orphan-tail heal still runs; - // mid-file garbage must not kill resume. - baseTurns = await readBaseTurnsFromDisk(dir); - } catch (parseCause) { - // Unrecoverable: rethrow with the file name in the message. - throw new Error( - `failed to load ${TURNS_FILE}: ${ - parseCause instanceof Error - ? parseCause.message - : String(parseCause) - }`, - { cause: parseCause }, - ); - } - const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE); - const turns = - extraTexts.length === 0 - ? baseTurns - : await loadTurnsWithoutMalformedToolSequence( - baseTurns, - extraTexts, - ); - const metadata = await loadMetadataSoft(() => base.loadMetadata()); - return { turns, ...metadata }; } + const metadata = await loadMetadataSoft(() => base.loadMetadata()); + return { turns, ...metadata }; }, setConnectorState: (state) => base.setConnectorState(state), branch: (name, signal) => base.branch(name, signal), log: (limit, signal) => base.log(limit, signal), async readAt(hash, signal) { const baseTurns = await base.readAt(hash, signal); - const extraNames = await extraSegmentNamesAtCommit(dir, hash); + const extraNames = await turnSegmentNamesAtCommit(dir, hash); if (extraNames.length === 0) return baseTurns; // Historical commits with orphans still in the tree may be malformed; - // prefer the longest well-formed prefix, with no on-disk effects. - const parsedExtras: ConversationTurn[][] = []; - for (const name of extraNames) { - const text = await blobTextAtCommit(dir, hash, name); - parsedExtras.push(parseSegmentTurns(text, false, name)); + // prefer the longest well-formed prefix, with no on-disk effects. The + // listed segment blobs read concurrently; the commit-time manifest + // replaces the per-read tree walk. + let parsedExtras: ConversationTurn[][]; + try { + parsedExtras = await Promise.all( + extraNames.map(async (name) => + parseSegmentTurns( + await blobTextAtCommit(dir, hash, name), + false, + name, + ), + ), + ); + } catch (cause) { + // A validated name may still be missing from the tree (torn or + // rewritten manifest); re-derive from the commit tree once before + // giving up, same degradation as an unreadable manifest blob. + log.warn( + "segment manifest at commit {hash} listed an unreadable segment; re-deriving from the commit tree", + { + hash, + cause: cause instanceof Error ? cause.message : String(cause), + }, + ); + const fallbackNames = await extraSegmentNamesAtCommit(dir, hash); + if (fallbackNames.length === 0) return baseTurns; + parsedExtras = await Promise.all( + fallbackNames.map(async (name) => + parseSegmentTurns( + await blobTextAtCommit(dir, hash, name), + false, + name, + ), + ), + ); } const keepExtras = longestWellFormedExtraCount(baseTurns, parsedExtras); if (keepExtras === 0) return baseTurns; @@ -1267,19 +1573,28 @@ export async function createSessionStores( promptToPublish === null ? null : await snapshotTurnSegments(dir, promptBaseName); + // One HEAD resolution per commit. The catch re-resolves on purpose: + // it must observe HEAD after the commit to detect a landed commit. + let headWithinLock: Promise | undefined; + const lockedHead = (): Promise => { + headWithinLock ??= headOid(dir); + return headWithinLock; + }; const headBefore = stagedRewrite === null && promptToPublish === null ? null - : await headOid(dir); + : await lockedHead(); let extraPaths: string[] = []; try { if (stagedRewrite !== null) { await writeSegmented(writeTurnsSegmented, stagedRewrite); + invalidateTurnsCache(); } if (promptToPublish !== null) { await writeSegmented(writePromptSegmented, promptToPublish); } + const listing = await surveyWorktree(dir); const toAdd: string[] = []; const toRemove: string[] = []; @@ -1287,20 +1602,61 @@ export async function createSessionStores( ...pendingSegmentPaths, ...pendingBlobFilepaths, ]) { - if (await pathExists(path.join(dir, filepath))) + if (await worktreeListingHas(dir, listing, filepath)) toAdd.push(filepath); else toRemove.push(filepath); } // Disk is the source of truth for which segments stay tracked after a - // rewrite or heal, even when pendingSegmentPaths was lost. - await reconcileSegmentStaging(dir, turnsBaseName, toAdd, toRemove); - await reconcileSegmentStaging(dir, promptBaseName, toAdd, toRemove); + // rewrite or heal, even when pendingSegmentPaths was lost. One + // shared worktree listing and one index listing serve every staged + // base name. + const tracked = await listIndexPaths(dir); + await reconcileSegmentStaging( + dir, + turnsBaseName, + toAdd, + toRemove, + listing, + tracked, + ); + await reconcileSegmentStaging( + dir, + promptBaseName, + toAdd, + toRemove, + listing, + tracked, + ); - if (await pathExists(path.join(dir, EVIDENCE_ARCHIVE_DIR))) { + if (listing.top.has(EVIDENCE_ARCHIVE_DIR)) { toAdd.push(EVIDENCE_ARCHIVE_DIR); } + // readAt reads this manifest instead of walking the commit tree. + // Desired content is recomputed from disk truth every commit, so a + // worktree manifest left behind by a failed commit self-corrects. + const contiguousTurnNames = contiguousSegmentNamesFromListing( + listing.top, + TURNS_FILE, + ); + const desiredManifest = JSON.stringify({ + turnSegments: contiguousTurnNames.slice(1), + }); + if ( + listing.top.has(SEGMENT_MANIFEST_FILE) || + desiredManifest !== EMPTY_SEGMENT_MANIFEST + ) { + const onDiskManifest = await readSegmentManifestText(dir); + if (onDiskManifest !== desiredManifest) { + await fs.promises.writeFile( + path.join(dir, SEGMENT_MANIFEST_FILE), + desiredManifest, + ); + } + toAdd.push(SEGMENT_MANIFEST_FILE); + } + // Empty managed checkpoints must not commit or stage session junk // (partial.jsonl). Discover extra segments first so a crash that // left them untracked cannot skip. A staged unpublished rewrite is @@ -1398,6 +1754,9 @@ export async function createSessionStores( turnsBaseName, ); liveTurnRefs = null; + // The restore rewrote the published generation; the cached + // live-turns parse predates it in general, so rebuild from disk. + invalidateTurnsCache(); } if (promptSnapshot !== null) { writePromptSegmented = createSegmentedJSONLWriter(