Stage compaction rewrites until the store publishes

Persist blobs and stage writeTurns first. Replace reactor memory
only after commit so an interrupt resumes the old generation.

Author: TheGreatAxios

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

@@ -606,3 +606,110 @@ describe("createOptimizedContextStore checkpoint", () => {
     expect(await headIdent(nameOnlyWhitespaceEmail)).toEqual(HARNESS_IDENT);
   });
 });
+
+function turnTexts(turns: ConversationTurn[]): string[] {
+  return turns.map((t) => (t.content[0] as { text: string }).text);
+}
+
+async function gitLsTree(dir: string): Promise<string[]> {
+  const proc = Bun.spawn(["git", "-C", dir, "ls-tree", "-r", "--name-only", "HEAD"], {
+    stdout: "pipe",
+    stderr: "pipe",
+  });
+  const [exitCode, stdout, stderr] = await Promise.all([
+    proc.exited,
+    new Response(proc.stdout).text(),
+    new Response(proc.stderr).text(),
+  ]);
+  if (exitCode !== 0) {
+    throw new Error(`git ls-tree failed: ${stderr.trim() || stdout.trim()}`);
+  }
+  return stdout.split("\n").filter((line) => line.length > 0);
+}
+
+describe("createOptimizedContextStore unpublished rewrite", () => {
+  test("rewrite writeTurns stays off the live generation until commit", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    const original = [turn("keep-a"), turn("keep-b"), turn("drop-me")];
+    await store.writeTurns(original);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    await store.commit({ message: "published original" });
+
+    const compacted = [turn("[Compacted prior context]"), turn("keep-b")];
+    await store.writeTurns(compacted);
+
+    const loaded = await store.load();
+    expect(turnTexts(loaded.turns)).toEqual(["keep-a", "keep-b", "drop-me"]);
+
+    await store.commit({ message: "publish compact" });
+    const published = await store.load();
+    expect(turnTexts(published.turns)).toEqual(["[Compacted prior context]", "keep-b"]);
+  });
+
+  test("omitting commit leaves a new store on the old generation", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    await store.writeTurns([turn("old-a"), turn("old-b")]);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    await store.commit({ message: "old" });
+
+    await store.writeBlob(
+      "new-blob",
+      new TextEncoder().encode("needed-by-new-turns"),
+      "text/plain",
+    );
+    await store.writeTurns([turn("[Compacted prior context]")]);
+
+    const crashed = await createOptimizedContextStore(dir);
+    const loaded = await crashed.load();
+    expect(turnTexts(loaded.turns)).toEqual(["old-a", "old-b"]);
+  });
+
+  test("append writeTurns is still visible before commit", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    const first = turn("one");
+    await store.writeTurns([first]);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    await store.commit({ message: "one" });
+
+    await store.writeTurns([first, turn("two")]);
+    const loaded = await store.load();
+    expect(turnTexts(loaded.turns)).toEqual(["one", "two"]);
+  });
+
+  test("folds evidence-archive into the compact commit tree", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    await store.writeTurns([turn("old")]);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    await store.commit({ message: "old" });
+
+    const archiveDir = path.join(dir, "evidence-archive");
+    fs.mkdirSync(archiveDir, { recursive: true });
+    fs.writeFileSync(path.join(archiveDir, "index.jsonl"), "{}\n");
+    await store.writeTurns([turn("compacted")]);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    await store.commit({ message: "compact" });
+
+    expect(await gitLsTree(dir)).toContain("evidence-archive/index.jsonl");
+    const loaded = await store.load();
+    expect(turnTexts(loaded.turns)).toEqual(["compacted"]);
+  });
+
+  test("readAt of the old hash is not the load completeness path", async () => {
+    const dir = tempDir();
+    const store = await createOptimizedContextStore(dir);
+    await store.writeTurns([turn("era-1")]);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    const first = await store.commit({ message: "era-1" });
+
+    await store.writeTurns([turn("era-2")]);
+    await store.writeMetadata(EMPTY_CHECKPOINT_METADATA);
+    await store.commit({ message: "era-2" });
+
+    expect(turnTexts(await store.readAt(first.hash))).toEqual(["era-1"]);
+    expect(turnTexts((await store.load()).turns)).toEqual(["era-2"]);
+  });
+});

src/session/optimized-context-store.ts

@@ -26,6 +26,7 @@ const RESPONSE_FILE = "response.jsonl";
 const MANIFEST_FILE = "manifest.jsonl";
 const METADATA_FILE = "metadata.json";
 const TOOL_OUTPUT_DIR = "tool-output";
+const EVIDENCE_ARCHIVE_DIR = "evidence-archive";
 
 const log = getLogger([LOG_NAMESPACE_ROOT, "session", "context-store"]);
 
@@ -447,6 +448,30 @@ export async function createOptimizedContextStore(
   const pendingSegmentPaths = new Set<string>();
   const writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE);
   const writePromptSegmented = createSegmentedJSONLWriter(dir, PROMPT_FILE);
+  let liveTurnRefs: readonly ConversationTurn[] | null = null;
+  let unpublishedRewrite: ConversationTurn[] | null = null;
+
+  function refPrefixLength(
+    prev: readonly ConversationTurn[],
+    next: readonly ConversationTurn[],
+  ): number {
+    const max = Math.min(prev.length, next.length);
+    let prefix = 0;
+    while (prefix < max && prev[prefix] === next[prefix]) prefix++;
+    return prefix;
+  }
+
+  function contentPrefixLength(
+    prev: readonly ConversationTurn[],
+    next: readonly ConversationTurn[],
+  ): number {
+    const max = Math.min(prev.length, next.length);
+    let prefix = 0;
+    while (prefix < max && JSON.stringify(prev[prefix]) === JSON.stringify(next[prefix])) {
+      prefix++;
+    }
+    return prefix;
+  }
 
   async function writeSegmented(
     writer: ReturnType<typeof createSegmentedJSONLWriter>,
@@ -488,6 +513,70 @@ export async function createOptimizedContextStore(
     return [...baseTurns, ...parsedExtras.slice(0, keepExtras).flat()];
   }
 
+  async function loadLive(signal?: AbortSignal) {
+    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 };
+    } catch (cause) {
+      log.warn("base context store load failed; recovering turns from disk segments", {
+        cause: cause instanceof Error ? cause.message : String(cause),
+      });
+      let baseTurns: ConversationTurn[];
+      try {
+        // Prefer resilient parse of segment 0 alone so orphan-tail heal still runs.
+        // skipMalformed: mid-file garbage/interleaved records must not kill resume
+        // (CL-7052); null-pad stripping and torn-tail drop still apply.
+        const basePath = path.join(dir, TURNS_FILE);
+        if (await pathExists(basePath)) {
+          const text = await fs.promises.readFile(basePath, "utf-8");
+          baseTurns = parseSegmentTurns(text, true, TURNS_FILE, true);
+        } else {
+          baseTurns = [];
+        }
+      } catch (parseCause) {
+        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 };
+    }
+  }
+
+  async function writeTurnsLiveOrStage(turns: readonly ConversationTurn[]): Promise<void> {
+    if (unpublishedRewrite !== null) {
+      unpublishedRewrite = [...turns];
+      return;
+    }
+    if (liveTurnRefs !== null) {
+      if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) {
+        unpublishedRewrite = [...turns];
+        return;
+      }
+      await writeSegmented(writeTurnsSegmented, turns);
+      liveTurnRefs = [...turns];
+      return;
+    }
+    const live = (await loadLive()).turns;
+    if (live.length > 0 && contentPrefixLength(live, turns) < live.length) {
+      unpublishedRewrite = [...turns];
+      return;
+    }
+    await writeSegmented(writeTurnsSegmented, turns);
+    liveTurnRefs = [...turns];
+  }
+
   return {
     // Full-history read. Called by the reactor during initialization, where
     // the complete turn history is the actual live conversation state, not an
@@ -498,47 +587,7 @@ export async function createOptimizedContextStore(
     // garbage turns.jsonl), 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 on a bare Bun JSON token or wipe pending ops.
-    async load(signal) {
-      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 };
-      } catch (cause) {
-        log.warn("base context store load failed; recovering turns from disk segments", {
-          cause: cause instanceof Error ? cause.message : String(cause),
-        });
-        let baseTurns: ConversationTurn[];
-        try {
-          // Prefer resilient parse of segment 0 alone so orphan-tail heal still runs.
-          // skipMalformed: mid-file garbage/interleaved records must not kill resume
-          // (CL-7052); null-pad stripping and torn-tail drop still apply.
-          const basePath = path.join(dir, TURNS_FILE);
-          if (await pathExists(basePath)) {
-            const text = await fs.promises.readFile(basePath, "utf-8");
-            baseTurns = parseSegmentTurns(text, true, TURNS_FILE, true);
-          } else {
-            baseTurns = [];
-          }
-        } 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 };
-      }
-    },
+    load: (signal) => loadLive(signal),
     setConnectorState: (state) => base.setConnectorState(state),
     branch: (name, signal) => base.branch(name, signal),
     log: (limit, signal) => base.log(limit, signal),
@@ -562,7 +611,7 @@ export async function createOptimizedContextStore(
     writePrompt: (turns) => writeSegmented(writePromptSegmented, turns),
     writeResponse: (turn, signal) => base.writeResponse(turn, signal),
     writeManifest: (records, signal) => base.writeManifest(records, signal),
-    writeTurns: (turns) => writeSegmented(writeTurnsSegmented, turns),
+    writeTurns: (turns) => writeTurnsLiveOrStage(turns),
     writeMetadata: (metadata, signal) => base.writeMetadata(metadata, signal),
     readManifestHistory: (limit, signal) => base.readManifestHistory(limit, signal),
     async writeBlob(key, bytes, contentType, signal) {
@@ -571,6 +620,12 @@ export async function createOptimizedContextStore(
       pendingBlobFilepaths.add(`${TOOL_OUTPUT_DIR}/${filename}`);
     },
     async commit(options, _signal) {
+      if (unpublishedRewrite !== null) {
+        await writeSegmented(writeTurnsSegmented, unpublishedRewrite);
+        liveTurnRefs = unpublishedRewrite;
+        unpublishedRewrite = null;
+      }
+
       const toAdd: string[] = [];
       const toRemove: string[] = [];
 
@@ -584,6 +639,10 @@ export async function createOptimizedContextStore(
         else toRemove.push(filepath);
       }
 
+      if (await pathExists(path.join(dir, EVIDENCE_ARCHIVE_DIR))) {
+        toAdd.push(EVIDENCE_ARCHIVE_DIR);
+      }
+
       // Disk is source of truth for which turn/prompt segments should remain
       // tracked after a rewrite or heal, even if pendingSegmentPaths was lost.
       await reconcileSegmentStaging(dir, TURNS_FILE, toAdd, toRemove);