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

@@ -722,3 +722,110 @@ describe("createSessionStores", () => {
     expect(typeof audit.loadAudit).toBe("function");
   });
 });
+
+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

@@ -33,6 +33,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"]);
 
@@ -439,6 +440,33 @@ export async function createSessionStores(
   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>,
@@ -448,6 +476,39 @@ export async function createSessionStores(
     for (const filepath of modifiedPaths) pendingSegmentPaths.add(filepath);
   }
 
+  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 extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE);
+    const baseResult = await base.load();
+    const live =
+      extraTexts.length === 0
+        ? baseResult.turns
+        : await loadTurnsWithoutMalformedToolSequence(
+            baseResult.turns,
+            extraTexts,
+          );
+    if (live.length > 0 && contentPrefixLength(live, turns) < live.length) {
+      unpublishedRewrite = [...turns];
+      return;
+    }
+    await writeSegmented(writeTurnsSegmented, turns);
+    liveTurnRefs = [...turns];
+  }
+
   // Prefer the longest prefix of base + extras whose tool sequence the reactor
   // will accept. Orphan tails left by a fresh-writer compaction rewrite are
   // dropped and unlinked so the next load does not re-poison the session.
@@ -569,7 +630,7 @@ export async function createSessionStores(
     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),
@@ -580,6 +641,11 @@ export async function createSessionStores(
     },
     async commit(options, signal) {
       return withResolvedDirLock(dir, async () => {
+        if (unpublishedRewrite !== null) {
+          await writeSegmented(writeTurnsSegmented, unpublishedRewrite);
+          liveTurnRefs = unpublishedRewrite;
+          unpublishedRewrite = null;
+        }
         const toAdd: string[] = [];
         const toRemove: string[] = [];
 
@@ -596,6 +662,10 @@ export async function createSessionStores(
         await reconcileSegmentStaging(dir, TURNS_FILE, toAdd, toRemove);
         await reconcileSegmentStaging(dir, PROMPT_FILE, toAdd, toRemove);
 
+        if (await pathExists(path.join(dir, EVIDENCE_ARCHIVE_DIR))) {
+          toAdd.push(EVIDENCE_ARCHIVE_DIR);
+        }
+
         const add = extraCommitPaths([...new Set(toAdd)]);
         const remove = extraCommitPaths([...new Set(toRemove)]).filter(
           (p) => !add.includes(p),