Skip to content

Commit 30ce6dd

Browse files
committed
refactor(session): drop transcript-ownership narration
1 parent 6c83635 commit 30ce6dd

3 files changed

Lines changed: 16 additions & 81 deletions

File tree

‎src/session/assemble-runtime.ts‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -675,10 +675,8 @@ 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-
// 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.
678+
// Session id is the stable owner so a resume's compaction stays on the
679+
// shared transcript.
682680
const { storage, audit } = await createSessionStores(workdir, {
683681
ownerId: sessionId,
684682
});

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

Lines changed: 4 additions & 22 deletions
Original file line numberDiff line numberDiff line change
@@ -598,52 +598,45 @@ describe("createSessionStores", () => {
598598
});
599599
});
600600

601-
describe("concurrent sessions write disjoint transcripts (CL-10372)", () => {
601+
describe("concurrent sessions write disjoint transcripts", () => {
602602
test("a second session sharing the dir never overwrites the first session's turns", async () => {
603603
const dir = tempDir();
604604
const storeA = await createOptimizedContextStore(dir);
605605
const storeB = await createOptimizedContextStore(dir);
606606

607-
// Session A lands a 107-record transcript (the ticket's verified size).
608607
const turnsA = Array.from({ length: 107 }, (_, i) => turn(`a-${i}`));
609608
await storeA.writeTurns([...turnsA]);
610609
await storeA.writeMetadata(EMPTY_CHECKPOINT_METADATA);
611610
await storeA.commit({ message: "session A" });
612611

613-
// Session B starts concurrently on the same dir with its own transcript.
614612
const turnsB = Array.from({ length: 7 }, (_, i) => turn(`b-${i}`));
615613
await storeB.writeTurns([...turnsB]);
616614
await storeB.writeMetadata(EMPTY_CHECKPOINT_METADATA);
617615
await storeB.commit({ message: "session B" });
618616

619-
// A's transcript must survive intact: no overwrite.
620617
expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual(
621618
turnTexts(turnsA),
622619
);
623620

624-
// B's records must survive in a disjoint, versioned file: no lost records.
625621
const versioned = versionedTurnsFiles(dir);
626622
expect(versioned.length).toBeGreaterThan(0);
627623
const bOnDisk = versioned.flatMap((name) =>
628624
turnTexts(readTurnsFromFile(path.join(dir, name))),
629625
);
630626
expect(bOnDisk).toEqual(turnTexts(turnsB));
631627

632-
// Disjoint: every record from both sessions survives exactly once.
633628
const all = [
634629
...turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE))),
635630
...bOnDisk,
636631
];
637632
expect(new Set(all).size).toBe(turnsA.length + turnsB.length);
638633
});
639634

640-
test("same-session resume with a differing first prompt keeps writing turns.jsonl (CL-10372 regression)", async () => {
635+
test("same-session resume with a differing first prompt keeps writing turns.jsonl", async () => {
641636
const dir = tempDir();
642637

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).
638+
// Reactor writes the prompt before writeTurns; that prompt may differ
639+
// from on-disk turns.
647640
const owner = await createOptimizedContextStore(dir);
648641
const live = [turn("a"), turn("b")];
649642
await owner.writeTurns([...live]);
@@ -657,8 +650,6 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => {
657650
await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA);
658651
await resumed.commit({ message: "resume append" });
659652

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().
662653
expect(versionedTurnsFiles(dir)).toEqual([]);
663654
expect(turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)))).toEqual(
664655
turnTexts(next),
@@ -676,9 +667,6 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => {
676667
await owner.writeMetadata(EMPTY_CHECKPOINT_METADATA);
677668
await owner.commit({ message: "owner" });
678669

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.
682670
const resumed = await createOptimizedContextStore(dir, {
683671
ownerId: "session-S",
684672
});
@@ -716,8 +704,6 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => {
716704
})(),
717705
]);
718706

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.
721707
const shared = turnTexts(readTurnsFromFile(path.join(dir, TURNS_FILE)));
722708
const versioned = versionedTurnsFiles(dir);
723709
expect(versioned).toHaveLength(1);
@@ -744,12 +730,8 @@ describe("concurrent sessions write disjoint transcripts (CL-10372)", () => {
744730

745731
const foreign = await createOptimizedContextStore(dir);
746732
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.
749733
expect(versionedTurnsFiles(dir)).toEqual([]);
750734

751-
// A later plain resume (prefix-extension, no stable id) sees no orphan
752-
// and keeps the shared transcript canonical.
753735
const resumed = await createOptimizedContextStore(dir);
754736
await resumed.writeTurns([turn("a"), turn("b"), turn("c")]);
755737
await resumed.writeMetadata(EMPTY_CHECKPOINT_METADATA);

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

Lines changed: 10 additions & 55 deletions
Original file line numberDiff line numberDiff line change
@@ -34,7 +34,6 @@ const MANIFEST_FILE = "manifest.jsonl";
3434
const METADATA_FILE = "metadata.json";
3535
const TOOL_OUTPUT_DIR = "tool-output";
3636
const EVIDENCE_ARCHIVE_DIR = "evidence-archive";
37-
/** Per-session ownership marker: records which store authored the shared base. */
3837
const OWNER_MARKER_FILE = "session-owner.json";
3938

4039
const log = getLogger([LOG_NAMESPACE_ROOT, "session", "context-store"]);
@@ -87,13 +86,8 @@ function sanitizeCallId(callId: string): string {
8786
}
8887

8988
/**
90-
* Unique versioned base name for a transcript file (e.g.
91-
* `turns-1730000000000-a1b2c3d4e5f6....jsonl`). A store whose transcript is
92-
* foreign to the on-disk owner writes to one of these instead of the shared
93-
* base, so concurrent sessions keep disjoint files. The `-<epoch>-<uuid>`
94-
* infix never collides with the numbered segment pattern (`turns-0001.jsonl`),
95-
* so the base readers ignore these by design. The UUID suffix keeps names
96-
* unique even when two diverts happen within the same millisecond.
89+
* Epoch+uuid avoids numbered segments (`turns-0001.jsonl`) that base
90+
* readers treat as live history.
9791
*/
9892
export function versionedBaseName(baseName: string): string {
9993
const dot = baseName.lastIndexOf(".");
@@ -360,12 +354,6 @@ async function readBaseTurnsFromDisk(dir: string): Promise<ConversationTurn[]> {
360354
return parseSegmentTurns(text, true, TURNS_FILE, true);
361355
}
362356

363-
/**
364-
* Ownership marker for the shared base. Absent on a fresh/legacy dir; once a
365-
* store writes the shared base first it claims the dir with its owner token,
366-
* so a later store can tell "I authored this transcript" from "a foreign
367-
* store did" without guessing from content shape.
368-
*/
369357
async function readOwnerMarker(
370358
dir: string,
371359
): Promise<{ ownerId: string } | null> {
@@ -567,15 +555,8 @@ export interface SessionStores {
567555
export interface SessionStoreOptions {
568556
signer?: CommitSigner;
569557
/**
570-
* Stable identity of the session this store belongs to (e.g. the session
571-
* id). Stores with the same id on one dir are the same session: resuming
572-
* stores recognize the on-disk ownership marker as their own, so their
573-
* rewrites (compaction) stay canonical on the shared base. Stores with
574-
* different (or no) ids are foreign to each other and divert to disjoint
575-
* versioned files instead of clobbering the shared transcript. When omitted
576-
* the store uses a per-instance random token, so any two stores on one dir
577-
* are treated as foreign unless the incoming transcript extends the on-disk
578-
* history (a plain resume).
558+
* Same id shares the marker; omit for a random token, foreign unless the
559+
* transcript extends history.
579560
*/
580561
ownerId?: string;
581562
}
@@ -594,13 +575,7 @@ export async function createSessionStores(
594575
const base = await createIsogitStore(dir, signer);
595576
const pendingBlobFilepaths = new Set<string>();
596577
const pendingSegmentPaths = new Set<string>();
597-
// Stable per-session owner token (the caller's session id, or a random
598-
// per-instance token). The on-disk ownership marker is compared against it
599-
// to tell "I authored this transcript" from "a foreign store did".
600578
const ownerToken = opts?.ownerId ?? globalThis.crypto.randomUUID();
601-
// Base file each writer owns. A store that detects a foreign on-disk owner
602-
// (a concurrent session sharing the dir) switches both bases to disjoint
603-
// versioned names so its writes never clobber the other session.
604579
let turnsBaseName = TURNS_FILE;
605580
let promptBaseName = PROMPT_FILE;
606581
let versionedBase: string | null = null;
@@ -667,9 +642,7 @@ export async function createSessionStores(
667642
liveTurnRefs = [...turns];
668643
return;
669644
}
670-
// Fresh-store first write: the ownership decision and the first write are
671-
// serialized per dir so two concurrent fresh stores cannot both claim the
672-
// shared base (one becomes owner, the other diverts instead of racing).
645+
// First claim is per-dir so two fresh stores cannot both own the base.
673646
await withResolvedDirLock(dir, async () => {
674647
if (liveTurnRefs !== null) {
675648
if (refPrefixLength(liveTurnRefs, turns) < liveTurnRefs.length) {
@@ -682,39 +655,30 @@ export async function createSessionStores(
682655
}
683656
const marker = await readOwnerMarker(dir);
684657
if (marker === null) {
685-
// No owner yet: this store claims the shared base.
686658
await writeOwnerMarker(dir, ownerToken);
687659
await writeSegmented(writeTurnsSegmented, turns);
688660
liveTurnRefs = [...turns];
689661
return;
690662
}
691663
const live = await loadLiveTurnsFromDisk();
692664
if (marker.ownerId === ownerToken) {
693-
// This store is the session owner (resuming with a stable id): any
694-
// rewrite, including a compaction that replaces history, is canonical
695-
// on the shared base — never divert an owner's own rewrite.
665+
// Matching owner keeps compaction on the shared base, not a divert.
696666
await writeSegmented(writeTurnsSegmented, turns);
697667
liveTurnRefs = [...turns];
698668
return;
699669
}
700670
if (live.length > 0 && contentPrefixLength(live, turns) < live.length) {
701-
// The on-disk transcript was authored by a foreign store and this
702-
// store's transcript is not an extension of it: rewriting the shared
703-
// base would clobber the other session's records. Divert to a
704-
// disjoint versioned base; the write is staged and lands at commit so
705-
// a crash before then leaves no orphaned versioned file.
671+
// Stage until commit so a crash leaves no orphaned versioned file.
706672
divertToVersionedBases();
707673
unpublishedRewrite = [...turns];
708674
return;
709675
}
710-
// Foreign marker but the incoming transcript extends the on-disk history
711-
// (a plain resume without a stable id): continue the shared base.
676+
// Prefix extension with no stable id resumes the shared base.
712677
await writeSegmented(writeTurnsSegmented, turns);
713678
liveTurnRefs = [...turns];
714679
});
715680
}
716681

717-
/** Live turns from disk for a fresh store (base + extra segments). */
718682
async function loadLiveTurnsFromDisk(): Promise<ConversationTurn[]> {
719683
const extraTexts = await readExtraSegmentTexts(dir, TURNS_FILE);
720684
let baseTurns: ConversationTurn[];
@@ -782,20 +746,11 @@ export async function createSessionStores(
782746
}
783747
}
784748
if (versionedBase !== null && liveTurnRefs === null) {
785-
// A divert is staged but has not landed at commit yet. No versioned
786-
// file may be written before commit — a crash would orphan it — so the
787-
// prompt snapshot is deferred to the next cycle (after the commit, the
788-
// versioned base exists and liveTurnRefs is set).
749+
// No versioned file before commit; a crash would orphan it.
789750
return;
790751
}
791752
if (live === null || !promptEqualsLiveTurns(live, turns)) {
792-
// The prompt is a materialized snapshot and legitimately differs from
793-
// the live turns on a normal resume (context transforms, ephemeral
794-
// turns, summaries), so a differing prompt alone never proves a foreign
795-
// store. Ownership is decided by writeTurns: only a foreign owner with a
796-
// divergent transcript diverts to versioned bases, and
797-
// writePromptSegmented is rebound with it. Prompt writes therefore stay
798-
// on the shared base unless that divert already happened.
753+
// A differing prompt is normal on resume and must not divert.
799754
await writeSegmented(writePromptSegmented, turns);
800755
return;
801756
}

0 commit comments

Comments
 (0)