diff --git a/packages/session-persistence/src/coordinator.ts b/packages/session-persistence/src/coordinator.ts index f35685c677..5371873097 100644 --- a/packages/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/src/coordinator.ts @@ -2,21 +2,20 @@ * The backend-agnostic write-path orchestration shared by every first-party * {@link SessionPersistence} backend. * - * The two durable backends (`dsh-session-persistence-jsonl` over file bytes, - * `dsh-session-persistence-sqlite` over `node:sqlite` rows) were byte-identical - * — or same-algorithm — for ALL of their orchestration: the in-memory - * bookkeeping (the per-id state, the write-behind buffers, the per-id - * serialization chains, the per-session init promises), the `session/event` → - * buffer → `session/flush` drain, lazy materialization, crash-tail repair on - * load, the four `session/created` adoption cases (new / HMR-adopt / collision / - * ownerless-claim), and dispose-time quiescence. Only the STORAGE primitives - * differed (write bytes vs. INSERT rows). {@link PersistenceCoordinator} owns - * the orchestration once; a backend supplies the storage primitives as a small + * Every durable backend needs the same orchestration: the in-memory bookkeeping + * (the per-id state, the write-behind buffers, the per-id serialization chains, + * the per-session init promises), the `session/event` → buffer → `session/flush` + * drain, lazy materialization, crash-tail repair on load, the four + * `session/created` adoption cases (new / HMR-adopt / collision / + * ownerless-claim), and dispose-time quiescence. Only the STORAGE primitives are + * backend-specific (file bytes for `dsh-session-persistence-jsonl`, `node:sqlite` + * rows for `dsh-session-persistence-sqlite`). {@link PersistenceCoordinator} owns + * the orchestration; a backend supplies the storage primitives as a small * {@link PersistenceBackend} hook object. * - * The abstract {@link SessionPersistence} service's public API is unchanged: a - * backend still IS a `SessionPersistence` (its six public methods delegate to a - * coordinator it composes), so a third-party backend MAY implement the service + * The abstract {@link SessionPersistence} service's public API is independent of + * this: a backend IS a `SessionPersistence` (its six public methods delegate to + * a coordinator it composes), so a third-party backend MAY implement the service * directly without using the coordinator at all. * * See the write-coordinator RFC (docs/rfc/implemented/2026-06-18-shared-persistence-write-coordinator.md) @@ -115,7 +114,19 @@ interface SessionState { meta: SessionHeader /** The next seq the backend expects to append (the stored log length). */ cursor: number - /** Whether the session has been physically materialized. */ + /** + * Whether the backend has physically written this session (a JSONL file / + * SQLite row exists). `create()` registers state LAZILY — cursor 0, + * materialized false, nothing on disk — so an empty session leaves no + * artifact and the FIRST `appendBatch` writes the header + its events in ONE + * transaction (the "a row exists ⇔ it has events" invariant `has`/`list` + * rely on; a separate up-front materialize could crash leaving a row with + * zero events). The flag is the only signal that distinguishes a session + * registered-but-never-written from one durably present, which two callers + * need: `has()` (lazy-but-unwritten is not yet durable) and the reclaim path + * (an abandoned id with no artifact AND no buffered events is free to reuse; + * a materialized one is a real collision). + */ materialized: boolean /** * The live Session this state was bound to via `onCreated`, if any. State @@ -450,9 +461,18 @@ export class PersistenceCoordinator { if (tracked.owner === session) return if (tracked.owner === undefined) { // Ownerless state from the public create()/load() API. The FIRST live - // session claims it — but ONLY if its seed reproduces the persisted - // prefix (else a fresh, unrelated session reusing the id would have its - // seq 0..cursor-1 events filtered as already-written and grafted on). + // session claims it — but ONLY if BOTH the cwd scope and the seed match. + // The cwd guard mirrors case-2's cwd-scoped loadLive(): a same-id + // ownerless artifact at a DIFFERENT cwd is a collision, not a claim + // (claiming it would append the live cwd's events under the stored + // header's cwd, the exact cross-cwd corruption the loadLive scope + // prevents). The seed guard then ensures the live events reproduce the + // persisted prefix (else a fresh, unrelated session reusing the id would + // have its seq 0..cursor-1 events filtered as already-written and + // grafted on). + if (tracked.meta.cwd !== session.header.cwd) { + throw new Error(`session "${id}" is already persisted at a different cwd (persisted: ${String(tracked.meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`) + } if (!await this.seedMatchesPersisted(id, seed, tracked.cursor)) { throw new Error(`session "${id}" is already persisted with ${tracked.cursor} event(s) that do not match this live session (id collision)`) } diff --git a/packages/session-persistence/tests/coordinator-contract.ts b/packages/session-persistence/tests/coordinator-contract.ts index 24a8f4c0cf..7769e6f5ec 100644 --- a/packages/session-persistence/tests/coordinator-contract.ts +++ b/packages/session-persistence/tests/coordinator-contract.ts @@ -68,6 +68,7 @@ export interface CoordinatorFixture { /** A constant absolute cwd; jsonl keys directories off it, memory/sqlite ignore it. */ const WORK = '/w' +const OTHER = '/other' /** The per-session init map a backend exposes for white-box init awaits. */ function inits(persistence: SessionPersistence): Map> { @@ -535,6 +536,58 @@ export function runCoordinatorContract(name: string, makeFixture: () => Promise< } }) + it('a live session at a DIFFERENT cwd cannot claim cursor-0 ownerless state (cwd scope)', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + try { + // create() registers ownerless state at cwd /a (cursor 0 — claims would + // otherwise match trivially on the seed). + await ctx.sessionPersistence.create(meta('wrong-cwd-claim', OTHER)) + // A live session reusing the id but at cwd WORK must NOT claim it — the + // cwd scope is the fence (without it, WORK events would append under the + // OTHER header). Rejected as a collision. + const live = ctx.sessions.create('wrong-cwd-claim', { seed: oneTurnLog(), meta: { cwd: WORK } }) + await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/) + } finally { + await fiber.dispose() + await fix.cleanup() + } + }) + + it('a live session at a DIFFERENT cwd cannot claim loaded-prefix ownerless state (cwd scope)', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + try { + // Materialize + load at cwd OTHER (ownerless, cursor = 6). + await ctx.sessionPersistence.create(meta('wrong-cwd-load', OTHER)) + await ctx.sessionPersistence.append(SessionId('wrong-cwd-load'), oneTurnLog()) + const { events } = await ctx.sessionPersistence.load(SessionId('wrong-cwd-load')) + // A live session whose SEED matches the loaded prefix but whose cwd is + // WORK must still be rejected — the cwd guard runs before the seed check. + const live = ctx.sessions.create('wrong-cwd-load', { seed: events, meta: { cwd: WORK } }) + await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/) + } finally { + await fiber.dispose() + await fix.cleanup() + } + }) + + it('a no-cwd ownerless state cannot be claimed by a live session WITH a cwd (cwd scope, undefined side)', async () => { + const fix = await makeFixture() + const { ctx, fiber } = await freshCtx(fix) + try { + // Ownerless state created WITHOUT a cwd (the no-cwd bucket). + await ctx.sessionPersistence.create(meta('no-cwd-state')) + // A live session reusing the id but WITH cwd WORK is a cwd mismatch + // (undefined vs WORK) and must be rejected. + const live = ctx.sessions.create('no-cwd-state', { seed: oneTurnLog(), meta: { cwd: WORK } }) + await expect(inits(ctx.sessionPersistence).get(live)).rejects.toThrow(/different cwd|id collision/) + } finally { + await fiber.dispose() + await fix.cleanup() + } + }) + // --- append adopts a storage-only session (fresh instance, no prior create/load) --- it('append adopts a storage-only session (fresh instance) and continues the seq', async () => {