From 6c83635708e4e156699d2daaf84039119becb15b Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Sun, 11 Oct 2026 23:32:08 -0700 Subject: [PATCH 1/2] fix(session): concurrent sessions write disjoint transcripts A fresh context store whose transcript is foreign to the on-disk owner (a concurrent session sharing the dir) previously rewrote the shared turns.jsonl at commit, losing the other session's records. Add an explicit ownership marker (session-owner.json) recorded when a store first claims the shared base, and divert only when the on-disk owner is genuinely someone else: - marker matches this store's owner id (same session resuming): any rewrite, including compaction that replaces history, stays canonical on the shared base; - marker belongs to a foreign owner and the incoming transcript is not an extension of the on-disk history: divert turns and prompt to unique versioned bases (turns--.jsonl); - prefix-extension resume without a stable id: continue the shared base. The first-write ownership decision is serialized with the per-dir lock so two concurrent fresh stores cannot both claim the shared base (one diverts). Diverted writes are staged and land at commit, so a crash before commit leaves no orphaned versioned file. Versioned names use a UUID suffix for same-millisecond uniqueness. The primary session caller passes its session id as the owner id so resumed sessions recognize their own marker. Tests: concurrent disjoint repro, same-session resume regression, same-session compaction-on-resume canonical, interleaved first writes, crash-mid-divert no orphan, and versioned-name uniqueness. --- src/session/assemble-runtime.ts | 8 +- src/session/optimized-context-store.test.ts | 190 ++++++++++++++++++ src/session/optimized-context-store.ts | 207 +++++++++++++++++--- 3 files changed, 379 insertions(+), 26 deletions(-) diff --git a/src/session/assemble-runtime.ts b/src/session/assemble-runtime.ts index 6b473ff91..9bdde8aed 100644 --- a/src/session/assemble-runtime.ts +++ b/src/session/assemble-runtime.ts @@ -675,7 +675,13 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent { // Stamp last* only on success: a failed /clear factory would otherwise // make the retry look like the same session and inherit stale extras. if (!inheritFromPrev) stickyFromStore = undefined; - const { storage, audit } = await createSessionStores(workdir); + // The session id is the store's stable owner identity: a resumed session + // recognizes the on-disk ownership marker as its own, so its rewrites + // (compaction) stay canonical on the shared transcript instead of being + // diverted as foreign. + const { storage, audit } = await createSessionStores(workdir, { + ownerId: sessionId, + }); // Primary-only: workers never pass evidenceArchiveHolder, so they keep // plain storage without admission / authorize recording wraps. const archiveHolder = wiring.evidenceArchiveHolder; diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 727933807..10dd69b6a 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -9,6 +9,7 @@ import { createOptimizedContextStore, createSessionStores, loadRecentTurns, + versionedBaseName, } from "./optimized-context-store.js"; import { segmentFileName, listSegmentFiles } from "./incremental-jsonl.js"; @@ -597,6 +598,195 @@ describe("createSessionStores", () => { }); }); +describe("concurrent sessions write disjoint transcripts (CL-10372)", () => { + test("a second session sharing the dir never overwrites the first session's turns", async () => { + const dir = tempDir(); + const storeA = await createOptimizedContextStore(dir); + const storeB = await createOptimizedContextStore(dir); + + // Session A lands a 107-record transcript (the ticket's verified size). + const turnsA = Array.from({ length: 107 }, (_, i) => turn(`a-${i}`)); + await storeA.writeTurns([...turnsA]); + await storeA.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await storeA.commit({ message: "session A" }); + + // Session B starts concurrently on the same dir with its own transcript. + const turnsB = Array.from({ length: 7 }, (_, i) => turn(`b-${i}`)); + await storeB.writeTurns([...turnsB]); + await storeB.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await storeB.commit({ message: "session B" }); + + // A's transcript must survive intact: no overwrite. + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(turnsA), + ); + + // B's records must survive in a disjoint, versioned file: no lost records. + const versioned = versionedTurnsFiles(dir); + expect(versioned.length).toBeGreaterThan(0); + const bOnDisk = versioned.flatMap((name) => + turnTexts(readTurnsFromFile(path.join(dir, name))), + ); + expect(bOnDisk).toEqual(turnTexts(turnsB)); + + // Disjoint: every record from both sessions survives exactly once. + const all = [ + ...turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE))), + ...bOnDisk, + ]; + expect(new Set(all).size).toBe(turnsA.length + turnsB.length); + }); + + test("same-session resume with a differing first prompt keeps writing turns.jsonl (CL-10372 regression)", async () => { + const dir = tempDir(); + + // Owner session lands a committed transcript, then is resumed by a fresh + // store in a new process (the reactor writes the materialized prompt in + // pre-inference BEFORE any writeTurns; the prompt legitimately differs + // from the on-disk turns when context transforms / ephemeral turns apply). + const owner = await createOptimizedContextStore(dir); + const live = [turn("a"), turn("b")]; + await owner.writeTurns([...live]); + await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await owner.commit({ message: "owner" }); + + const resumed = await createOptimizedContextStore(dir); + await resumed.writePrompt([turn("a"), turn("b"), turn("ephemeral")]); + const next = [...live, turn("c")]; + await resumed.writeTurns([...next]); + await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await resumed.commit({ message: "resume append" }); + + // No versioned base: this store is the session owner resuming, so it + // keeps owning turns.jsonl and the new turn is visible to load(). + expect(versionedTurnsFiles(dir)).toEqual([]); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(next), + ); + expect(turnTexts((await resumed.load()).turns)).toEqual(turnTexts(next)); + }); + + test("same-session compaction rewrite on resume is canonical, not diverted", async () => { + const dir = tempDir(); + const owner = await createOptimizedContextStore(dir, { + ownerId: "session-S", + }); + const original = [turn("a"), turn("b"), turn("c"), turn("d")]; + await owner.writeTurns([...original]); + await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await owner.commit({ message: "owner" }); + + // The same session resumes (same owner id) and its FIRST write is a + // compaction that replaces the head. Ownership is by marker, not content + // shape: this is the owner, so the rewrite is canonical on turns.jsonl. + const resumed = await createOptimizedContextStore(dir, { + ownerId: "session-S", + }); + const compacted = [turn("[Compacted prior context]"), turn("c"), turn("d")]; + await resumed.writeTurns([...compacted]); + await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await resumed.commit({ message: "resume compact" }); + + expect(versionedTurnsFiles(dir)).toEqual([]); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( + turnTexts(compacted), + ); + expect(turnTexts((await resumed.load()).turns)).toEqual( + turnTexts(compacted), + ); + }); + + test("interleaved first writes cannot both claim the shared base", async () => { + const dir = tempDir(); + const storeA = await createOptimizedContextStore(dir); + const storeB = await createOptimizedContextStore(dir); + const turnsA = Array.from({ length: 50 }, (_, i) => turn(`a-${i}`)); + const turnsB = Array.from({ length: 50 }, (_, i) => turn(`b-${i}`)); + + await Promise.all([ + (async () => { + await storeA.writeTurns([...turnsA]); + await storeA.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await storeA.commit({ message: "A" }); + })(), + (async () => { + await storeB.writeTurns([...turnsB]); + await storeB.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await storeB.commit({ message: "B" }); + })(), + ]); + + // The dir lock serializes the first writes: exactly one store owns the + // shared base and the other lands in a single disjoint versioned file. + 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 sharedIsA = + JSON.stringify(shared) === JSON.stringify(turnTexts(turnsA)); + const sharedIsB = + JSON.stringify(shared) === JSON.stringify(turnTexts(turnsB)); + expect(sharedIsA || sharedIsB).toBe(true); + expect(other).toEqual(sharedIsA ? turnTexts(turnsB) : turnTexts(turnsA)); + expect(new Set([...shared, ...other]).size).toBe( + turnsA.length + turnsB.length, + ); + }); + + test("crash before the divert lands leaves no orphaned versioned file", async () => { + const dir = tempDir(); + const owner = await createOptimizedContextStore(dir); + await owner.writeTurns([turn("a"), turn("b")]); + await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await owner.commit({ message: "owner" }); + + const foreign = await createOptimizedContextStore(dir); + await foreign.writeTurns([turn("x"), turn("y")]); + // Simulated crash before commit: the divert is staged in memory only, so + // no versioned file may exist on disk yet. + expect(versionedTurnsFiles(dir)).toEqual([]); + + // A later plain resume (prefix-extension, no stable id) sees no orphan + // and keeps the shared transcript canonical. + const resumed = await createOptimizedContextStore(dir); + await resumed.writeTurns([turn("a"), turn("b"), turn("c")]); + await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA); + await resumed.commit({ message: "resume" }); + expect(versionedTurnsFiles(dir)).toEqual([]); + expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual([ + "a", + "b", + "c", + ]); + }); + + test("versioned base names are unique even within the same millisecond", () => { + const names = Array.from({ length: 100 }, () => + versionedBaseName(TURNS_FILE), + ); + expect(new Set(names).size).toBe(100); + for (const name of names) { + expect(/^turns-\d+-[0-9a-f]+\.jsonl$/.test(name)).toBe(true); + } + }); +}); + +function versionedTurnsFiles(dir: string): string[] { + return fs + .readdirSync(dir) + .filter((name) => /^turns-\d+-[0-9a-f]+\.jsonl$/.test(name)); +} + +function readTurnsFromFile(fullPath: string): ConversationTurn[] { + const text = fs.readFileSync(fullPath, "utf8"); + return text + .split("\n") + .filter((line) => line.length > 0) + .map((line) => JSON.parse(line) as ConversationTurn); +} + async function gitLsTree(dir: string): Promise { const proc = Bun.spawn( ["git", "-C", dir, "ls-tree", "-r", "--name-only", "HEAD"], diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index a490998d5..8c900a91d 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -34,6 +34,8 @@ const MANIFEST_FILE = "manifest.jsonl"; const METADATA_FILE = "metadata.json"; const TOOL_OUTPUT_DIR = "tool-output"; const EVIDENCE_ARCHIVE_DIR = "evidence-archive"; +/** Per-session ownership marker: records which store authored the shared base. */ +const OWNER_MARKER_FILE = "session-owner.json"; const log = getLogger([LOG_NAMESPACE_ROOT, "session", "context-store"]); @@ -84,6 +86,23 @@ function sanitizeCallId(callId: string): string { return callId.replace(UNSAFE_FILENAME_CHARS, "_"); } +/** + * Unique versioned base name for a transcript file (e.g. + * `turns-1730000000000-a1b2c3d4e5f6....jsonl`). A store whose transcript is + * foreign to the on-disk owner writes to one of these instead of the shared + * base, so concurrent sessions keep disjoint files. The `--` + * infix never collides with the numbered segment pattern (`turns-0001.jsonl`), + * so the base readers ignore these by design. The UUID suffix keeps names + * unique even when two diverts happen within the same millisecond. + */ +export function versionedBaseName(baseName: string): string { + const dot = baseName.lastIndexOf("."); + const stem = dot === -1 ? baseName : baseName.slice(0, dot); + const ext = dot === -1 ? "" : baseName.slice(dot); + const uuid = globalThis.crypto.randomUUID().replaceAll("-", ""); + return `${stem}-${Date.now()}-${uuid}${ext}`; +} + /** * Parse conversation turns out of one JSONL segment. A crash can tear the * final line of the active segment mid-write; `tolerateTornTail` drops it. @@ -341,6 +360,44 @@ async function readBaseTurnsFromDisk(dir: string): Promise { return parseSegmentTurns(text, true, TURNS_FILE, true); } +/** + * Ownership marker for the shared base. Absent on a fresh/legacy dir; once a + * store writes the shared base first it claims the dir with its owner token, + * so a later store can tell "I authored this transcript" from "a foreign + * store did" without guessing from content shape. + */ +async function readOwnerMarker( + dir: string, +): Promise<{ ownerId: string } | null> { + try { + const raw = await fs.promises.readFile( + path.join(dir, OWNER_MARKER_FILE), + "utf-8", + ); + const parsed: unknown = JSON.parse(raw); + if ( + typeof parsed === "object" && + parsed !== null && + typeof (parsed as { ownerId?: unknown }).ownerId === "string" + ) { + return { ownerId: (parsed as { ownerId: string }).ownerId }; + } + return null; + } catch { + return null; + } +} + +async function writeOwnerMarker(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() }), + ); + await fs.promises.rename(tmp, full); +} + async function listIndexPaths(dir: string): Promise> { return new Set(await git.listFiles({ fs, dir })); } @@ -395,11 +452,14 @@ async function headOid(dir: string): Promise { * Captured before a staged rewrite lands so a failed commit can restore the * published generation. */ -async function snapshotTurnSegments(dir: string): Promise> { +async function snapshotTurnSegments( + dir: string, + baseName: string, +): Promise> { const snapshot = new Map(); - const highest = await highestSegmentIndex(dir, TURNS_FILE); + const highest = await highestSegmentIndex(dir, baseName); for (let index = 0; index <= highest; index++) { - const name = segmentFileName(TURNS_FILE, index); + const name = segmentFileName(baseName, index); try { snapshot.set( name, @@ -421,11 +481,12 @@ async function snapshotTurnSegments(dir: string): Promise> { async function restoreTurnSegments( dir: string, snapshot: ReadonlyMap, + baseName: string, ): Promise { const names = new Set(snapshot.keys()); - const highest = await highestSegmentIndex(dir, TURNS_FILE); + const highest = await highestSegmentIndex(dir, baseName); for (let index = 0; index <= highest; index++) { - names.add(segmentFileName(TURNS_FILE, index)); + names.add(segmentFileName(baseName, index)); } for (const name of names) { const full = path.join(dir, name); @@ -503,6 +564,22 @@ export interface SessionStores { audit: AuditStore; } +export interface SessionStoreOptions { + signer?: CommitSigner; + /** + * Stable identity of the session this store belongs to (e.g. the session + * id). Stores with the same id on one dir are the same session: resuming + * stores recognize the on-disk ownership marker as their own, so their + * rewrites (compaction) stay canonical on the shared base. Stores with + * different (or no) ids are foreign to each other and divert to disjoint + * versioned files instead of clobbering the shared transcript. When omitted + * the store uses a per-instance random token, so any two stores on one dir + * are treated as foreign unless the incoming transcript extends the on-disk + * history (a plain resume). + */ + ownerId?: string; +} + /** * Interchange git-store wrapper that avoids O(session length) work per * checkpoint: turns and prompts write as rolling segments so only the active @@ -511,17 +588,36 @@ export interface SessionStores { */ export async function createSessionStores( dir: string, - opts?: { signer?: CommitSigner }, + opts?: SessionStoreOptions, ): Promise { const signer = opts?.signer ?? (await loadOrCreateCommitSigner(dir)); const base = await createIsogitStore(dir, signer); const pendingBlobFilepaths = new Set(); const pendingSegmentPaths = new Set(); + // Stable per-session owner token (the caller's session id, or a random + // per-instance token). The on-disk ownership marker is compared against it + // to tell "I authored this transcript" from "a foreign store did". + const ownerToken = opts?.ownerId ?? globalThis.crypto.randomUUID(); + // Base file each writer owns. A store that detects a foreign on-disk owner + // (a concurrent session sharing the dir) switches both bases to disjoint + // versioned names so its writes never clobber the other session. + let turnsBaseName = TURNS_FILE; + let promptBaseName = PROMPT_FILE; + let versionedBase: string | null = null; let writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE); let writePromptSegmented = createSegmentedJSONLWriter(dir, PROMPT_FILE); let liveTurnRefs: readonly ConversationTurn[] | null = null; let unpublishedRewrite: ConversationTurn[] | null = null; + function divertToVersionedBases(): void { + if (versionedBase !== null) return; + versionedBase = versionedBaseName(TURNS_FILE); + turnsBaseName = versionedBase; + writeTurnsSegmented = createSegmentedJSONLWriter(dir, turnsBaseName); + promptBaseName = versionedBaseName(PROMPT_FILE); + writePromptSegmented = createSegmentedJSONLWriter(dir, promptBaseName); + } + function refPrefixLength( prev: readonly ConversationTurn[], next: readonly ConversationTurn[], @@ -571,6 +667,55 @@ export async function createSessionStores( liveTurnRefs = [...turns]; return; } + // Fresh-store first write: the ownership decision and the first write are + // serialized per dir so two concurrent fresh stores cannot both claim the + // shared base (one becomes owner, the other diverts instead of racing). + await withResolvedDirLock(dir, async () => { + if (liveTurnRefs !== null) { + if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) { + unpublishedRewrite = [...turns]; + return; + } + await writeSegmented(writeTurnsSegmented, turns); + liveTurnRefs = [...turns]; + return; + } + const marker = await readOwnerMarker(dir); + if (marker === null) { + // No owner yet: this store claims the shared base. + await writeOwnerMarker(dir, ownerToken); + await writeSegmented(writeTurnsSegmented, turns); + liveTurnRefs = [...turns]; + return; + } + const live = await loadLiveTurnsFromDisk(); + if (marker.ownerId === ownerToken) { + // This store is the session owner (resuming with a stable id): any + // rewrite, including a compaction that replaces history, is canonical + // on the shared base — never divert an owner's own rewrite. + await writeSegmented(writeTurnsSegmented, turns); + liveTurnRefs = [...turns]; + return; + } + if (live.length > 0 && contentPrefixLength(live, turns) < live.length) { + // The on-disk transcript was authored by a foreign store and this + // store's transcript is not an extension of it: rewriting the shared + // base would clobber the other session's records. Divert to a + // disjoint versioned base; the write is staged and lands at commit so + // a crash before then leaves no orphaned versioned file. + divertToVersionedBases(); + unpublishedRewrite = [...turns]; + return; + } + // Foreign marker but the incoming transcript extends the on-disk history + // (a plain resume without a stable id): continue the shared base. + await writeSegmented(writeTurnsSegmented, turns); + liveTurnRefs = [...turns]; + }); + } + + /** Live turns from disk for a fresh store (base + extra segments). */ + async function loadLiveTurnsFromDisk(): Promise { const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE); let baseTurns: ConversationTurn[]; try { @@ -585,16 +730,9 @@ export async function createSessionStores( ); baseTurns = await readBaseTurnsFromDisk(dir); } - const live = - extraTexts.length === 0 - ? baseTurns - : await loadTurnsWithoutMalformedToolSequence(baseTurns, extraTexts); - if (live.length > 0 && contentPrefixLength(live, turns) < live.length) { - unpublishedRewrite = [...turns]; - return; - } - await writeSegmented(writeTurnsSegmented, turns); - liveTurnRefs = [...turns]; + return extraTexts.length === 0 + ? baseTurns + : await loadTurnsWithoutMalformedToolSequence(baseTurns, extraTexts); } function promptEqualsLiveTurns( @@ -643,23 +781,37 @@ export async function createSessionStores( live = null; } } + if (versionedBase !== null && liveTurnRefs === null) { + // A divert is staged but has not landed at commit yet. No versioned + // file may be written before commit — a crash would orphan it — so the + // prompt snapshot is deferred to the next cycle (after the commit, the + // versioned base exists and liveTurnRefs is set). + return; + } if (live === null || !promptEqualsLiveTurns(live, turns)) { + // The prompt is a materialized snapshot and legitimately differs from + // the live turns on a normal resume (context transforms, ephemeral + // turns, summaries), so a differing prompt alone never proves a foreign + // store. Ownership is decided by writeTurns: only a foreign owner with a + // divergent transcript diverts to versioned bases, and + // writePromptSegmented is rebound with it. Prompt writes therefore stay + // on the shared base unless that divert already happened. await writeSegmented(writePromptSegmented, turns); return; } // Identical: converge disk to no prompt segment so a stale snapshot // cannot linger; removals join pendingSegmentPaths for commit staging. - const highest = await highestSegmentIndex(dir, PROMPT_FILE); + const highest = await highestSegmentIndex(dir, promptBaseName); let removed = false; for (let index = 0; index <= highest; index++) { - const name = segmentFileName(PROMPT_FILE, index); + const name = segmentFileName(promptBaseName, index); if (!(await pathExists(path.join(dir, name)))) continue; await fs.promises.unlink(path.join(dir, name)); pendingSegmentPaths.add(name); removed = true; } if (removed) { - writePromptSegmented = createSegmentedJSONLWriter(dir, PROMPT_FILE); + writePromptSegmented = createSegmentedJSONLWriter(dir, promptBaseName); } } @@ -792,7 +944,9 @@ export async function createSessionStores( // published generation. The rewrite stays staged so a retry can still // publish it. const segmentSnapshot = - stagedRewrite === null ? null : await snapshotTurnSegments(dir); + stagedRewrite === null + ? null + : await snapshotTurnSegments(dir, turnsBaseName); const headBefore = stagedRewrite === null ? null : await headOid(dir); let extraPaths: string[] = []; @@ -814,8 +968,8 @@ export async function createSessionStores( // Disk is the source of truth for which segments stay tracked after a // rewrite or heal, even when pendingSegmentPaths was lost. - await reconcileSegmentStaging(dir, TURNS_FILE, toAdd, toRemove); - await reconcileSegmentStaging(dir, PROMPT_FILE, toAdd, toRemove); + await reconcileSegmentStaging(dir, turnsBaseName, toAdd, toRemove); + await reconcileSegmentStaging(dir, promptBaseName, toAdd, toRemove); if (await pathExists(path.join(dir, EVIDENCE_ARCHIVE_DIR))) { toAdd.push(EVIDENCE_ARCHIVE_DIR); @@ -882,14 +1036,17 @@ export async function createSessionStores( headBefore !== null && headNow !== null && headNow !== headBefore; if (!landed) { try { - await restoreTurnSegments(dir, segmentSnapshot); + await restoreTurnSegments(dir, segmentSnapshot, turnsBaseName); } 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, TURNS_FILE); + writeTurnsSegmented = createSegmentedJSONLWriter( + dir, + turnsBaseName, + ); liveTurnRefs = null; } throw cause; @@ -909,7 +1066,7 @@ export async function createSessionStores( export async function createOptimizedContextStore( dir: string, - opts?: { signer?: CommitSigner }, + opts?: SessionStoreOptions, ): Promise { const { storage } = await createSessionStores(dir, opts); return storage; From 30ce6dddae8f0f3817edbe57227a86374b84dbb3 Mon Sep 17 00:00:00 2001 From: Sawyer Cutler Date: Mon, 12 Oct 2026 16:41:10 -0700 Subject: [PATCH 2/2] refactor(session): drop transcript-ownership narration --- src/session/assemble-runtime.ts | 6 +- src/session/optimized-context-store.test.ts | 26 ++------- src/session/optimized-context-store.ts | 65 ++++----------------- 3 files changed, 16 insertions(+), 81 deletions(-) diff --git a/src/session/assemble-runtime.ts b/src/session/assemble-runtime.ts index 9bdde8aed..38184a3b5 100644 --- a/src/session/assemble-runtime.ts +++ b/src/session/assemble-runtime.ts @@ -675,10 +675,8 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent { // Stamp last* only on success: a failed /clear factory would otherwise // make the retry look like the same session and inherit stale extras. if (!inheritFromPrev) stickyFromStore = undefined; - // The session id is the store's stable owner identity: a resumed session - // recognizes the on-disk ownership marker as its own, so its rewrites - // (compaction) stay canonical on the shared transcript instead of being - // diverted as foreign. + // Session id is the stable owner so a resume's compaction stays on the + // shared transcript. const { storage, audit } = await createSessionStores(workdir, { ownerId: sessionId, }); diff --git a/src/session/optimized-context-store.test.ts b/src/session/optimized-context-store.test.ts index 10dd69b6a..f118421c5 100644 --- a/src/session/optimized-context-store.test.ts +++ b/src/session/optimized-context-store.test.ts @@ -598,30 +598,26 @@ describe("createSessionStores", () => { }); }); -describe("concurrent sessions write disjoint transcripts (CL-10372)", () => { +describe("concurrent sessions write disjoint transcripts", () => { test("a second session sharing the dir never overwrites the first session's turns", async () => { const dir = tempDir(); const storeA = await createOptimizedContextStore(dir); const storeB = await createOptimizedContextStore(dir); - // Session A lands a 107-record transcript (the ticket's verified size). const turnsA = Array.from({ length: 107 }, (_, i) => turn(`a-${i}`)); await storeA.writeTurns([...turnsA]); await storeA.writeMetadata(EMPTY_CHECKPOINT_METADATA); await storeA.commit({ message: "session A" }); - // Session B starts concurrently on the same dir with its own transcript. const turnsB = Array.from({ length: 7 }, (_, i) => turn(`b-${i}`)); await storeB.writeTurns([...turnsB]); await storeB.writeMetadata(EMPTY_CHECKPOINT_METADATA); await storeB.commit({ message: "session B" }); - // A's transcript must survive intact: no overwrite. expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( turnTexts(turnsA), ); - // B's records must survive in a disjoint, versioned file: no lost records. const versioned = versionedTurnsFiles(dir); expect(versioned.length).toBeGreaterThan(0); const bOnDisk = versioned.flatMap((name) => @@ -629,7 +625,6 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => { ); expect(bOnDisk).toEqual(turnTexts(turnsB)); - // Disjoint: every record from both sessions survives exactly once. const all = [ ...turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE))), ...bOnDisk, @@ -637,13 +632,11 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => { expect(new Set(all).size).toBe(turnsA.length + turnsB.length); }); - test("same-session resume with a differing first prompt keeps writing turns.jsonl (CL-10372 regression)", async () => { + test("same-session resume with a differing first prompt keeps writing turns.jsonl", async () => { const dir = tempDir(); - // Owner session lands a committed transcript, then is resumed by a fresh - // store in a new process (the reactor writes the materialized prompt in - // pre-inference BEFORE any writeTurns; the prompt legitimately differs - // from the on-disk turns when context transforms / ephemeral turns apply). + // Reactor writes the prompt before writeTurns; that prompt may differ + // from on-disk turns. const owner = await createOptimizedContextStore(dir); const live = [turn("a"), turn("b")]; await owner.writeTurns([...live]); @@ -657,8 +650,6 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => { await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA); await resumed.commit({ message: "resume append" }); - // No versioned base: this store is the session owner resuming, so it - // keeps owning turns.jsonl and the new turn is visible to load(). expect(versionedTurnsFiles(dir)).toEqual([]); expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual( turnTexts(next), @@ -676,9 +667,6 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => { await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA); await owner.commit({ message: "owner" }); - // The same session resumes (same owner id) and its FIRST write is a - // compaction that replaces the head. Ownership is by marker, not content - // shape: this is the owner, so the rewrite is canonical on turns.jsonl. const resumed = await createOptimizedContextStore(dir, { ownerId: "session-S", }); @@ -716,8 +704,6 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => { })(), ]); - // The dir lock serializes the first writes: exactly one store owns the - // shared base and the other lands in a single disjoint versioned file. const shared = turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE))); const versioned = versionedTurnsFiles(dir); expect(versioned).toHaveLength(1); @@ -744,12 +730,8 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => { const foreign = await createOptimizedContextStore(dir); await foreign.writeTurns([turn("x"), turn("y")]); - // Simulated crash before commit: the divert is staged in memory only, so - // no versioned file may exist on disk yet. expect(versionedTurnsFiles(dir)).toEqual([]); - // A later plain resume (prefix-extension, no stable id) sees no orphan - // and keeps the shared transcript canonical. const resumed = await createOptimizedContextStore(dir); await resumed.writeTurns([turn("a"), turn("b"), turn("c")]); await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA); diff --git a/src/session/optimized-context-store.ts b/src/session/optimized-context-store.ts index 8c900a91d..011e624af 100644 --- a/src/session/optimized-context-store.ts +++ b/src/session/optimized-context-store.ts @@ -34,7 +34,6 @@ const MANIFEST_FILE = "manifest.jsonl"; const METADATA_FILE = "metadata.json"; const TOOL_OUTPUT_DIR = "tool-output"; const EVIDENCE_ARCHIVE_DIR = "evidence-archive"; -/** Per-session ownership marker: records which store authored the shared base. */ const OWNER_MARKER_FILE = "session-owner.json"; const log = getLogger([LOG_NAMESPACE_ROOT, "session", "context-store"]); @@ -87,13 +86,8 @@ function sanitizeCallId(callId: string): string { } /** - * Unique versioned base name for a transcript file (e.g. - * `turns-1730000000000-a1b2c3d4e5f6....jsonl`). A store whose transcript is - * foreign to the on-disk owner writes to one of these instead of the shared - * base, so concurrent sessions keep disjoint files. The `--` - * infix never collides with the numbered segment pattern (`turns-0001.jsonl`), - * so the base readers ignore these by design. The UUID suffix keeps names - * unique even when two diverts happen within the same millisecond. + * Epoch+uuid avoids numbered segments (`turns-0001.jsonl`) that base + * readers treat as live history. */ export function versionedBaseName(baseName: string): string { const dot = baseName.lastIndexOf("."); @@ -360,12 +354,6 @@ async function readBaseTurnsFromDisk(dir: string): Promise { return parseSegmentTurns(text, true, TURNS_FILE, true); } -/** - * Ownership marker for the shared base. Absent on a fresh/legacy dir; once a - * store writes the shared base first it claims the dir with its owner token, - * so a later store can tell "I authored this transcript" from "a foreign - * store did" without guessing from content shape. - */ async function readOwnerMarker( dir: string, ): Promise<{ ownerId: string } | null> { @@ -567,15 +555,8 @@ export interface SessionStores { export interface SessionStoreOptions { signer?: CommitSigner; /** - * Stable identity of the session this store belongs to (e.g. the session - * id). Stores with the same id on one dir are the same session: resuming - * stores recognize the on-disk ownership marker as their own, so their - * rewrites (compaction) stay canonical on the shared base. Stores with - * different (or no) ids are foreign to each other and divert to disjoint - * versioned files instead of clobbering the shared transcript. When omitted - * the store uses a per-instance random token, so any two stores on one dir - * are treated as foreign unless the incoming transcript extends the on-disk - * history (a plain resume). + * Same id shares the marker; omit for a random token, foreign unless the + * transcript extends history. */ ownerId?: string; } @@ -594,13 +575,7 @@ export async function createSessionStores( const base = await createIsogitStore(dir, signer); const pendingBlobFilepaths = new Set(); const pendingSegmentPaths = new Set(); - // Stable per-session owner token (the caller's session id, or a random - // per-instance token). The on-disk ownership marker is compared against it - // to tell "I authored this transcript" from "a foreign store did". const ownerToken = opts?.ownerId ?? globalThis.crypto.randomUUID(); - // Base file each writer owns. A store that detects a foreign on-disk owner - // (a concurrent session sharing the dir) switches both bases to disjoint - // versioned names so its writes never clobber the other session. let turnsBaseName = TURNS_FILE; let promptBaseName = PROMPT_FILE; let versionedBase: string | null = null; @@ -667,9 +642,7 @@ export async function createSessionStores( liveTurnRefs = [...turns]; return; } - // Fresh-store first write: the ownership decision and the first write are - // serialized per dir so two concurrent fresh stores cannot both claim the - // shared base (one becomes owner, the other diverts instead of racing). + // First claim is per-dir so two fresh stores cannot both own the base. await withResolvedDirLock(dir, async () => { if (liveTurnRefs !== null) { if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) { @@ -682,7 +655,6 @@ export async function createSessionStores( } const marker = await readOwnerMarker(dir); if (marker === null) { - // No owner yet: this store claims the shared base. await writeOwnerMarker(dir, ownerToken); await writeSegmented(writeTurnsSegmented, turns); liveTurnRefs = [...turns]; @@ -690,31 +662,23 @@ export async function createSessionStores( } const live = await loadLiveTurnsFromDisk(); if (marker.ownerId === ownerToken) { - // This store is the session owner (resuming with a stable id): any - // rewrite, including a compaction that replaces history, is canonical - // on the shared base — never divert an owner's own rewrite. + // Matching owner keeps compaction on the shared base, not a divert. await writeSegmented(writeTurnsSegmented, turns); liveTurnRefs = [...turns]; return; } if (live.length > 0 && contentPrefixLength(live, turns) < live.length) { - // The on-disk transcript was authored by a foreign store and this - // store's transcript is not an extension of it: rewriting the shared - // base would clobber the other session's records. Divert to a - // disjoint versioned base; the write is staged and lands at commit so - // a crash before then leaves no orphaned versioned file. + // Stage until commit so a crash leaves no orphaned versioned file. divertToVersionedBases(); unpublishedRewrite = [...turns]; return; } - // Foreign marker but the incoming transcript extends the on-disk history - // (a plain resume without a stable id): continue the shared base. + // Prefix extension with no stable id resumes the shared base. await writeSegmented(writeTurnsSegmented, turns); liveTurnRefs = [...turns]; }); } - /** Live turns from disk for a fresh store (base + extra segments). */ async function loadLiveTurnsFromDisk(): Promise { const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE); let baseTurns: ConversationTurn[]; @@ -782,20 +746,11 @@ export async function createSessionStores( } } if (versionedBase !== null && liveTurnRefs === null) { - // A divert is staged but has not landed at commit yet. No versioned - // file may be written before commit — a crash would orphan it — so the - // prompt snapshot is deferred to the next cycle (after the commit, the - // versioned base exists and liveTurnRefs is set). + // No versioned file before commit; a crash would orphan it. return; } if (live === null || !promptEqualsLiveTurns(live, turns)) { - // The prompt is a materialized snapshot and legitimately differs from - // the live turns on a normal resume (context transforms, ephemeral - // turns, summaries), so a differing prompt alone never proves a foreign - // store. Ownership is decided by writeTurns: only a foreign owner with a - // divergent transcript diverts to versioned bases, and - // writePromptSegmented is rebound with it. Prompt writes therefore stay - // on the shared base unless that divert already happened. + // A differing prompt is normal on resume and must not divert. await writeSegmented(writePromptSegmented, turns); return; }