Skip to content

Commit 6c83635

Browse files
committed
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-<epoch>-<uuid>.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.
1 parent 40e19ce commit 6c83635

3 files changed

Lines changed: 379 additions & 26 deletions

File tree

‎src/session/assemble-runtime.ts‎

Lines changed: 7 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -675,7 +675,13 @@ export function assembleChatAgent(wiring: ChatAgentWiring): AssembledChatAgent {
675675
// Stamp last* only on success: a failed /clear factory would otherwise
676676
// make the retry look like the same session and inherit stale extras.
677677
if (!inheritFromPrev) stickyFromStore = undefined;
678-
const { storage, audit } = await createSessionStores(workdir);
678+
// The session id is the store's stable owner identity: a resumed session
679+
// recognizes the on-disk ownership marker as its own, so its rewrites
680+
// (compaction) stay canonical on the shared transcript instead of being
681+
// diverted as foreign.
682+
const { storage, audit } = await createSessionStores(workdir, {
683+
ownerId: sessionId,
684+
});
679685
// Primary-only: workers never pass evidenceArchiveHolder, so they keep
680686
// plain storage without admission / authorize recording wraps.
681687
const archiveHolder = wiring.evidenceArchiveHolder;

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

Lines changed: 190 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -9,6 +9,7 @@ import {
99
createOptimizedContextStore,
1010
createSessionStores,
1111
loadRecentTurns,
12+
versionedBaseName,
1213
} from "./optimized-context-store.js";
1314
import { segmentFileName, listSegmentFiles } from "./incremental-jsonl.js";
1415

@@ -597,6 +598,195 @@ describe("createSessionStores", () => {
597598
});
598599
});
599600

601+
describe("concurrent sessions write disjoint transcripts (CL-10372)", () => {
602+
test("a second session sharing the dir never overwrites the first session's turns", async () => {
603+
const dir = tempDir();
604+
const storeA = await createOptimizedContextStore(dir);
605+
const storeB = await createOptimizedContextStore(dir);
606+
607+
// Session A lands a 107-record transcript (the ticket's verified size).
608+
const turnsA = Array.from({ length: 107 }, (_, i) => turn(`a-${i}`));
609+
await storeA.writeTurns([...turnsA]);
610+
await storeA.writeMetadata(EMPTY_CHECKPOINT_METADATA);
611+
await storeA.commit({ message: "session A" });
612+
613+
// Session B starts concurrently on the same dir with its own transcript.
614+
const turnsB = Array.from({ length: 7 }, (_, i) => turn(`b-${i}`));
615+
await storeB.writeTurns([...turnsB]);
616+
await storeB.writeMetadata(EMPTY_CHECKPOINT_METADATA);
617+
await storeB.commit({ message: "session B" });
618+
619+
// A's transcript must survive intact: no overwrite.
620+
expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual(
621+
turnTexts(turnsA),
622+
);
623+
624+
// B's records must survive in a disjoint, versioned file: no lost records.
625+
const versioned = versionedTurnsFiles(dir);
626+
expect(versioned.length).toBeGreaterThan(0);
627+
const bOnDisk = versioned.flatMap((name) =>
628+
turnTexts(readTurnsFromFile(path.join(dir, name))),
629+
);
630+
expect(bOnDisk).toEqual(turnTexts(turnsB));
631+
632+
// Disjoint: every record from both sessions survives exactly once.
633+
const all = [
634+
...turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE))),
635+
...bOnDisk,
636+
];
637+
expect(new Set(all).size).toBe(turnsA.length + turnsB.length);
638+
});
639+
640+
test("same-session resume with a differing first prompt keeps writing turns.jsonl (CL-10372 regression)", async () => {
641+
const dir = tempDir();
642+
643+
// Owner session lands a committed transcript, then is resumed by a fresh
644+
// store in a new process (the reactor writes the materialized prompt in
645+
// pre-inference BEFORE any writeTurns; the prompt legitimately differs
646+
// from the on-disk turns when context transforms / ephemeral turns apply).
647+
const owner = await createOptimizedContextStore(dir);
648+
const live = [turn("a"), turn("b")];
649+
await owner.writeTurns([...live]);
650+
await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA);
651+
await owner.commit({ message: "owner" });
652+
653+
const resumed = await createOptimizedContextStore(dir);
654+
await resumed.writePrompt([turn("a"), turn("b"), turn("ephemeral")]);
655+
const next = [...live, turn("c")];
656+
await resumed.writeTurns([...next]);
657+
await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA);
658+
await resumed.commit({ message: "resume append" });
659+
660+
// No versioned base: this store is the session owner resuming, so it
661+
// keeps owning turns.jsonl and the new turn is visible to load().
662+
expect(versionedTurnsFiles(dir)).toEqual([]);
663+
expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual(
664+
turnTexts(next),
665+
);
666+
expect(turnTexts((await resumed.load()).turns)).toEqual(turnTexts(next));
667+
});
668+
669+
test("same-session compaction rewrite on resume is canonical, not diverted", async () => {
670+
const dir = tempDir();
671+
const owner = await createOptimizedContextStore(dir, {
672+
ownerId: "session-S",
673+
});
674+
const original = [turn("a"), turn("b"), turn("c"), turn("d")];
675+
await owner.writeTurns([...original]);
676+
await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA);
677+
await owner.commit({ message: "owner" });
678+
679+
// The same session resumes (same owner id) and its FIRST write is a
680+
// compaction that replaces the head. Ownership is by marker, not content
681+
// shape: this is the owner, so the rewrite is canonical on turns.jsonl.
682+
const resumed = await createOptimizedContextStore(dir, {
683+
ownerId: "session-S",
684+
});
685+
const compacted = [turn("[Compacted prior context]"), turn("c"), turn("d")];
686+
await resumed.writeTurns([...compacted]);
687+
await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA);
688+
await resumed.commit({ message: "resume compact" });
689+
690+
expect(versionedTurnsFiles(dir)).toEqual([]);
691+
expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual(
692+
turnTexts(compacted),
693+
);
694+
expect(turnTexts((await resumed.load()).turns)).toEqual(
695+
turnTexts(compacted),
696+
);
697+
});
698+
699+
test("interleaved first writes cannot both claim the shared base", async () => {
700+
const dir = tempDir();
701+
const storeA = await createOptimizedContextStore(dir);
702+
const storeB = await createOptimizedContextStore(dir);
703+
const turnsA = Array.from({ length: 50 }, (_, i) => turn(`a-${i}`));
704+
const turnsB = Array.from({ length: 50 }, (_, i) => turn(`b-${i}`));
705+
706+
await Promise.all([
707+
(async () => {
708+
await storeA.writeTurns([...turnsA]);
709+
await storeA.writeMetadata(EMPTY_CHECKPOINT_METADATA);
710+
await storeA.commit({ message: "A" });
711+
})(),
712+
(async () => {
713+
await storeB.writeTurns([...turnsB]);
714+
await storeB.writeMetadata(EMPTY_CHECKPOINT_METADATA);
715+
await storeB.commit({ message: "B" });
716+
})(),
717+
]);
718+
719+
// The dir lock serializes the first writes: exactly one store owns the
720+
// shared base and the other lands in a single disjoint versioned file.
721+
const shared = turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)));
722+
const versioned = versionedTurnsFiles(dir);
723+
expect(versioned).toHaveLength(1);
724+
const other = turnTexts(
725+
readTurnsFromFile(path.join(dir, defined(versioned[0]))),
726+
);
727+
const sharedIsA =
728+
JSON.stringify(shared) === JSON.stringify(turnTexts(turnsA));
729+
const sharedIsB =
730+
JSON.stringify(shared) === JSON.stringify(turnTexts(turnsB));
731+
expect(sharedIsA || sharedIsB).toBe(true);
732+
expect(other).toEqual(sharedIsA ? turnTexts(turnsB) : turnTexts(turnsA));
733+
expect(new Set([...shared, ...other]).size).toBe(
734+
turnsA.length + turnsB.length,
735+
);
736+
});
737+
738+
test("crash before the divert lands leaves no orphaned versioned file", async () => {
739+
const dir = tempDir();
740+
const owner = await createOptimizedContextStore(dir);
741+
await owner.writeTurns([turn("a"), turn("b")]);
742+
await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA);
743+
await owner.commit({ message: "owner" });
744+
745+
const foreign = await createOptimizedContextStore(dir);
746+
await foreign.writeTurns([turn("x"), turn("y")]);
747+
// Simulated crash before commit: the divert is staged in memory only, so
748+
// no versioned file may exist on disk yet.
749+
expect(versionedTurnsFiles(dir)).toEqual([]);
750+
751+
// A later plain resume (prefix-extension, no stable id) sees no orphan
752+
// and keeps the shared transcript canonical.
753+
const resumed = await createOptimizedContextStore(dir);
754+
await resumed.writeTurns([turn("a"), turn("b"), turn("c")]);
755+
await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA);
756+
await resumed.commit({ message: "resume" });
757+
expect(versionedTurnsFiles(dir)).toEqual([]);
758+
expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual([
759+
"a",
760+
"b",
761+
"c",
762+
]);
763+
});
764+
765+
test("versioned base names are unique even within the same millisecond", () => {
766+
const names = Array.from({ length: 100 }, () =>
767+
versionedBaseName(TURNS_FILE),
768+
);
769+
expect(new Set(names).size).toBe(100);
770+
for (const name of names) {
771+
expect(/^turns-\d+-[0-9a-f]+\.jsonl$/.test(name)).toBe(true);
772+
}
773+
});
774+
});
775+
776+
function versionedTurnsFiles(dir: string): string[] {
777+
return fs
778+
.readdirSync(dir)
779+
.filter((name) => /^turns-\d+-[0-9a-f]+\.jsonl$/.test(name));
780+
}
781+
782+
function readTurnsFromFile(fullPath: string): ConversationTurn[] {
783+
const text = fs.readFileSync(fullPath, "utf8");
784+
return text
785+
.split("\n")
786+
.filter((line) => line.length > 0)
787+
.map((line) => JSON.parse(line) as ConversationTurn);
788+
}
789+
600790
async function gitLsTree(dir: string): Promise<string[]> {
601791
const proc = Bun.spawn(
602792
["git", "-C", dir, "ls-tree", "-r", "--name-only", "HEAD"],

0 commit comments

Comments
 (0)