Skip to content

Commit abf0ca4

Browse files
committed
fix(session): concurrent sessions write disjoint transcripts
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. Ownership is decided solely by writeTurns: only a fresh store whose first writeTurns diverges from the on-disk history (a genuinely foreign concurrent session) diverts to versioned bases. A differing first prompt on a same-session resume never diverts — the reactor materializes the prompt before any writeTurns, so prompt divergence alone does not prove a foreign store — and resumes keep writing turns.jsonl/prompt.jsonl as before. Repro: two stores on one dir, 107-record transcript plus a second session commit; asserts no overwrite, no lost records, disjoint files. Regression: same-session resume with a differing first prompt keeps updating turns.jsonl and stays visible to load().
1 parent 40e19ce commit abf0ca4

2 files changed

Lines changed: 150 additions & 14 deletions

File tree

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

Lines changed: 83 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -597,6 +597,89 @@ describe("createSessionStores", () => {
597597
});
598598
});
599599

600+
describe("concurrent sessions write disjoint transcripts (CL-10372)", () => {
601+
test("a second session sharing the dir never overwrites the first session's turns", async () => {
602+
const dir = tempDir();
603+
const storeA = await createOptimizedContextStore(dir);
604+
const storeB = await createOptimizedContextStore(dir);
605+
606+
// Session A lands a 107-record transcript (the ticket's verified size).
607+
const turnsA = Array.from({ length: 107 }, (_, i) => turn(`a-${i}`));
608+
await storeA.writeTurns([...turnsA]);
609+
await storeA.writeMetadata(EMPTY_CHECKPOINT_METADATA);
610+
await storeA.commit({ message: "session A" });
611+
612+
// Session B starts concurrently on the same dir with its own transcript.
613+
const turnsB = Array.from({ length: 7 }, (_, i) => turn(`b-${i}`));
614+
await storeB.writeTurns([...turnsB]);
615+
await storeB.writeMetadata(EMPTY_CHECKPOINT_METADATA);
616+
await storeB.commit({ message: "session B" });
617+
618+
// A's transcript must survive intact: no overwrite.
619+
expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual(
620+
turnTexts(turnsA),
621+
);
622+
623+
// B's records must survive in a disjoint, versioned file: no lost records.
624+
const versioned = fs
625+
.readdirSync(dir)
626+
.filter((name) => /^turns-\d+-[0-9a-f]+\.jsonl$/.test(name));
627+
expect(versioned.length).toBeGreaterThan(0);
628+
const bOnDisk = versioned.flatMap((name) =>
629+
turnTexts(readTurnsFromFile(path.join(dir, name))),
630+
);
631+
expect(bOnDisk).toEqual(turnTexts(turnsB));
632+
633+
// Disjoint: every record from both sessions survives exactly once.
634+
const all = [
635+
...turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE))),
636+
...bOnDisk,
637+
];
638+
expect(new Set(all).size).toBe(turnsA.length + turnsB.length);
639+
});
640+
641+
test("same-session resume with a differing first prompt keeps writing turns.jsonl (CL-10372 regression)", async () => {
642+
const dir = tempDir();
643+
644+
// Owner session lands a committed transcript, then is resumed by a fresh
645+
// store in a new process (the reactor writes the materialized prompt in
646+
// pre-inference BEFORE any writeTurns; the prompt legitimately differs
647+
// from the on-disk turns when context transforms / ephemeral turns apply).
648+
const owner = await createOptimizedContextStore(dir);
649+
const live = [turn("a"), turn("b")];
650+
await owner.writeTurns([...live]);
651+
await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA);
652+
await owner.commit({ message: "owner" });
653+
654+
const resumed = await createOptimizedContextStore(dir);
655+
await resumed.writePrompt([turn("a"), turn("b"), turn("ephemeral")]);
656+
const next = [...live, turn("c")];
657+
await resumed.writeTurns([...next]);
658+
await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA);
659+
await resumed.commit({ message: "resume append" });
660+
661+
// No versioned base: this store is the session owner resuming, so it
662+
// keeps owning turns.jsonl and the new turn is visible to load().
663+
expect(
664+
fs
665+
.readdirSync(dir)
666+
.filter((name) => /^turns-\d+-[0-9a-f]+\.jsonl$/.test(name)),
667+
).toEqual([]);
668+
expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual(
669+
turnTexts(next),
670+
);
671+
expect(turnTexts((await resumed.load()).turns)).toEqual(turnTexts(next));
672+
});
673+
});
674+
675+
function readTurnsFromFile(fullPath: string): ConversationTurn[] {
676+
const text = fs.readFileSync(fullPath, "utf8");
677+
return text
678+
.split("\n")
679+
.filter((line) => line.length > 0)
680+
.map((line) => JSON.parse(line) as ConversationTurn);
681+
}
682+
600683
async function gitLsTree(dir: string): Promise<string[]> {
601684
const proc = Bun.spawn(
602685
["git", "-C", dir, "ls-tree", "-r", "--name-only", "HEAD"],

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

Lines changed: 67 additions & 14 deletions
Original file line numberDiff line numberDiff line change
@@ -84,6 +84,22 @@ function sanitizeCallId(callId: string): string {
8484
return callId.replace(UNSAFE_FILENAME_CHARS, "_");
8585
}
8686

87+
/**
88+
* Unique versioned base name for a transcript file (e.g.
89+
* `turns-1730000000000-a1b2c3d4.jsonl`). A store whose transcript diverges
90+
* from an on-disk transcript it did not author writes to one of these instead
91+
* of the shared base, so concurrent sessions keep disjoint files. The
92+
* `-<epoch>-<hex>` infix never collides with the numbered segment pattern
93+
* (`turns-0001.jsonl`), so the base readers ignore these by design.
94+
*/
95+
function versionedBaseName(baseName: string): string {
96+
const dot = baseName.lastIndexOf(".");
97+
const stem = dot === -1 ? baseName : baseName.slice(0, dot);
98+
const ext = dot === -1 ? "" : baseName.slice(dot);
99+
const stamp = `${Date.now()}-${Math.random().toString(16).slice(2, 10)}`;
100+
return `${stem}-${stamp}${ext}`;
101+
}
102+
87103
/**
88104
* Parse conversation turns out of one JSONL segment. A crash can tear the
89105
* final line of the active segment mid-write; `tolerateTornTail` drops it.
@@ -395,11 +411,14 @@ async function headOid(dir: string): Promise<string | null> {
395411
* Captured before a staged rewrite lands so a failed commit can restore the
396412
* published generation.
397413
*/
398-
async function snapshotTurnSegments(dir: string): Promise<Map<string, string>> {
414+
async function snapshotTurnSegments(
415+
dir: string,
416+
baseName: string,
417+
): Promise<Map<string, string>> {
399418
const snapshot = new Map<string, string>();
400-
const highest = await highestSegmentIndex(dir, TURNS_FILE);
419+
const highest = await highestSegmentIndex(dir, baseName);
401420
for (let index = 0; index <= highest; index++) {
402-
const name = segmentFileName(TURNS_FILE, index);
421+
const name = segmentFileName(baseName, index);
403422
try {
404423
snapshot.set(
405424
name,
@@ -421,11 +440,12 @@ async function snapshotTurnSegments(dir: string): Promise<Map<string, string>> {
421440
async function restoreTurnSegments(
422441
dir: string,
423442
snapshot: ReadonlyMap<string, string>,
443+
baseName: string,
424444
): Promise<void> {
425445
const names = new Set<string>(snapshot.keys());
426-
const highest = await highestSegmentIndex(dir, TURNS_FILE);
446+
const highest = await highestSegmentIndex(dir, baseName);
427447
for (let index = 0; index <= highest; index++) {
428-
names.add(segmentFileName(TURNS_FILE, index));
448+
names.add(segmentFileName(baseName, index));
429449
}
430450
for (const name of names) {
431451
const full = path.join(dir, name);
@@ -517,11 +537,26 @@ export async function createSessionStores(
517537
const base = await createIsogitStore(dir, signer);
518538
const pendingBlobFilepaths = new Set<string>();
519539
const pendingSegmentPaths = new Set<string>();
540+
// Base file each writer owns. A store that detects an on-disk transcript it
541+
// did not author (concurrent session sharing the dir) switches both bases to
542+
// disjoint versioned names so its writes never clobber the other session.
543+
let turnsBaseName = TURNS_FILE;
544+
let promptBaseName = PROMPT_FILE;
545+
let versionedBase: string | null = null;
520546
let writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE);
521547
let writePromptSegmented = createSegmentedJSONLWriter(dir, PROMPT_FILE);
522548
let liveTurnRefs: readonly ConversationTurn[] | null = null;
523549
let unpublishedRewrite: ConversationTurn[] | null = null;
524550

551+
function divertToVersionedBases(): void {
552+
if (versionedBase !== null) return;
553+
versionedBase = versionedBaseName(TURNS_FILE);
554+
turnsBaseName = versionedBase;
555+
writeTurnsSegmented = createSegmentedJSONLWriter(dir, turnsBaseName);
556+
promptBaseName = versionedBaseName(PROMPT_FILE);
557+
writePromptSegmented = createSegmentedJSONLWriter(dir, promptBaseName);
558+
}
559+
525560
function refPrefixLength(
526561
prev: readonly ConversationTurn[],
527562
next: readonly ConversationTurn[],
@@ -590,7 +625,13 @@ export async function createSessionStores(
590625
? baseTurns
591626
: await loadTurnsWithoutMalformedToolSequence(baseTurns, extraTexts);
592627
if (live.length > 0 && contentPrefixLength(live, turns) < live.length) {
593-
unpublishedRewrite = [...turns];
628+
// This fresh store's transcript is not an extension of the on-disk
629+
// history, so the on-disk transcript was authored by another session
630+
// sharing the dir (CL-10372). Rewriting the shared base would clobber
631+
// that session's records; divert to a disjoint versioned base instead.
632+
divertToVersionedBases();
633+
await writeSegmented(writeTurnsSegmented, turns);
634+
liveTurnRefs = [...turns];
594635
return;
595636
}
596637
await writeSegmented(writeTurnsSegmented, turns);
@@ -644,22 +685,29 @@ export async function createSessionStores(
644685
}
645686
}
646687
if (live === null || !promptEqualsLiveTurns(live, turns)) {
688+
// The prompt is a materialized snapshot and legitimately differs from
689+
// the live turns on a normal resume (context transforms, ephemeral
690+
// turns, summaries), so a differing prompt alone never proves a foreign
691+
// store. Ownership is decided by writeTurns: only its fresh-store
692+
// divergence diverts to versioned bases, and writePromptSegmented is
693+
// rebound with it. Prompt writes therefore stay on the shared base
694+
// unless that divert already happened.
647695
await writeSegmented(writePromptSegmented, turns);
648696
return;
649697
}
650698
// Identical: converge disk to no prompt segment so a stale snapshot
651699
// cannot linger; removals join pendingSegmentPaths for commit staging.
652-
const highest = await highestSegmentIndex(dir, PROMPT_FILE);
700+
const highest = await highestSegmentIndex(dir, promptBaseName);
653701
let removed = false;
654702
for (let index = 0; index <= highest; index++) {
655-
const name = segmentFileName(PROMPT_FILE, index);
703+
const name = segmentFileName(promptBaseName, index);
656704
if (!(await pathExists(path.join(dir, name)))) continue;
657705
await fs.promises.unlink(path.join(dir, name));
658706
pendingSegmentPaths.add(name);
659707
removed = true;
660708
}
661709
if (removed) {
662-
writePromptSegmented = createSegmentedJSONLWriter(dir, PROMPT_FILE);
710+
writePromptSegmented = createSegmentedJSONLWriter(dir, promptBaseName);
663711
}
664712
}
665713

@@ -792,7 +840,9 @@ export async function createSessionStores(
792840
// published generation. The rewrite stays staged so a retry can still
793841
// publish it.
794842
const segmentSnapshot =
795-
stagedRewrite === null ? null : await snapshotTurnSegments(dir);
843+
stagedRewrite === null
844+
? null
845+
: await snapshotTurnSegments(dir, turnsBaseName);
796846
const headBefore = stagedRewrite === null ? null : await headOid(dir);
797847
let extraPaths: string[] = [];
798848

@@ -814,8 +864,8 @@ export async function createSessionStores(
814864

815865
// Disk is the source of truth for which segments stay tracked after a
816866
// rewrite or heal, even when pendingSegmentPaths was lost.
817-
await reconcileSegmentStaging(dir, TURNS_FILE, toAdd, toRemove);
818-
await reconcileSegmentStaging(dir, PROMPT_FILE, toAdd, toRemove);
867+
await reconcileSegmentStaging(dir, turnsBaseName, toAdd, toRemove);
868+
await reconcileSegmentStaging(dir, promptBaseName, toAdd, toRemove);
819869

820870
if (await pathExists(path.join(dir, EVIDENCE_ARCHIVE_DIR))) {
821871
toAdd.push(EVIDENCE_ARCHIVE_DIR);
@@ -882,14 +932,17 @@ export async function createSessionStores(
882932
headBefore !== null && headNow !== null && headNow !== headBefore;
883933
if (!landed) {
884934
try {
885-
await restoreTurnSegments(dir, segmentSnapshot);
935+
await restoreTurnSegments(dir, segmentSnapshot, turnsBaseName);
886936
} catch {
887937
// A partial restore must not mask the real commit error.
888938
}
889939
}
890940
// Drop the writer's stale in-memory state so a retry rewrites from
891941
// scratch.
892-
writeTurnsSegmented = createSegmentedJSONLWriter(dir, TURNS_FILE);
942+
writeTurnsSegmented = createSegmentedJSONLWriter(
943+
dir,
944+
turnsBaseName,
945+
);
893946
liveTurnRefs = null;
894947
}
895948
throw cause;

0 commit comments

Comments
 (0)