fix(session): concurrent sessions write disjoint transcripts (CL-10372)

A fresh context store whose transcript diverges from an on-disk transcript it
did not author (a concurrent session sharing the dir) previously staged an
unpublished rewrite that clobbered the shared turns.jsonl at commit, losing the
other session's records. Detect the divergence on the fresh store's first write
and divert that store's turns and prompt to unique versioned base names
(turns-<epoch>-<rand>.jsonl), so concurrent sessions keep disjoint files.

Repro test: two stores on one dir, 107-record transcript + second session
commit; asserts no overwrite, no lost records, disjoint files.

Author: TheGreatAxios

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

@@ -597,6 +597,56 @@ 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 = fs
+      .readdirSync(dir)
+      .filter((name) => /^turns-\d+-[0-9a-f]+\.jsonl$/.test(name));
+    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);
+  });
+});
+
+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<string[]> {
   const proc = Bun.spawn(
     ["git", "-C", dir, "ls-tree", "-r", "--name-only", "HEAD"],

src/session/optimized-context-store.ts

@@ -84,6 +84,22 @@ function sanitizeCallId(callId: string): string {
   return callId.replace(UNSAFE_FILENAME_CHARS, "_");
 }
 
+/**
+ * Unique versioned base name for a transcript file (e.g.
+ * `turns-1730000000000-a1b2c3d4.jsonl`). A store whose transcript diverges
+ * from an on-disk transcript it did not author writes to one of these instead
+ * of the shared base, so concurrent sessions keep disjoint files. The
+ * `-<epoch>-<hex>` infix never collides with the numbered segment pattern
+ * (`turns-0001.jsonl`), so the base readers ignore these by design.
+ */
+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 stamp = `${Date.now()}-${Math.random().toString(16).slice(2, 10)}`;
+  return `${stem}-${stamp}${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.
@@ -395,11 +411,14 @@ async function headOid(dir: string): Promise<string | null> {
  * Captured before a staged rewrite lands so a failed commit can restore the
  * published generation.
  */
-async function snapshotTurnSegments(dir: string): Promise<Map<string, string>> {
+async function snapshotTurnSegments(
+  dir: string,
+  baseName: string,
+): Promise<Map<string, string>> {
   const snapshot = new Map<string, string>();
-  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 +440,12 @@ async function snapshotTurnSegments(dir: string): Promise<Map<string, string>> {
 async function restoreTurnSegments(
   dir: string,
   snapshot: ReadonlyMap<string, string>,
+  baseName: string,
 ): Promise<void> {
   const names = new Set<string>(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);
@@ -517,11 +537,26 @@ export async function createSessionStores(
   const base = await createIsogitStore(dir, signer);
   const pendingBlobFilepaths = new Set<string>();
   const pendingSegmentPaths = new Set<string>();
+  // Base file each writer owns. A store that detects an on-disk transcript it
+  // did not author (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 +606,13 @@ export async function createSessionStores(
       liveTurnRefs = [...turns];
       return;
     }
+    if (versionedBase !== null) {
+      // Already diverted (e.g. a colliding prompt wrote first): this store's
+      // transcript owns a disjoint versioned base; write turns there too.
+      await writeSegmented(writeTurnsSegmented, turns);
+      liveTurnRefs = [...turns];
+      return;
+    }
     const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE);
     let baseTurns: ConversationTurn[];
     try {
@@ -590,7 +632,13 @@ export async function createSessionStores(
         ? baseTurns
         : await loadTurnsWithoutMalformedToolSequence(baseTurns, extraTexts);
     if (live.length > 0 && contentPrefixLength(live, turns) < live.length) {
-      unpublishedRewrite = [...turns];
+      // This fresh store's transcript is not an extension of the on-disk
+      // history, so the on-disk transcript was authored by another session
+      // sharing the dir (CL-10372). Rewriting the shared base would clobber
+      // that session's records; divert to a disjoint versioned base instead.
+      divertToVersionedBases();
+      await writeSegmented(writeTurnsSegmented, turns);
+      liveTurnRefs = [...turns];
       return;
     }
     await writeSegmented(writeTurnsSegmented, turns);
@@ -644,22 +692,33 @@ export async function createSessionStores(
       }
     }
     if (live === null || !promptEqualsLiveTurns(live, turns)) {
+      // A fresh instance whose on-disk live turns came from another session
+      // (CL-10372) must not clobber that session's prompt snapshot either;
+      // divert this store's prompt to its own disjoint versioned base.
+      if (
+        versionedBase === null &&
+        liveTurnRefs === null &&
+        live !== null &&
+        live.length > 0
+      ) {
+        divertToVersionedBases();
+      }
       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 +851,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 +875,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 +943,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;