From f5b32f21a991cfaea386b4ed0d102230d66f2d4a Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 17:10:18 -0700 Subject: [PATCH 1/4] fix(session): keep foreign transcripts off the shared log --- src/session/optimized-context-store.test.ts | 415 ++++++++++++++++++++ src/session/optimized-context-store.ts | 277 +++++++++++-- 2 files changed, 655 insertions(+), 37 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index f118421c5..b3c6fb12a 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -744,6 +744,376 @@ describe("concurrent sessions write disjoint transcripts", () => { ]); }); + test("markerless foreign turns stay off the shared log", async () => { + const dir = tempDir(); + const sessionA = await createOptimizedContextStore(dir, { + ownerId: "session-A", + }); + const turnsA = [ + turn("a-0"), + turn("a-1"), + turn("a-2"), + turn("a-3"), + turn("a-4"), + ]; + await sessionA.writeTurns([...turnsA]); + await sessionA.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionA.commit({ message: "session A" }); + fs.unlinkSync(path.join(dir, "session-owner.json")); + + const sessionB = await createOptimizedContextStore(dir, { + ownerId: "session-B", + }); + const turnsB = [turn("b-0"), turn("b-1"), turn("b-2")]; + await sessionB.writeTurns([...turnsB]); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(turnsA), + ); + expect(versionedTurnsFiles(dir)).toEqual([]); + + await sessionB.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionB.commit({ message: "session B" }); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(turnsA), + ); + const versioned = versionedTurnsFiles(dir); + expect(versioned).toHaveLength(1); + expect( + turnTexts(readTurnsFromFile(path.join(dir, defined(versioned[0])))), + ).toEqual(turnTexts(turnsB)); + }); + + test("markerless resume can still extend the shared log", async () => { + const dir = tempDir(); + const sessionA = await createOptimizedContextStore(dir); + const turnsA = [turn("a"), turn("b")]; + await sessionA.writeTurns([...turnsA]); + await sessionA.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionA.commit({ message: "session A" }); + fs.unlinkSync(path.join(dir, "session-owner.json")); + + const resumed = await createOptimizedContextStore(dir); + const next = [...turnsA, turn("c")]; + await resumed.writeTurns([...next]); + await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await resumed.commit({ message: "resume" }); + expect(versionedTurnsFiles(dir)).toEqual([]); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(next), + ); + }); + + test("compaction without an owner id stays on the live transcript", async () => { + const dir = tempDir(); + const first = await createOptimizedContextStore(dir); + const original = [turn("a"), turn("b"), turn("c"), turn("d")]; + await first.writeTurns([...original]); + await first.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await first.commit({ message: "original" }); + + const second = await createOptimizedContextStore(dir); + const compacted = [turn("[Compacted prior context]"), turn("c"), turn("d")]; + await second.writeTurns([...compacted]); + await second.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await second.commit({ message: "compact" }); + + expect(versionedTurnsFiles(dir)).toEqual([]); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(compacted), + ); + expect(turnTexts((await second.load()).turns)).toEqual( + turnTexts(compacted), + ); + expect(turnTexts((await loadRecentTurns(dir, 10)).turns)).toEqual( + turnTexts(compacted), + ); + }); + + test("a foreign transcript without an owner id still diverts", async () => { + const dir = tempDir(); + const first = await createOptimizedContextStore(dir); + const original = [turn("a"), turn("b"), turn("c"), turn("d")]; + await first.writeTurns([...original]); + await first.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await first.commit({ message: "original" }); + + const second = await createOptimizedContextStore(dir); + const foreign = [turn("x"), turn("y")]; + await second.writeTurns([...foreign]); + await second.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await second.commit({ message: "foreign" }); + + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(original), + ); + const versioned = versionedTurnsFiles(dir); + expect(versioned).toHaveLength(1); + expect( + turnTexts(readTurnsFromFile(path.join(dir, defined(versioned[0])))), + ).toEqual(turnTexts(foreign)); + expect(turnTexts((await loadRecentTurns(dir, 10)).turns)).toEqual( + turnTexts(original), + ); + }); + + test("an owner id can compact a markerless transcript in place", async () => { + const dir = tempDir(); + const legacy = await createOptimizedContextStore(dir); + const original = [turn("a"), turn("b"), turn("c"), turn("d")]; + await legacy.writeTurns([...original]); + await legacy.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await legacy.commit({ message: "legacy" }); + fs.unlinkSync(path.join(dir, "session-owner.json")); + + const upgraded = await createOptimizedContextStore(dir, { + ownerId: "session-S", + }); + const compacted = [turn("[Compacted prior context]"), turn("c"), turn("d")]; + await upgraded.writeTurns([...compacted]); + await upgraded.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await upgraded.commit({ message: "upgrade compact" }); + + expect(versionedTurnsFiles(dir)).toEqual([]); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(compacted), + ); + expect(turnTexts((await upgraded.load()).turns)).toEqual( + turnTexts(compacted), + ); + }); + + test("a foreign prefix extension is kept when the owner compacts", async () => { + const dir = tempDir(); + const sessionA = await createOptimizedContextStore(dir, { + ownerId: "session-A", + }); + await sessionA.writeTurns([turn("a"), turn("b")]); + await sessionA.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionA.commit({ message: "session A" }); + + const sessionB = await createOptimizedContextStore(dir, { + ownerId: "session-B", + }); + await sessionB.writeTurns([turn("a"), turn("b"), turn("from-B")]); + await sessionB.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionB.commit({ message: "session B" }); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual([ + "a", + "b", + ]); + + const sessionA2 = await createOptimizedContextStore(dir, { + ownerId: "session-A", + }); + await sessionA2.writeTurns([turn("[Compacted prior context]"), turn("b")]); + await sessionA2.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionA2.commit({ message: "compact" }); + + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual([ + "[Compacted prior context]", + "b", + ]); + const preserved = versionedTurnsFiles(dir).flatMap((name) => + turnTexts(readTurnsFromFile(path.join(dir, name))), + ); + expect(preserved).toContain("from-B"); + }); + + test("a foreign write after an empty owned file is not discarded", async () => { + const dir = tempDir(); + const sessionA = await createOptimizedContextStore(dir, { + ownerId: "session-A", + }); + await sessionA.writeTurns([turn("seed")]); + fs.unlinkSync(path.join(dir, TURNS_FILE)); + + const sessionB = await createOptimizedContextStore(dir, { + ownerId: "session-B", + }); + await sessionB.writeTurns([turn("a"), turn("b"), turn("from-B")]); + await sessionB.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionB.commit({ message: "session B" }); + + const resumed = await createOptimizedContextStore(dir, { + ownerId: "session-A", + }); + await resumed.writeTurns([turn("[Compacted prior context]"), turn("b")]); + await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await resumed.commit({ message: "compact" }); + + const preserved = versionedTurnsFiles(dir).flatMap((name) => + turnTexts(readTurnsFromFile(path.join(dir, name))), + ); + expect(preserved).toContain("from-B"); + expect(turnTexts((await resumed.load()).turns)).not.toContain("from-B"); + }); + + test("a prompt is not written to the shared file before ownership is decided", async () => { + const dir = tempDir(); + const promptFile = "prompt.jsonl"; + const sessionA = await createOptimizedContextStore(dir, { + ownerId: "session-A", + }); + await sessionA.writeTurns([turn("a"), turn("b")]); + await sessionA.writePrompt([turn("a"), turn("b"), turn("ephemeral-A")]); + await sessionA.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionA.commit({ message: "session A" }); + const promptPath = path.join(dir, promptFile); + const ownedPrompt = fs.readFileSync(promptPath, "utf8"); + + const sessionB = await createOptimizedContextStore(dir, { + ownerId: "session-B", + }); + await sessionB.writePrompt([turn("foreign-prompt")]); + expect(fs.readFileSync(promptPath, "utf8")).toBe(ownedPrompt); + await sessionB.writeTurns([turn("x"), turn("y")]); + expect(fs.readFileSync(promptPath, "utf8")).toBe(ownedPrompt); + await sessionB.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await sessionB.commit({ message: "session B" }); + + expect(fs.readFileSync(promptPath, "utf8")).toBe(ownedPrompt); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual([ + "a", + "b", + ]); + const versionedPrompts = fs + .readdirSync(dir) + .filter((name) => /^prompt-\d+-[0-9a-f]+\.jsonl$/.test(name)); + expect(versionedPrompts).toHaveLength(1); + expect( + fs.readFileSync(path.join(dir, defined(versionedPrompts[0])), "utf8"), + ).toContain("foreign-prompt"); + }); + + test( + "two processes cannot both claim an empty session directory", + async () => { + const workerPath = path.join( + os.tmpdir(), + `claim-worker-${process.pid}.ts`, + ); + const storeModule = path.join( + import.meta.dir, + "optimized-context-store.ts", + ); + fs.writeFileSync( + workerPath, + `import fs from "node:fs"; +import { createOptimizedContextStore } from ${JSON.stringify(storeModule)}; + +const dir = process.env.CLAIM_DIR; +const ownerId = process.env.CLAIM_OWNER; +const label = process.env.CLAIM_LABEL; +if (dir === undefined || ownerId === undefined || label === undefined) { + throw new Error("missing claim env"); +} +const turn = (text) => ({ role: "user", content: [{ type: "text", text }], timestamp: 1 }); +await fs.promises.writeFile(dir + "/ready-" + ownerId, ""); +while (!fs.existsSync(dir + "/go")) { + await Bun.sleep(5); +} +const store = await createOptimizedContextStore(dir, { ownerId }); +const turns = Array.from({ length: 30 }, (_, i) => turn(label + "-" + i)); +await store.writeTurns(turns); +await store.writeMetadata({ + pendingOperations: [], + tokenUsage: { input: 0, output: 0, cacheRead: 0, cacheWrite: 0, thinking: 0 }, +}); +const lockPath = dir + "/commit.lock"; +for (;;) { + try { + const handle = await fs.promises.open(lockPath, "wx"); + await handle.close(); + break; + } catch (cause) { + if (cause instanceof Error && "code" in cause && cause.code === "EEXIST") { + await Bun.sleep(10); + continue; + } + throw cause; + } +} +try { + await store.commit({ message: label }); +} finally { + await fs.promises.unlink(lockPath).catch(() => undefined); +} +`, + ); + + try { + for (let attempt = 0; attempt < 6; attempt++) { + const dir = tempDir(); + const seed = await createOptimizedContextStore(dir, { + ownerId: "seed", + }); + await seed.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await seed.commit({ message: "init" }); + + const spawnClaim = (ownerId: string, label: string) => + Bun.spawn(["bun", workerPath], { + cwd: path.resolve(import.meta.dir, "../.."), + env: { + ...process.env, + CLAIM_DIR: dir, + CLAIM_OWNER: ownerId, + CLAIM_LABEL: label, + }, + stdout: "pipe", + stderr: "pipe", + }); + const procA = spawnClaim("owner-A", "A"); + const procB = spawnClaim("owner-B", "B"); + const readyA = path.join(dir, "ready-owner-A"); + const readyB = path.join(dir, "ready-owner-B"); + const deadline = Date.now() + 15_000; + while ( + (!fs.existsSync(readyA) || !fs.existsSync(readyB)) && + Date.now() < deadline + ) { + await Bun.sleep(5); + } + fs.writeFileSync(path.join(dir, "go"), ""); + const [codeA, codeB, errA, errB] = await Promise.all([ + procA.exited, + procB.exited, + new Response(procA.stderr).text(), + new Response(procB.stderr).text(), + ]); + if (codeA !== 0 || codeB !== 0) { + throw new Error( + `claim workers failed (${codeA}, ${codeB})\n${errA}\n${errB}`, + ); + } + const shared = turnTexts( + readTurnsFromFile(path.join(dir, TURNS_FILE)), + ); + const versioned = versionedTurnsFiles(dir); + expect(versioned).toHaveLength(1); + const other = turnTexts( + readTurnsFromFile(path.join(dir, defined(versioned[0]))), + ); + const sharedLabel = shared[0]?.split("-")[0]; + const otherLabel = other[0]?.split("-")[0]; + expect(new Set([sharedLabel, otherLabel])).toEqual( + new Set(["A", "B"]), + ); + expect(shared).toHaveLength(30); + expect(other).toHaveLength(30); + expect( + shared.every((text) => text.startsWith(`${sharedLabel}-`)), + ).toBe(true); + expect(other.every((text) => text.startsWith(`${otherLabel}-`))).toBe( + true, + ); + } + } finally { + fs.unlinkSync(workerPath); + } + }, + { timeout: 60_000 }, + ); + test("versioned base names are unique even within the same millisecond", () => { const names = Array.from({ length: 100 }, () => versionedBaseName(TURNS_FILE), @@ -874,6 +1244,51 @@ describe("createOptimizedContextStore unpublished rewrite", () => { ]); }); + test("a failed commit does not publish a same-owner compaction", async () => { + const dir = tempDir(); + const owner = await createOptimizedContextStore(dir, { + ownerId: "session-S", + }); + const original = [turn("keep-a"), turn("keep-b"), turn("drop-me")]; + await owner.writeTurns([...original]); + await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA); + const published = await owner.commit({ message: "published original" }); + + const resumed = await createOptimizedContextStore(dir, { + ownerId: "session-S", + }); + await resumed.writeTurns([ + turn("[Compacted prior context]"), + turn("keep-b"), + ]); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(original), + ); + + const objectsDir = path.join(dir, ".git", "objects"); + const objectsHeld = `${objectsDir}.held`; + fs.renameSync(objectsDir, objectsHeld); + fs.writeFileSync(objectsDir, "held"); + try { + await expect( + resumed.commit({ message: "publish compact" }), + ).rejects.toThrow(); + } finally { + fs.unlinkSync(objectsDir); + fs.renameSync(objectsHeld, objectsDir); + } + + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(original), + ); + expect(turnTexts((await resumed.load()).turns)).toEqual( + turnTexts(original), + ); + expect(turnTexts(await resumed.readAt(published.hash))).toEqual( + turnTexts(original), + ); + }); + test("append writeTurns is still visible before commit", async () => { const dir = tempDir(); const store = await createOptimizedContextStore(dir); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 011e624af..53226a150 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -376,14 +376,82 @@ async function readOwnerMarker( } } -async function writeOwnerMarker(dir: string, ownerId: string): Promise { +async function tryCreateOwnerMarker( + dir: string, + ownerId: string, +): Promise { const full = path.join(dir, OWNER_MARKER_FILE); - const tmp = `${full}.${process.pid}.${Math.random().toString(16).slice(2)}.tmp`; - await fs.promises.writeFile( - tmp, - JSON.stringify({ ownerId, createdAt: Date.now() }), + try { + await fs.promises.writeFile( + full, + JSON.stringify({ ownerId, createdAt: Date.now() }), + { encoding: "utf8", flag: "wx" }, + ); + return true; + } catch (cause) { + if (cause instanceof Error && "code" in cause && cause.code === "EEXIST") { + return false; + } + throw cause; + } +} + +const COMPACTED_TURN_PREFIX = "[Compacted prior context]"; + +function firstTurnText(turn: ConversationTurn): string | undefined { + for (const block of turn.content) { + if (block.type === "text") return block.text; + } + return undefined; +} + +function isCompactionMarker(turn: ConversationTurn): boolean { + const text = firstTurnText(turn); + if (text === undefined) return false; + return ( + text.startsWith(COMPACTED_TURN_PREFIX) || + text === "[compact]" || + text === "[compaction]" ); - await fs.promises.rename(tmp, full); +} + +function sameTurnContent(a: ConversationTurn, b: ConversationTurn): boolean { + return JSON.stringify(a) === JSON.stringify(b); +} + +function isCompactionOf( + live: readonly ConversationTurn[], + next: readonly ConversationTurn[], +): boolean { + if (live.length === 0 || next.length === 0) return false; + let head = 0; + while (head < next.length) { + const turn = next[head]; + if (turn === undefined || !isCompactionMarker(turn)) break; + head++; + } + if (head === 0) return false; + const rest = next.slice(head); + if (rest.length === 0) return true; + const max = Math.min(live.length, rest.length); + for (let kept = max; kept >= 1; kept--) { + const start = live.length - kept; + let matched = true; + for (let index = 0; index < kept; index++) { + const liveTurn = live[start + index]; + const restTurn = rest[index]; + if ( + liveTurn === undefined || + restTurn === undefined || + !sameTurnContent(liveTurn, restTurn) + ) { + matched = false; + break; + } + } + if (matched) return true; + } + return false; } async function listIndexPaths(dir: string): Promise> { @@ -555,8 +623,8 @@ export interface SessionStores { export interface SessionStoreOptions { signer?: CommitSigner; /** - * Same id shares the marker; omit for a random token, foreign unless the - * transcript extends history. + * Same id shares the marker. Omit for a random token: a prefix extension or + * a compaction of the on-disk transcript stays on the shared log. */ ownerId?: string; } @@ -576,6 +644,7 @@ export async function createSessionStores( const pendingBlobFilepaths = new Set(); const pendingSegmentPaths = new Set(); const ownerToken = opts?.ownerId ?? globalThis.crypto.randomUUID(); + const explicitOwnerId = opts?.ownerId; let turnsBaseName = TURNS_FILE; let promptBaseName = PROMPT_FILE; let versionedBase: string | null = null; @@ -583,6 +652,8 @@ export async function createSessionStores( let writePromptSegmented = createSegmentedJSONLWriter(dir, PROMPT_FILE); let liveTurnRefs: readonly ConversationTurn[] | null = null; let unpublishedRewrite: ConversationTurn[] | null = null; + let pendingPrompt: ConversationTurn[] | null = null; + let stagedPrompt: ConversationTurn[] | null = null; function divertToVersionedBases(): void { if (versionedBase !== null) return; @@ -626,6 +697,81 @@ 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(); + } + + async function resolveFirstWrite( + turns: readonly ConversationTurn[], + ): Promise<"append" | "rewrite" | "divert"> { + for (let attempt = 0; attempt < 3; attempt++) { + const marker = await readOwnerMarker(dir); + if (marker !== null && marker.ownerId === ownerToken) { + const live = await readLiveTurns(); + return contentPrefixLength(live, turns) === live.length + ? "append" + : "rewrite"; + } + if ( + marker !== null && + explicitOwnerId !== undefined && + marker.ownerId !== ownerToken + ) { + return "divert"; + } + const live = await readLiveTurns(); + if (marker !== null && live.length === 0) return "divert"; + if ( + live.length === 0 || + contentPrefixLength(live, turns) === live.length + ) { + if (marker === null && !(await tryCreateOwnerMarker(dir, ownerToken))) { + continue; + } + return "append"; + } + if (isCompactionOf(live, turns)) { + if (marker === null && !(await tryCreateOwnerMarker(dir, ownerToken))) { + continue; + } + return "rewrite"; + } + return "divert"; + } + return "divert"; + } + + async function queuePrompt( + turns: readonly ConversationTurn[], + ): Promise { + if (versionedBase !== null && liveTurnRefs === null) { + stagedPrompt = [...turns]; + pendingPrompt = null; + return; + } + if ( + liveTurnRefs === null && + unpublishedRewrite === null && + versionedBase === null + ) { + pendingPrompt = [...turns]; + return; + } + await writePromptIfDiffered(turns); + } + + async function flushPendingPrompt(): Promise { + if (pendingPrompt === null) return; + const prompt = pendingPrompt; + pendingPrompt = null; + await writePromptIfDiffered(prompt); + } + async function writeTurnsLiveOrStage( turns: readonly ConversationTurn[], ): Promise { @@ -642,8 +788,11 @@ export async function createSessionStores( liveTurnRefs = [...turns]; return; } - // First claim is per-dir so two fresh stores cannot both own the base. await withResolvedDirLock(dir, async () => { + if (unpublishedRewrite !== null) { + unpublishedRewrite = [...turns]; + return; + } if (liveTurnRefs !== null) { if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) { unpublishedRewrite = [...turns]; @@ -653,29 +802,24 @@ export async function createSessionStores( liveTurnRefs = [...turns]; return; } - const marker = await readOwnerMarker(dir); - if (marker === null) { - await writeOwnerMarker(dir, ownerToken); - await writeSegmented(writeTurnsSegmented, turns); - liveTurnRefs = [...turns]; - return; - } - const live = await loadLiveTurnsFromDisk(); - if (marker.ownerId === ownerToken) { - // Matching owner keeps compaction on the shared base, not a divert. - await writeSegmented(writeTurnsSegmented, turns); - liveTurnRefs = [...turns]; + const decision = await resolveFirstWrite(turns); + if (decision === "divert") { + divertToVersionedBases(); + unpublishedRewrite = [...turns]; + if (pendingPrompt !== null) { + stagedPrompt = pendingPrompt; + pendingPrompt = null; + } return; } - if (live.length > 0 && contentPrefixLength(live, turns) < live.length) { - // Stage until commit so a crash leaves no orphaned versioned file. - divertToVersionedBases(); + if (decision === "rewrite") { unpublishedRewrite = [...turns]; + await flushPendingPrompt(); return; } - // Prefix extension with no stable id resumes the shared base. await writeSegmented(writeTurnsSegmented, turns); liveTurnRefs = [...turns]; + await flushPendingPrompt(); }); } @@ -879,7 +1023,7 @@ export async function createSessionStores( readBlob: (key, signal) => base.readBlob(key, signal), // Skipped identical snapshots fall back to live turns; load() never reads // prompt.jsonl. - writePrompt: (turns) => writePromptIfDiffered(turns), + writePrompt: (turns) => queuePrompt(turns), writeResponse: (turn, signal) => base.writeResponse(turn, signal), writeManifest: (records, signal) => base.writeManifest(records, signal), writeTurns: (turns) => writeTurnsLiveOrStage(turns), @@ -893,7 +1037,30 @@ export async function createSessionStores( }, async commit(options, signal) { return withResolvedDirLock(dir, async () => { + if ( + pendingPrompt !== null && + versionedBase === null && + unpublishedRewrite === null && + liveTurnRefs === null + ) { + const marker = await readOwnerMarker(dir); + if ( + marker !== null && + explicitOwnerId !== undefined && + marker.ownerId !== ownerToken + ) { + divertToVersionedBases(); + stagedPrompt = pendingPrompt; + pendingPrompt = null; + } else { + const prompt = pendingPrompt; + pendingPrompt = null; + await writePromptIfDiffered(prompt); + } + } + const stagedRewrite = unpublishedRewrite; + const promptToPublish = stagedPrompt; // The staged rewrite lands on the working-tree segments before the git // operations below; snapshot them so a failed commit can restore the // published generation. The rewrite stays staged so a retry can still @@ -902,13 +1069,23 @@ export async function createSessionStores( stagedRewrite === null ? null : await snapshotTurnSegments(dir, turnsBaseName); - const headBefore = stagedRewrite === null ? null : await headOid(dir); + const promptSnapshot = + promptToPublish === null + ? null + : await snapshotTurnSegments(dir, promptBaseName); + const headBefore = + stagedRewrite === null && promptToPublish === null + ? null + : await headOid(dir); let extraPaths: string[] = []; try { if (stagedRewrite !== null) { await writeSegmented(writeTurnsSegmented, stagedRewrite); } + if (promptToPublish !== null) { + await writeSegmented(writePromptSegmented, promptToPublish); + } const toAdd: string[] = []; const toRemove: string[] = []; @@ -978,10 +1155,11 @@ export async function createSessionStores( liveTurnRefs = stagedRewrite; unpublishedRewrite = null; } + if (promptToPublish !== null) stagedPrompt = null; return committed; } catch (cause) { await resetIndexPaths(dir, extraPaths); - if (segmentSnapshot !== null) { + if (segmentSnapshot !== null || promptSnapshot !== null) { // Restore the published generation unless the commit actually // landed despite throwing (a ref write or post-commit check can // fail after HEAD moved), in which case the on-disk rewrite @@ -990,19 +1168,44 @@ export async function createSessionStores( const landed = headBefore !== null && headNow !== null && headNow !== headBefore; if (!landed) { - try { - await restoreTurnSegments(dir, segmentSnapshot, turnsBaseName); - } catch { - // A partial restore must not mask the real commit error. + if (segmentSnapshot !== null) { + try { + await restoreTurnSegments( + dir, + segmentSnapshot, + turnsBaseName, + ); + } catch { + // A partial restore must not mask the real commit error. + } + } + if (promptSnapshot !== null) { + try { + await restoreTurnSegments( + dir, + promptSnapshot, + promptBaseName, + ); + } catch { + // A partial restore must not mask the real commit error. + } } } // Drop the writer's stale in-memory state so a retry rewrites from // scratch. - writeTurnsSegmented = createSegmentedJSONLWriter( - dir, - turnsBaseName, - ); - liveTurnRefs = null; + if (segmentSnapshot !== null) { + writeTurnsSegmented = createSegmentedJSONLWriter( + dir, + turnsBaseName, + ); + liveTurnRefs = null; + } + if (promptSnapshot !== null) { + writePromptSegmented = createSegmentedJSONLWriter( + dir, + promptBaseName, + ); + } } throw cause; } From 81b84b8d26629845963eec3086bdc6f1be7ae914 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 17:29:30 -0700 Subject: [PATCH 2/4] fix(session): keep pruned compactions on the shared log --- src/session/optimized-context-store.test.ts | 97 ++++++++ src/session/optimized-context-store.ts | 243 ++++++++++++++++++-- 2 files changed, 315 insertions(+), 25 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index b3c6fb12a..d4e99cae1 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -12,6 +12,7 @@ import { versionedBaseName, } from "./optimized-context-store.js"; import { segmentFileName, listSegmentFiles } from "./incremental-jsonl.js"; +import { createPruningCompactor } from "./compactor.js"; const TURNS_FILE = "turns.jsonl"; @@ -856,6 +857,102 @@ describe("concurrent sessions write disjoint transcripts", () => { ); }); + test("a pruned compaction stays on the live transcript", async () => { + const readCall = (id: string): ConversationTurn => ({ + role: "assistant", + content: [ + { + type: "tool_call", + id, + name: "read_file", + arguments: { path: "src/hot.ts" }, + }, + ], + timestamp: 1, + }); + const readResult = (id: string, text: string): ConversationTurn => ({ + role: "user", + content: [ + { + type: "tool_result", + callId: id, + content: [{ type: "text", text }], + }, + ], + timestamp: 1, + }); + const original: ConversationTurn[] = [ + turn("start the migration and keep the auth notes"), + turn("constraint: do not rewrite the session log"), + readCall("old-a"), + readResult("old-a", `FOLD_${"c".repeat(8000)}`), + readCall("old-b"), + readResult("old-b", `FOLD_${"d".repeat(8000)}`), + readCall("old-c"), + readResult("old-c", `FOLD_${"e".repeat(8000)}`), + readCall("r1"), + readResult("r1", `OLD_${"a".repeat(3000)}`), + readCall("r2"), + readResult("r2", `NEW_${"b".repeat(3000)}`), + turn("continue from the latest read"), + ]; + const folded = (await createPruningCompactor().apply(original, {} as never)) + .output; + const unequal = folded.filter( + (candidate) => + !original.some( + (source) => JSON.stringify(source) === JSON.stringify(candidate), + ), + ); + expect(unequal.length).toBeGreaterThan(0); + const foldedText = JSON.stringify(folded); + expect(foldedText).toContain("[tail-shortened "); + expect(foldedText).toContain("omitted from context"); + expect( + folded.some((candidate) => + candidate.content.some( + (block) => + block.type === "text" && + block.text.startsWith("[Compacted prior context]"), + ), + ), + ).toBe(true); + + const publish = async ( + ownerId: string | undefined, + dropMarker: boolean, + ): Promise => { + const dir = tempDir(); + const first = await createOptimizedContextStore(dir); + await first.writeTurns([...original]); + await first.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await first.commit({ message: "original" }); + if (dropMarker) fs.unlinkSync(path.join(dir, "session-owner.json")); + const second = await createOptimizedContextStore( + dir, + ownerId === undefined ? undefined : { ownerId }, + ); + await second.writeTurns(folded); + await second.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await second.commit({ message: "fold" }); + const later = [...folded, turn("after-fold")]; + await second.writeTurns(later); + await second.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await second.commit({ message: "later" }); + + expect(versionedTurnsFiles(dir)).toEqual([]); + expect( + JSON.stringify(readTurnsFromFile(path.join(dir, TURNS_FILE))), + ).toBe(JSON.stringify(later)); + expect(JSON.stringify((await second.load()).turns)).toBe( + JSON.stringify(later), + ); + }; + + await publish(undefined, false); + await publish("session-S", true); + }); + test("an owner id can compact a markerless transcript in place", async () => { const dir = tempDir(); const legacy = await createOptimizedContextStore(dir); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 53226a150..29671e79a 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -397,6 +397,9 @@ async function tryCreateOwnerMarker( } const COMPACTED_TURN_PREFIX = "[Compacted prior context]"; +const THINKING_ONLY_OMITTED = "[thinking-only turn omitted]"; +const EXCERPT_MARKER = "\n[tail-shortened "; +const EXCERPT_CLOSE = "full text remains in the archived transcript]"; function firstTurnText(turn: ConversationTurn): string | undefined { for (const block of turn.content) { @@ -415,43 +418,233 @@ function isCompactionMarker(turn: ConversationTurn): boolean { ); } +function isSummaryTurn(turn: ConversationTurn): boolean { + const text = firstTurnText(turn); + return text !== undefined && text.startsWith(COMPACTED_TURN_PREFIX); +} + function sameTurnContent(a: ConversationTurn, b: ConversationTurn): boolean { return JSON.stringify(a) === JSON.stringify(b); } -function isCompactionOf( - live: readonly ConversationTurn[], - next: readonly ConversationTurn[], +function joinedBlockText( + content: readonly { type: string; text?: string }[], +): string { + let text = ""; + for (const part of content) { + if (part.type === "text" && part.text !== undefined) text += part.text; + } + return text; +} + +function isExcerptText(original: string, excerpted: string): boolean { + const at = excerpted.indexOf(EXCERPT_MARKER); + if (at < 0) return false; + const head = excerpted.slice(0, at); + if (head.length === 0 || !original.startsWith(head)) return false; + const close = excerpted.indexOf(EXCERPT_CLOSE); + if (close < 0) return false; + let tail = excerpted.slice(close + EXCERPT_CLOSE.length); + if (tail.startsWith("\n")) tail = tail.slice(1); + if (tail.length === 0) return head.length < original.length; + return original.endsWith(tail) && head.length + tail.length < original.length; +} + +function isOmissionStub(stub: string): boolean { + return ( + /\[.+ — \d+ chars omitted from context; source unchanged\./.test(stub) || + /\[.+ — \d+ chars, omitted\]/.test(stub) + ); +} + +function toolCallIds(turn: ConversationTurn): string[] { + const ids: string[] = []; + for (const block of turn.content) { + if (block.type === "tool_call") ids.push(block.id); + } + return ids; +} + +function resultCallIds(turn: ConversationTurn): string[] { + const ids: string[] = []; + for (const block of turn.content) { + if (block.type === "tool_result") ids.push(block.callId); + } + return ids; +} + +function sameIdList( + left: readonly string[], + right: readonly string[], ): boolean { - if (live.length === 0 || next.length === 0) return false; - let head = 0; - while (head < next.length) { - const turn = next[head]; - if (turn === undefined || !isCompactionMarker(turn)) break; - head++; + if (left.length !== right.length) return false; + for (let index = 0; index < left.length; index++) { + if (left[index] !== right[index]) return false; } - if (head === 0) return false; - const rest = next.slice(head); - if (rest.length === 0) return true; - const max = Math.min(live.length, rest.length); - for (let kept = max; kept >= 1; kept--) { - const start = live.length - kept; - let matched = true; - for (let index = 0; index < kept; index++) { - const liveTurn = live[start + index]; - const restTurn = rest[index]; + return true; +} + +function resultProjects( + source: Extract, + out: Extract, +): boolean { + if (source.callId !== out.callId) return false; + if ((source.isError === true) !== (out.isError === true)) return false; + if (JSON.stringify(source.content) === JSON.stringify(out.content)) + return true; + const sourceText = joinedBlockText(source.content); + const outText = joinedBlockText(out.content); + const stub = out.content[0]; + if ( + out.content.length === 1 && + stub !== undefined && + stub.type === "text" && + isOmissionStub(outText) + ) { + return sourceText.length > 0; + } + if (source.content.length !== out.content.length) return false; + for (let index = 0; index < source.content.length; index++) { + const sourcePart = source.content[index]; + const outPart = out.content[index]; + if (sourcePart === undefined || outPart === undefined) return false; + if (sourcePart.type !== outPart.type) return false; + if (sourcePart.type === "text" && outPart.type === "text") { + if (sourcePart.text === outPart.text) continue; + if (isExcerptText(sourcePart.text, outPart.text)) continue; + return false; + } + if (JSON.stringify(sourcePart) !== JSON.stringify(outPart)) return false; + } + return true; +} + +function blocksProject( + source: ConversationTurn["content"], + out: ConversationTurn["content"], +): boolean { + const sourceBlocks = source.filter( + (block) => block.type !== "thinking" && block.type !== "redacted_thinking", + ); + if ( + sourceBlocks.length === 0 && + out.length === 1 && + out[0]?.type === "text" && + out[0].text === THINKING_ONLY_OMITTED + ) { + return source.some( + (block) => + block.type === "thinking" || block.type === "redacted_thinking", + ); + } + if (sourceBlocks.length !== out.length) return false; + for (let index = 0; index < sourceBlocks.length; index++) { + const sourceBlock = sourceBlocks[index]; + const outBlock = out[index]; + if (sourceBlock === undefined || outBlock === undefined) return false; + if (JSON.stringify(sourceBlock) === JSON.stringify(outBlock)) continue; + if ( + sourceBlock.type === "tool_result" && + outBlock.type === "tool_result" && + resultProjects(sourceBlock, outBlock) + ) { + continue; + } + return false; + } + return true; +} + +function turnText(turn: ConversationTurn): string { + let text = ""; + for (const block of turn.content) { + if (block.type === "text") text += block.text; + } + return text; +} + +function isPlainTextTurn(turn: ConversationTurn): boolean { + return !turn.content.some( + (block) => block.type === "tool_call" || block.type === "tool_result", + ); +} + +function projectsFrom( + source: ConversationTurn, + out: ConversationTurn, +): boolean { + if (sameTurnContent(source, out)) return true; + if (source.role !== out.role) return false; + if (!sameIdList(toolCallIds(source), toolCallIds(out))) return false; + if (!sameIdList(resultCallIds(source), resultCallIds(out))) return false; + return blocksProject(source.content, out.content); +} + +function coalescedProjects( + first: ConversationTurn, + second: ConversationTurn, + out: ConversationTurn, +): boolean { + if (first.role !== out.role || second.role !== out.role) return false; + if ( + !isPlainTextTurn(first) || + !isPlainTextTurn(second) || + !isPlainTextTurn(out) + ) { + return false; + } + return turnText(out) === turnText(first) + turnText(second); +} + +function tailProjectsFrom( + live: readonly ConversationTurn[], + rest: readonly ConversationTurn[], +): boolean { + if (rest.length > live.length) return false; + let cursor = 0; + for (const out of rest) { + let found = false; + while (cursor < live.length) { + const candidate = live[cursor]; + if (candidate !== undefined && projectsFrom(candidate, out)) { + cursor++; + found = true; + break; + } + const next = live[cursor + 1]; if ( - liveTurn === undefined || - restTurn === undefined || - !sameTurnContent(liveTurn, restTurn) + candidate !== undefined && + next !== undefined && + coalescedProjects(candidate, next, out) ) { - matched = false; + cursor += 2; + found = true; break; } + cursor++; } - if (matched) return true; + if (!found) return false; } - return false; + return true; +} + +function isCompactionOf( + live: readonly ConversationTurn[], + next: readonly ConversationTurn[], +): boolean { + if (live.length === 0 || next.length === 0) return false; + let sawSummary = false; + const rest: ConversationTurn[] = []; + for (const turn of next) { + if (isCompactionMarker(turn)) { + if (isSummaryTurn(turn)) sawSummary = true; + continue; + } + rest.push(turn); + } + if (!sawSummary) return false; + if (rest.length === 0) return true; + return tailProjectsFrom(live, rest); } async function listIndexPaths(dir: string): Promise> { From 47210c0b412a19477822685c1a23482787e28559 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 17:44:12 -0700 Subject: [PATCH 3/4] fix(session): reject a compaction that is only a shared turn --- src/session/optimized-context-store.test.ts | 57 +++++++++++++++++++++ src/session/optimized-context-store.ts | 46 ++++++++--------- 2 files changed, 78 insertions(+), 25 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index d4e99cae1..051aa6633 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -830,6 +830,24 @@ describe("concurrent sessions write disjoint transcripts", () => { ); }); + test("a summary plus one shared turn does not replace the transcript", async () => { + await expectSharedTurnAttacksToDivert([ + turn("[Compacted prior context]"), + turn("ok"), + ]); + }); + + test("a summary plus one middle turn does not replace the transcript", async () => { + await expectSharedTurnAttacksToDivert([ + turn("[Compacted prior context]"), + turn("middle"), + ]); + }); + + test("a summary alone does not replace the transcript", async () => { + await expectSharedTurnAttacksToDivert([turn("[Compacted prior context]")]); + }); + test("a foreign transcript without an owner id still diverts", async () => { const dir = tempDir(); const first = await createOptimizedContextStore(dir); @@ -1228,6 +1246,45 @@ function versionedTurnsFiles(dir: string): string[] { .filter((name) => /^turns-\d+-[0-9a-f]+\.jsonl$/.test(name)); } +function fiveTurnsEndingOk(): ConversationTurn[] { + return [ + turn("alpha"), + turn("middle"), + turn("gamma"), + turn("delta"), + turn("ok"), + ]; +} + +async function expectSharedTurnAttacksToDivert( + attack: readonly ConversationTurn[], +): Promise { + for (const markerless of [true, false]) { + const dir = tempDir(); + const owner = await createOptimizedContextStore(dir, { + ownerId: "session-owner", + }); + const original = fiveTurnsEndingOk(); + await owner.writeTurns([...original]); + await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await owner.commit({ message: "seed" }); + if (markerless) fs.unlinkSync(path.join(dir, "session-owner.json")); + + const attacker = await createOptimizedContextStore(dir); + await attacker.writeTurns([...attack]); + await attacker.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await attacker.commit({ message: "attack" }); + + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(original), + ); + expect(turnTexts((await attacker.load()).turns)).toEqual( + turnTexts(original), + ); + expect(versionedTurnsFiles(dir)).toHaveLength(1); + } +} + function readTurnsFromFile(fullPath: string): ConversationTurn[] { const text = fs.readFileSync(fullPath, "utf8"); return text diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 29671e79a..755c6e236 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -600,32 +600,29 @@ function tailProjectsFrom( live: readonly ConversationTurn[], rest: readonly ConversationTurn[], ): boolean { - if (rest.length > live.length) return false; - let cursor = 0; - for (const out of rest) { - let found = false; - while (cursor < live.length) { - const candidate = live[cursor]; - if (candidate !== undefined && projectsFrom(candidate, out)) { - cursor++; - found = true; - break; - } - const next = live[cursor + 1]; - if ( - candidate !== undefined && - next !== undefined && - coalescedProjects(candidate, next, out) - ) { - cursor += 2; - found = true; - break; - } - cursor++; + if (rest.length < 2 || rest.length > live.length) return false; + let cursor = live.length; + for (let index = rest.length - 1; index >= 0; index--) { + const out = rest[index]; + if (out === undefined || cursor <= 0) return false; + const candidate = live[cursor - 1]; + if (candidate !== undefined && projectsFrom(candidate, out)) { + cursor--; + continue; } - if (!found) return false; + const earlier = live[cursor - 2]; + const later = live[cursor - 1]; + if ( + earlier !== undefined && + later !== undefined && + coalescedProjects(earlier, later, out) + ) { + cursor -= 2; + continue; + } + return false; } - return true; + return cursor > 0; } function isCompactionOf( @@ -643,7 +640,6 @@ function isCompactionOf( rest.push(turn); } if (!sawSummary) return false; - if (rest.length === 0) return true; return tailProjectsFrom(live, rest); } From 4dbc20b7bb4c1cf19bca53cd7c6d21608f18a7df Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 17:53:48 -0700 Subject: [PATCH 4/4] fix(session): accept only a mutated compaction tail --- src/session/optimized-context-store.test.ts | 123 ++++++++++++++++++-- src/session/optimized-context-store.ts | 65 ++++++----- 2 files changed, 146 insertions(+), 42 deletions(-) diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 051aa6633..c6bfd9835 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -42,6 +42,24 @@ function turn(text: string): ConversationTurn { return { role: "user", content: [{ type: "text", text }], timestamp: 1 }; } +function joinedUserTurn(texts: readonly string[]): ConversationTurn { + return { + role: "user", + content: texts.map((text) => ({ type: "text", text })), + timestamp: 1, + }; +} + +function isCompactionMarkerTurn(candidate: ConversationTurn): boolean { + const block = candidate.content[0]; + if (block === undefined || block.type !== "text") return false; + return ( + block.text.startsWith("[Compacted prior context]") || + block.text === "[compact]" || + block.text === "[compaction]" + ); +} + function jsonl(turns: ConversationTurn[]): string { return turns.map((t) => JSON.stringify(t)).join("\n") + "\n"; } @@ -813,20 +831,23 @@ describe("concurrent sessions write disjoint transcripts", () => { await first.commit({ message: "original" }); const second = await createOptimizedContextStore(dir); - const compacted = [turn("[Compacted prior context]"), turn("c"), turn("d")]; + const compacted = [ + turn("[Compacted prior context]"), + joinedUserTurn(["c", "d"]), + ]; await second.writeTurns([...compacted]); await second.writeMetadata(EMPTY_CHECKPOINT_METADATA); await second.commit({ message: "compact" }); expect(versionedTurnsFiles(dir)).toEqual([]); - expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( - turnTexts(compacted), + expect(JSON.stringify(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toBe( + JSON.stringify(compacted), ); - expect(turnTexts((await second.load()).turns)).toEqual( - turnTexts(compacted), + expect(JSON.stringify((await second.load()).turns)).toBe( + JSON.stringify(compacted), ); - expect(turnTexts((await loadRecentTurns(dir, 10)).turns)).toEqual( - turnTexts(compacted), + expect(JSON.stringify((await loadRecentTurns(dir, 10)).turns)).toBe( + JSON.stringify(compacted), ); }); @@ -848,6 +869,81 @@ describe("concurrent sessions write disjoint transcripts", () => { await expectSharedTurnAttacksToDivert([turn("[Compacted prior context]")]); }); + test("a summary plus an unmodified suffix does not replace the transcript", async () => { + const dir = tempDir(); + const owner = await createOptimizedContextStore(dir); + const original = Array.from({ length: 12 }, (_, index) => + turn(`unique-turn-${index}`), + ); + await owner.writeTurns([...original]); + await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await owner.commit({ message: "seed" }); + fs.unlinkSync(path.join(dir, "session-owner.json")); + + const attacker = await createOptimizedContextStore(dir); + await attacker.writeTurns([ + turn("[Compacted prior context]\nforged"), + turn("unique-turn-10"), + turn("unique-turn-11"), + ]); + await attacker.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await attacker.commit({ message: "attack" }); + + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(original), + ); + expect(turnTexts((await attacker.load()).turns)).toEqual( + turnTexts(original), + ); + expect(versionedTurnsFiles(dir)).toHaveLength(1); + }); + + test("a coalesced plain-text fold stays on the live transcript", async () => { + const original = [ + turn("H".repeat(20000)), + turn("older"), + turn("tail-a"), + turn("tail-b"), + turn("tail-c"), + ]; + const folded = (await createPruningCompactor().apply(original, {} as never)) + .output; + const kept = folded.filter( + (candidate) => !isCompactionMarkerTurn(candidate), + ); + expect(kept).toHaveLength(1); + expect( + kept[0]?.content.map((block) => + block.type === "text" ? block.text : block.type, + ), + ).toEqual(["older", "tail-a", "tail-b", "tail-c"]); + + for (const dropMarker of [true, false]) { + const dir = tempDir(); + const first = await createOptimizedContextStore(dir); + await first.writeTurns([...original]); + await first.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await first.commit({ message: "original" }); + if (dropMarker) fs.unlinkSync(path.join(dir, "session-owner.json")); + const second = await createOptimizedContextStore(dir); + await second.writeTurns(folded); + await second.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await second.commit({ message: "fold" }); + const later = [...folded, turn("after-fold")]; + await second.writeTurns(later); + await second.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await second.commit({ message: "later" }); + + expect(versionedTurnsFiles(dir)).toEqual([]); + expect( + JSON.stringify(readTurnsFromFile(path.join(dir, TURNS_FILE))), + ).toBe(JSON.stringify(later)); + expect(JSON.stringify((await second.load()).turns)).toBe( + JSON.stringify(later), + ); + } + }); + test("a foreign transcript without an owner id still diverts", async () => { const dir = tempDir(); const first = await createOptimizedContextStore(dir); @@ -983,17 +1079,20 @@ describe("concurrent sessions write disjoint transcripts", () => { const upgraded = await createOptimizedContextStore(dir, { ownerId: "session-S", }); - const compacted = [turn("[Compacted prior context]"), turn("c"), turn("d")]; + const compacted = [ + turn("[Compacted prior context]"), + joinedUserTurn(["c", "d"]), + ]; await upgraded.writeTurns([...compacted]); await upgraded.writeMetadata(EMPTY_CHECKPOINT_METADATA); await upgraded.commit({ message: "upgrade compact" }); expect(versionedTurnsFiles(dir)).toEqual([]); - expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( - turnTexts(compacted), + expect(JSON.stringify(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toBe( + JSON.stringify(compacted), ); - expect(turnTexts((await upgraded.load()).turns)).toEqual( - turnTexts(compacted), + expect(JSON.stringify((await upgraded.load()).turns)).toBe( + JSON.stringify(compacted), ); }); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 755c6e236..5558ca5ea 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -555,14 +555,6 @@ function blocksProject( return true; } -function turnText(turn: ConversationTurn): string { - let text = ""; - for (const block of turn.content) { - if (block.type === "text") text += block.text; - } - return text; -} - function isPlainTextTurn(turn: ConversationTurn): boolean { return !turn.content.some( (block) => block.type === "tool_call" || block.type === "tool_result", @@ -580,49 +572,62 @@ function projectsFrom( return blocksProject(source.content, out.content); } -function coalescedProjects( - first: ConversationTurn, - second: ConversationTurn, +function sameTurnBody(a: ConversationTurn, b: ConversationTurn): boolean { + return ( + a.role === b.role && JSON.stringify(a.content) === JSON.stringify(b.content) + ); +} + +function coalescedSpan( + live: readonly ConversationTurn[], + end: number, out: ConversationTurn, -): boolean { - if (first.role !== out.role || second.role !== out.role) return false; - if ( - !isPlainTextTurn(first) || - !isPlainTextTurn(second) || - !isPlainTextTurn(out) - ) { - return false; +): number { + if (!isPlainTextTurn(out)) return 0; + const target = JSON.stringify(out.content); + const parts: ConversationTurn["content"][number][] = []; + for (let start = end - 1; start >= 0; start--) { + const source = live[start]; + if ( + source === undefined || + source.role !== out.role || + !isPlainTextTurn(source) + ) { + return 0; + } + parts.unshift(...source.content); + const span = end - start; + if (span >= 2 && JSON.stringify(parts) === target) return span; + if (parts.length > out.content.length) return 0; } - return turnText(out) === turnText(first) + turnText(second); + return 0; } function tailProjectsFrom( live: readonly ConversationTurn[], rest: readonly ConversationTurn[], ): boolean { - if (rest.length < 2 || rest.length > live.length) return false; + if (rest.length === 0 || rest.length > live.length) return false; let cursor = live.length; + let mutated = false; for (let index = rest.length - 1; index >= 0; index--) { const out = rest[index]; if (out === undefined || cursor <= 0) return false; const candidate = live[cursor - 1]; if (candidate !== undefined && projectsFrom(candidate, out)) { + if (!sameTurnBody(candidate, out)) mutated = true; cursor--; continue; } - const earlier = live[cursor - 2]; - const later = live[cursor - 1]; - if ( - earlier !== undefined && - later !== undefined && - coalescedProjects(earlier, later, out) - ) { - cursor -= 2; + const span = coalescedSpan(live, cursor, out); + if (span >= 2) { + mutated = true; + cursor -= span; continue; } return false; } - return cursor > 0; + return cursor > 0 && mutated; } function isCompactionOf(