fix(persistence): bind JSONL identity before mutation
JSONL discovered a log by the requested session id but later routed repair and append from the parsed header. A log selected for session A could therefore declare session B and redirect mutation to B. Validate the requested id and exact header-derived cwd-bucket path before returning a stored prefix, reject duplicate ids across buckets, and repeat the id/cwd guards in the coordinator before repair or state publication. Collapse the redundant loadLive hook into loadStored while retaining the existing bucket layout and one-live-writer topology, avoiding flat-layout churn and a locator generic that SQLite and test backends do not need.
This commit is contained in:
@@ -11,7 +11,7 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
|
||||
```
|
||||
|
||||
- The first `.jsonl` line is the immutable `SessionHeader` tagged `{ type: 'session', version, id, cwd?, createdAt, parentSession?, seedLength? }`; every subsequent line is one `SessionEvent` JSON, **verbatim including `assistant/chunk`** so `seq` stays contiguous (`events[i].seq === i`).
|
||||
- Session ids are unvalidated branded strings, so they are percent-encoded to a single safe path segment before use (no traversal, no collision).
|
||||
- Session ids are unvalidated branded strings, so they are injectively encoded as one safe path segment before use (no traversal, no collision).
|
||||
|
||||
## Config
|
||||
|
||||
@@ -23,6 +23,7 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
|
||||
|
||||
## Durability and crash semantics
|
||||
|
||||
- **Bound storage identity.** Lookup requires one matching encoded filename across the cwd buckets, then verifies that the header id equals the requested id and that the header's id/cwd derive the selected path. Listing applies the same path check and rejects duplicate ids. Identity failures occur before repair or append.
|
||||
- **Lazy materialization.** `create(meta)` writes nothing; on the first `append`, the backend writes and `fsync`s a temporary file, publishes it without overwrite via a hard link, then `fsync`s the directory when the host supports it. A created-but-never-appended session leaves nothing on disk and is absent from `list`.
|
||||
- **Append-only.** Committed events (at or below a flushed `turn/end`) are never rewritten. Subsequent appends are line appends at EOF + `fsync`.
|
||||
- **Crash recovery — preserve valid tail work.** `load` keeps the contiguous valid prefix of an interrupted final turn. It truncates from the first unparsable or sequence-gapped uncommitted record, then appends the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md); the same defect at or before the last committed `turn/end` rejects.
|
||||
@@ -30,7 +31,7 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
|
||||
|
||||
## Write path
|
||||
|
||||
The plugin buffers frozen session events and drains them on flush or disposal. A per-session cursor prevents resumed sessions from re-appending stored events, and live sessions are seeded when the plugin loads. Operations for one session are serialized; disposal waits for initialization and the final drain so no write lands after teardown.
|
||||
The plugin buffers frozen session events and drains them on flush or disposal. A per-session cursor prevents resumed sessions from re-appending stored events, and live sessions are seeded when the plugin loads. The owning backend instance serializes operations for one session; disposal waits for initialization and the final drain so no write lands after teardown.
|
||||
|
||||
## Model Experience
|
||||
|
||||
@@ -52,6 +53,6 @@ JSONL storage does not mutate live request prefixes. A resumed loop can reuse pr
|
||||
|
||||
- **Only the current `SESSION_FORMAT_VERSION` (v0) loads** — the on-disk format is pre-release/unstable: a breaking format change is absorbed at v0 and non-current logs are rejected; there is no migration.
|
||||
- **Nothing deletes session files** — logs accumulate under `root` until removed externally (the seam has no deletion surface).
|
||||
- **Single-process assumption** — per-session serialization and the write cursor live in this process; two processes appending to the same `root` are not coordinated.
|
||||
- **One live writer per session** — append and repair are coordinated only inside the owning backend instance. Another backend instance or process must not write the same session until that owner reaches quiescent disposal; initial same-id publication remains collision-safe through the no-overwrite hard link.
|
||||
- **Initial materialization requires hard-link support** — first append uses `link()` so same-id races fail instead of overwriting a committed log; a filesystem that cannot create hard links cannot host this backend.
|
||||
- **Windows cannot `fsync` directory handles through Node** — the backend tolerates only Windows `EPERM` from directory `fsync`; file-content `fsync` remains mandatory, but a crash can lose a newly published directory entry on a host without an equivalent directory-sync primitive.
|
||||
|
||||
@@ -9,7 +9,7 @@
|
||||
import { Context } from 'cordis'
|
||||
import z from 'schemastery'
|
||||
import { open, mkdir, readFile, readdir, link, rm, truncate } from 'node:fs/promises'
|
||||
import { dirname, resolve } from 'node:path'
|
||||
import { dirname, join, resolve } from 'node:path'
|
||||
import { randomBytes } from 'node:crypto'
|
||||
import {
|
||||
SessionPersistence, PersistenceCoordinator,
|
||||
@@ -97,28 +97,19 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
|
||||
/** Read a stored prefix by id across all cwd buckets when cwd is unknown. */
|
||||
async loadStored(id: SessionId): Promise<StoredPrefix<number> | undefined> {
|
||||
const file = await this.findLog(id)
|
||||
if (file === undefined) return undefined
|
||||
return this.readPrefix(file.path)
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a stored prefix within one cwd for HMR adoption. `undefined` names the
|
||||
* no-cwd bucket rather than an unknown cwd, so this never scans other buckets.
|
||||
*/
|
||||
async loadLive(id: SessionId, cwd: string | undefined): Promise<StoredPrefix<number> | undefined> {
|
||||
const path = logPath(this.root, cwd, id)
|
||||
if (!await this.exists(path)) return undefined
|
||||
return this.readPrefix(path)
|
||||
const path = await this.findLog(id)
|
||||
if (path === undefined) return undefined
|
||||
return this.readPrefix(path, id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a stored prefix and convert torn-tail state to the byte offset the
|
||||
* coordinator can round-trip without knowing the file format.
|
||||
*/
|
||||
private async readPrefix(path: string): Promise<StoredPrefix<number>> {
|
||||
private async readPrefix(path: string, expectedId: SessionId): Promise<StoredPrefix<number>> {
|
||||
const buffer = await readFile(path)
|
||||
const { meta, events, committedBytes } = scanLog(buffer)
|
||||
this.assertStoredIdentity(path, meta, expectedId)
|
||||
return {
|
||||
meta,
|
||||
events,
|
||||
@@ -145,16 +136,23 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
if (closers.length > 0) await this.appendLines(meta, closers)
|
||||
}
|
||||
|
||||
/** List all stored sessions' metadata (header line only — no full-log parse). */
|
||||
/** List valid unique stored sessions' metadata (header line only — no full-log parse). */
|
||||
async list(): Promise<SessionHeader[]> {
|
||||
const metas: SessionHeader[] = []
|
||||
const ids = new Set<SessionId>()
|
||||
for (const dir of await this.listCwdDirs()) {
|
||||
for (const name of await this.listJsonl(dir)) {
|
||||
const path = join(dir, name)
|
||||
// Read only headers so listing scales with session count, not log size.
|
||||
const first = await this.readFirstLine(`${dir}/${name}`)
|
||||
const first = await this.readFirstLine(path)
|
||||
if (first === undefined) continue // empty/half-written file
|
||||
const meta = parseHeaderMeta(first)
|
||||
if (meta === undefined) continue // not a session header
|
||||
this.assertStoredIdentity(path, meta)
|
||||
if (ids.has(meta.id)) {
|
||||
throw new Error(`duplicate JSONL session id "${meta.id}" appears in multiple cwd buckets`)
|
||||
}
|
||||
ids.add(meta.id)
|
||||
metas.push(meta)
|
||||
}
|
||||
}
|
||||
@@ -292,28 +290,41 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Find a session by id across cwd buckets for resume. Cwd-scoped HMR adoption
|
||||
* bypasses this scan so a no-cwd session cannot claim another bucket.
|
||||
*/
|
||||
private async findLog(id: SessionId): Promise<{ path: string; cwd: string | undefined } | undefined> {
|
||||
/** Find the unique physical log for an id across every cwd bucket. */
|
||||
private async findLog(id: SessionId): Promise<string | undefined> {
|
||||
const target = encodeSegment(id) + '.jsonl'
|
||||
const matches: string[] = []
|
||||
for (const dir of await this.listCwdDirs()) {
|
||||
const path = `${dir}/${target}`
|
||||
if (await this.exists(path)) {
|
||||
// Recover the cwd from the header so the caller has the session's bucket.
|
||||
const { meta } = scanLog(await readFile(path))
|
||||
return { path, cwd: meta.cwd }
|
||||
}
|
||||
const path = join(dir, target)
|
||||
if (await this.exists(path)) matches.push(path)
|
||||
}
|
||||
if (matches.length > 1) {
|
||||
throw new Error(`duplicate JSONL session id "${id}" appears in multiple cwd buckets`)
|
||||
}
|
||||
return matches[0]
|
||||
}
|
||||
|
||||
/** Reject metadata that does not identify the selected physical log. */
|
||||
private assertStoredIdentity(path: string, meta: SessionHeader, expectedId?: SessionId): void {
|
||||
if (expectedId !== undefined && meta.id !== expectedId) {
|
||||
throw new Error(`corrupt session log "${path}": requested id "${expectedId}" does not match header id "${meta.id}"`)
|
||||
}
|
||||
let expectedPath: string
|
||||
try {
|
||||
expectedPath = logPath(this.root, meta.cwd, meta.id)
|
||||
} catch (error) {
|
||||
throw new Error(`corrupt session log "${path}": header id cannot name a storage path`, { cause: error })
|
||||
}
|
||||
if (path !== expectedPath) {
|
||||
throw new Error(`corrupt session log "${path}": header id "${meta.id}" and cwd belong at "${expectedPath}"`)
|
||||
}
|
||||
return undefined
|
||||
}
|
||||
|
||||
/** The cwd-bucket directories under the root (absolute paths). */
|
||||
private async listCwdDirs(): Promise<string[]> {
|
||||
try {
|
||||
const entries = await readdir(this.root, { withFileTypes: true })
|
||||
return entries.filter(e => e.isDirectory()).map(e => `${this.root}/${e.name}`)
|
||||
return entries.filter(e => e.isDirectory()).map(e => join(this.root, e.name))
|
||||
} catch (error) {
|
||||
// Only an absent root means no sessions; rethrow every other I/O failure.
|
||||
if (isENOENT(error)) return []
|
||||
|
||||
@@ -21,6 +21,15 @@ function mutableHeader(header: SessionHeader): MutableSessionHeader {
|
||||
return header
|
||||
}
|
||||
|
||||
/** Rewrite only a stored header while preserving every event byte below it. */
|
||||
async function rewriteHeader(path: string, update: (header: Record<string, unknown>) => void): Promise<void> {
|
||||
const lines = (await readFile(path, 'utf8')).split('\n')
|
||||
const header = JSON.parse(lines[0] as string) as Record<string, unknown>
|
||||
update(header)
|
||||
lines[0] = JSON.stringify(header)
|
||||
await writeFile(path, lines.join('\n'))
|
||||
}
|
||||
|
||||
async function expectParallelFlushError(promise: Promise<unknown>, message: RegExp): Promise<void> {
|
||||
try {
|
||||
await promise
|
||||
@@ -393,6 +402,31 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
|
||||
expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
|
||||
})
|
||||
|
||||
it('rejects a mismatched header before repairing either session log', async () => {
|
||||
const a = meta('identity-a', '/same')
|
||||
const b = meta('identity-b', '/same')
|
||||
await ctx.sessionPersistence.create(a)
|
||||
await ctx.sessionPersistence.append(a.id, [{
|
||||
type: 'turn/start',
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
|
||||
}])
|
||||
await ctx.sessionPersistence.create(b)
|
||||
await ctx.sessionPersistence.append(b.id, oneTurnLog())
|
||||
|
||||
const aPath = logPath(root, a.cwd, a.id)
|
||||
const bPath = logPath(root, b.cwd, b.id)
|
||||
await rewriteHeader(aPath, (header) => { header.id = b.id })
|
||||
const beforeA = await readFile(aPath)
|
||||
const beforeB = await readFile(bPath)
|
||||
|
||||
await expect(ctx.sessionPersistence.load(a.id))
|
||||
.rejects.toThrow(/requested id "identity-a" does not match header id "identity-b"/)
|
||||
expect(await readFile(aPath)).toEqual(beforeA)
|
||||
expect(await readFile(bPath)).toEqual(beforeB)
|
||||
})
|
||||
|
||||
it('rejects a re-append of an already-stored seq', async () => {
|
||||
const m = meta('reappend')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
@@ -599,6 +633,28 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
expect(ids).toContain('big')
|
||||
})
|
||||
|
||||
it('list rejects a header whose cwd does not identify its physical log', async () => {
|
||||
const m = meta('misplaced', '/stored')
|
||||
await ctx.sessionPersistence.create(m)
|
||||
await ctx.sessionPersistence.append(m.id, oneTurnLog())
|
||||
await rewriteHeader(logPath(root, m.cwd, m.id), (header) => { header.cwd = '/elsewhere' })
|
||||
|
||||
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/and cwd belong at/)
|
||||
})
|
||||
|
||||
it('load and list reject one id materialized in multiple cwd buckets', async () => {
|
||||
const id = SessionId('duplicate')
|
||||
for (const cwd of ['/a', '/b']) {
|
||||
const m = meta(id, cwd)
|
||||
await mkdir(sessionDir(root, cwd), { recursive: true })
|
||||
const content = [JSON.stringify(toHeaderLine(m)), ...oneTurnLog().map(event => JSON.stringify(event))].join('\n') + '\n'
|
||||
await writeFile(logPath(root, cwd, id), content)
|
||||
}
|
||||
|
||||
await expect(ctx.sessionPersistence.load(id)).rejects.toThrow(/appears in multiple cwd buckets/)
|
||||
await expect(ctx.sessionPersistence.list()).rejects.toThrow(/appears in multiple cwd buckets/)
|
||||
})
|
||||
|
||||
it('a DIFFERENT live session object reusing a disposed id gets its own init (no stale cache)', async () => {
|
||||
// Session A materializes a log under id "reuse".
|
||||
const sessFiberA = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
@@ -619,18 +675,16 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
await expect(ctx.sessions.flush(b)).rejects.toThrow(/already bound to a different live session|already has a persisted log on disk/)
|
||||
})
|
||||
|
||||
it('a NO-CWD live session does NOT cross-cwd-adopt a same-id log from a real cwd bucket (loadLive is scope-exact)', async () => {
|
||||
it('a no-cwd live session cannot adopt a same-id log from another cwd', async () => {
|
||||
// Backend 1: materialize a log under id "x" in the cwd "/w" bucket, then
|
||||
// dispose the WHOLE backend (so backend 2 mounts with an EMPTY states map —
|
||||
// the HMR/reload path where onCreated goes through loadLive, not a tracked
|
||||
// collision).
|
||||
// the HMR/reload path with no tracked collision state).
|
||||
await ctx.sessionPersistence.create(meta('x', '/w'))
|
||||
await ctx.sessionPersistence.append(SessionId('x'), oneTurnLog())
|
||||
await ctx.fiber.dispose()
|
||||
|
||||
// Backend 2 creates a no-cwd session whose id exists only in `/w`. Exact `loadLive(id,
|
||||
// undefined)` must not adopt across buckets; the any-cwd collision check then rejects instead
|
||||
// of grafting no-cwd events onto a log with mismatched cwd.
|
||||
// Backend 2 creates a no-cwd session whose id exists only in `/w`. The
|
||||
// stored cwd check rejects instead of grafting no-cwd events onto that log.
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
await ctx2.plugin(SessionPersistenceJsonl, { root })
|
||||
@@ -638,7 +692,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
await ctx2.plugin(Object.assign((inner: Context) => {
|
||||
b = inner.sessions.create(SessionId('x')) // no cwd
|
||||
}, { inject: ['sessions'] }))
|
||||
await expect(ctx2.sessions.flush(b)).rejects.toThrow(/already has a persisted log on disk/)
|
||||
await expect(ctx2.sessions.flush(b)).rejects.toThrow(/different cwd|id collision/)
|
||||
|
||||
// The "/w" log is untouched — no no-cwd events were grafted onto it, and no
|
||||
// `_no-cwd` log for "x" was created.
|
||||
@@ -706,9 +760,7 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
it('loadLive surfaces a non-ENOENT lookup error (ENOTDIR) instead of reporting absent', async () => {
|
||||
// A non-ENOENT per-id open error must surface rather than become "not found" and permit false
|
||||
// live adoption. Making the cwd bucket a regular file forces ENOTDIR for its child log path.
|
||||
it('materialization surfaces a cwd-bucket storage fault', async () => {
|
||||
const cwd = '/x'
|
||||
const ctx2 = new Context()
|
||||
await ctx2.plugin(SessionStore)
|
||||
@@ -717,8 +769,9 @@ describe('SessionPersistenceJsonl: edge cases', () => {
|
||||
let s!: Session
|
||||
await ctx2.plugin(Object.assign((inner: Context) => {
|
||||
s = inner.sessions.create(SessionId('exists-fault'), { meta: { cwd } })
|
||||
appendClosedTurn(s)
|
||||
}, { inject: ['sessions'] }))
|
||||
await expect(ctx2.sessions.flush(s)).rejects.toThrow(/ENOTDIR/)
|
||||
await expect(ctx2.sessions.flush(s)).rejects.toThrow(/EEXIST|ENOTDIR/)
|
||||
await ctx2.fiber.dispose()
|
||||
})
|
||||
|
||||
|
||||
@@ -144,11 +144,6 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
|
||||
return this.readPrefix(id)
|
||||
}
|
||||
|
||||
/** Read a stored prefix; `cwd` is ignored (the id is globally unique in SQLite). */
|
||||
loadLive(id: SessionId, _cwd: string | undefined): Promise<StoredPrefix<number> | undefined> {
|
||||
return this.readPrefix(id)
|
||||
}
|
||||
|
||||
/**
|
||||
* Read a session's row + ordered events into a {@link StoredPrefix}. The
|
||||
* torn-tail marker is the seq from which a never-committed tail must be deleted
|
||||
|
||||
@@ -34,14 +34,13 @@ The `PersistenceBackend<TornMarker>` hooks (the only seam between the coordinato
|
||||
| Hook | Role |
|
||||
|---|---|
|
||||
| `name` | Backend label for the dispose-failure `AggregateError`. |
|
||||
| `loadStored(id)` | Read a stored prefix by id, scanning ANY storage scope. Used by resume/load and, via `!== undefined`, the create-collision probe. Returns an opaque `tornMarker` iff a torn tail must be truncated. |
|
||||
| `loadLive(id, cwd)` | Read a stored prefix SCOPED to `cwd` (HMR live-adoption must only adopt a log at the SAME cwd; a same-id log elsewhere is a collision, not a resume). A globally-unique-id backend ignores `cwd`. |
|
||||
| `loadStored(id)` | Read a stored prefix by id across every storage scope. Used by resume/load, live adoption, and the create-collision probe. Returned metadata identifies `id`; an opaque `tornMarker` is present iff a torn tail must be truncated. |
|
||||
| `appendBatch(meta, events, isMaterialized)` | Durably append a contiguous batch, lazily materializing ATOMICALLY when not yet materialized. |
|
||||
| `commitRepair(meta, tornMarker, closers)` | Make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined` — a marker may be falsy, e.g. seq/offset `0`) and append `closers`. NOT required to be atomic. Used by load (truncate + closers) and live-adoption (truncate only). |
|
||||
| `list()` | List all stored metadata. |
|
||||
| `close?()` | Optional lifecycle teardown (e.g. close a db handle), awaited after the dispose drain. |
|
||||
|
||||
The `tornMarker` is fully OPAQUE: the coordinator only tests `!== undefined` and round-trips it to `commitRepair`, never inspecting its value (the JSONL backend uses the byte offset to truncate to, the SQLite backend the seq to delete from). The public `SessionPersistence` service shape is unchanged, so a third-party backend MAY still implement the abstract service directly without the coordinator. See [the write-coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
|
||||
The coordinator asserts the stored id and compares stored/live cwd before repair or live adoption. The `tornMarker` is fully OPAQUE: the coordinator only tests `!== undefined` and round-trips it to `commitRepair`, never inspecting its value (the JSONL backend uses the byte offset to truncate to, the SQLite backend the seq to delete from). The public `SessionPersistence` service shape is unchanged, so a third-party backend MAY still implement the abstract service directly without the coordinator. See [the write-coordinator Agent Note](../../../.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md).
|
||||
|
||||
## Testing backends
|
||||
|
||||
|
||||
@@ -35,22 +35,14 @@ export interface PersistenceBackend<TornMarker = unknown> {
|
||||
readonly name: string
|
||||
|
||||
/**
|
||||
* Read a stored prefix by id, scanning ANY storage scope (for JSONL: every
|
||||
* cwd bucket). Returns `undefined` if no stored artifact exists. Used by
|
||||
* resume/load, and — via `!== undefined` — by the create-collision probe.
|
||||
* The returned `tornMarker` is present iff there is a torn tail to truncate.
|
||||
* Read a stored prefix by id, scanning every backend storage scope. Returns
|
||||
* `undefined` if no stored artifact exists. Returned metadata must identify
|
||||
* `id` before repair or state publication. Used by resume/load, live adoption,
|
||||
* and — via `!== undefined` — the create-collision probe. The returned
|
||||
* `tornMarker` is present iff there is a torn tail to truncate.
|
||||
*/
|
||||
loadStored(id: SessionId): Promise<StoredPrefix<TornMarker> | undefined>
|
||||
|
||||
/**
|
||||
* Read a stored prefix SCOPED to `cwd`. Deliberately distinct from
|
||||
* {@link loadStored}: HMR live-adoption must only adopt a persisted log at the
|
||||
* SAME cwd as the live session (a same-id log at a different cwd is a
|
||||
* collision, not a resume) — conflating the two reintroduces a cross-cwd
|
||||
* adoption bug. For a globally-unique-id backend (SQLite) `cwd` is ignored.
|
||||
*/
|
||||
loadLive(id: SessionId, cwd: string | undefined): Promise<StoredPrefix<TornMarker> | undefined>
|
||||
|
||||
/**
|
||||
* Durably append a CONTIGUOUS batch, lazily materializing the session first
|
||||
* when `!isMaterialized`. The materialize-write and the first event batch MUST
|
||||
@@ -259,6 +251,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
const stored = await this.backend.loadStored(id)
|
||||
if (stored === undefined) throw new Error(`session "${id}" not found`)
|
||||
const { meta, events, tornMarker } = stored
|
||||
this.assertStoredId(id, meta)
|
||||
this.assertVersion(meta)
|
||||
assertSupportedEvents(events, id)
|
||||
|
||||
@@ -317,6 +310,13 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
}
|
||||
}
|
||||
|
||||
/** Reject backend metadata that is not bound to the requested session id. */
|
||||
private assertStoredId(id: SessionId, meta: SessionHeader): void {
|
||||
if (meta.id !== id) {
|
||||
throw new Error(`stored session identity mismatch: requested "${id}", header contains "${meta.id}"`)
|
||||
}
|
||||
}
|
||||
|
||||
// --- write path (session/event → flush drain) ---
|
||||
|
||||
private installWritePath(): void {
|
||||
@@ -439,6 +439,7 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
const stored = await this.backend.loadStored(id)
|
||||
/* v8 ignore next -- a cursor > 0 means the session was materialized, so it exists */
|
||||
if (stored === undefined) return false
|
||||
this.assertStoredId(id, stored.meta)
|
||||
return seedCoversPrefix(seed, stored.events.slice(0, cursor))
|
||||
}
|
||||
|
||||
@@ -448,9 +449,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
* Cases, by whether this backend tracks the id and whether an artifact exists:
|
||||
* 1. Already tracked → no-op (or claim ownerless state if the seed matches,
|
||||
* or reclaim a truly-abandoned id, else reject as a collision).
|
||||
* 2. Not tracked, an artifact EXISTS at this cwd and is a seq-aligned PREFIX
|
||||
* of the live events → ADOPT it (HMR/reload), persisting any live suffix.
|
||||
* 3. Not tracked, an artifact EXISTS but is NOT a prefix → REJECT (collision).
|
||||
* 2. Not tracked, an artifact EXISTS at the same cwd and is a seq-aligned
|
||||
* PREFIX of the live events → ADOPT it, persisting any live suffix.
|
||||
* 3. Not tracked, an artifact EXISTS at another cwd or is NOT a prefix →
|
||||
* REJECT (collision).
|
||||
* 4. Not tracked and NO artifact → a genuinely new session: register meta
|
||||
* (lazy) and persist its seed once.
|
||||
*/
|
||||
@@ -464,14 +466,11 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
if (tracked.owner === undefined) {
|
||||
// Ownerless state from the public create()/load() API. The FIRST live
|
||||
// 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).
|
||||
// A same-id ownerless artifact at a different cwd is a collision, not a
|
||||
// claim: accepting it would append this live session's events through
|
||||
// the stored header's cwd. The seed guard then ensures the live events
|
||||
// reproduce the persisted prefix; otherwise a fresh session reusing the
|
||||
// id could have its leading events filtered as already written.
|
||||
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)`)
|
||||
}
|
||||
@@ -495,11 +494,9 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
}
|
||||
}
|
||||
|
||||
// case 2/3: an artifact at THIS cwd is adopted as a live prefix (or rejected
|
||||
// as a collision inside adoptLivePrefix). cwd-scoped (loadLive), never
|
||||
// any-scope: a same-id artifact at a different cwd is a collision, not a
|
||||
// resume.
|
||||
const live = await this.backend.loadLive(id, session.header.cwd)
|
||||
// case 2/3: resolve the id once across storage, then let adoption reject a
|
||||
// cwd mismatch before repair or state publication.
|
||||
const live = await this.backend.loadStored(id)
|
||||
if (live !== undefined) {
|
||||
// Do NOT route through loadCore(): that crash-repairs open turns as
|
||||
// interrupted, which is wrong for HMR while the live Session is still the
|
||||
@@ -528,6 +525,10 @@ export class PersistenceCoordinator<TornMarker = unknown> {
|
||||
*/
|
||||
private async adoptLivePrefix(session: Session, seed: readonly SessionEvent[], stored: StoredPrefix<TornMarker>): Promise<void> {
|
||||
const { meta, events, tornMarker } = stored
|
||||
this.assertStoredId(session.header.id, meta)
|
||||
if (meta.cwd !== session.header.cwd) {
|
||||
throw new Error(`session "${session.header.id}" is already persisted at a different cwd (persisted: ${String(meta.cwd)}, live: ${String(session.header.cwd)}) (id collision)`)
|
||||
}
|
||||
this.assertVersion(meta)
|
||||
assertSupportedEvents(events, session.header.id)
|
||||
if (!seedCoversPrefix(seed, events)) {
|
||||
|
||||
@@ -63,7 +63,7 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
|
||||
super(ctx)
|
||||
// Assign the store BEFORE constructing the coordinator: the coordinator's
|
||||
// constructor installs the write path and synchronously seeds existing live
|
||||
// sessions (onCreated → loadLive → this.store), so store must exist first.
|
||||
// sessions through loadStored(), so store must exist first.
|
||||
this.store = config?.store ?? new Map<string, { meta: SessionHeader; events: SessionEvent[] }>()
|
||||
this.coordinator = new PersistenceCoordinator<never>(this.ctx, this)
|
||||
}
|
||||
@@ -88,18 +88,13 @@ class MemoryPersistence extends SessionPersistence implements PersistenceBackend
|
||||
|
||||
// --- PersistenceBackend hooks (the Map storage primitives) ---
|
||||
|
||||
// A Map-backed store has no torn tails, so `tornMarker` is never set. Ids are
|
||||
// globally unique, so loadStored and loadLive are identical (cwd is ignored).
|
||||
// A Map-backed store has no torn tails, so `tornMarker` is never set.
|
||||
async loadStored(id: SessionId): Promise<StoredPrefix<never> | undefined> {
|
||||
const entry = this.store.get(id)
|
||||
if (!entry) return undefined
|
||||
return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) }
|
||||
}
|
||||
|
||||
loadLive(id: SessionId, _cwd: string | undefined): Promise<StoredPrefix<never> | undefined> {
|
||||
return this.loadStored(id)
|
||||
}
|
||||
|
||||
async appendBatch(m: SessionHeader, events: readonly SessionEvent[], _isMaterialized: boolean): Promise<void> {
|
||||
// Defense-in-depth: the coordinator already validates serializability, but a
|
||||
// durable store must reject non-JSON data at its own boundary too.
|
||||
@@ -137,6 +132,7 @@ class ControlledBackend implements PersistenceBackend<never> {
|
||||
readonly lifecycle: string[] = []
|
||||
appendAttempts = 0
|
||||
loadAttempts = 0
|
||||
repairAttempts = 0
|
||||
beforeAppend?: (attempt: number) => Promise<void>
|
||||
beforeLoadStored?: (attempt: number) => Promise<void>
|
||||
|
||||
@@ -147,10 +143,6 @@ class ControlledBackend implements PersistenceBackend<never> {
|
||||
return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) }
|
||||
}
|
||||
|
||||
loadLive(id: SessionId, _cwd: string | undefined): Promise<StoredPrefix<never> | undefined> {
|
||||
return this.loadStored(id)
|
||||
}
|
||||
|
||||
async appendBatch(m: SessionHeader, events: readonly SessionEvent[], _isMaterialized: boolean): Promise<void> {
|
||||
const attempt = ++this.appendAttempts
|
||||
await this.beforeAppend?.(attempt)
|
||||
@@ -162,7 +154,9 @@ class ControlledBackend implements PersistenceBackend<never> {
|
||||
}
|
||||
}
|
||||
|
||||
async commitRepair(_m: SessionHeader, _tornMarker: undefined, _closers: readonly SessionEvent[]): Promise<void> {}
|
||||
async commitRepair(_m: SessionHeader, _tornMarker: undefined, _closers: readonly SessionEvent[]): Promise<void> {
|
||||
this.repairAttempts += 1
|
||||
}
|
||||
|
||||
async list(): Promise<SessionHeader[]> {
|
||||
return [...this.store.values()].map(entry => structuredClone(entry.meta))
|
||||
@@ -194,6 +188,36 @@ runCoordinatorContract('memory', async (): Promise<CoordinatorFixture> => {
|
||||
}
|
||||
})
|
||||
|
||||
describe('PersistenceCoordinator stored identity', () => {
|
||||
it('rejects a mismatched backend header before repair or state publication', async () => {
|
||||
const ctx = new Context()
|
||||
await ctx.plugin(SessionStore)
|
||||
const backend = new ControlledBackend()
|
||||
const requested = SessionId('requested')
|
||||
backend.store.set(requested, {
|
||||
meta: meta('different'),
|
||||
events: [{
|
||||
type: 'turn/start',
|
||||
seq: 0,
|
||||
time: 1,
|
||||
data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } },
|
||||
}],
|
||||
})
|
||||
let coordinator!: PersistenceCoordinator<never>
|
||||
const fiber = await ctx.plugin(Object.assign((inner: Context) => {
|
||||
coordinator = new PersistenceCoordinator(inner, backend)
|
||||
}, { inject: ['sessions'] }))
|
||||
try {
|
||||
await expect(coordinator.load(requested)).rejects.toThrow(/stored session identity mismatch/)
|
||||
expect(backend.repairAttempts).toBe(0)
|
||||
expect((coordinator as unknown as CoordinatorInternals).states.size).toBe(0)
|
||||
} finally {
|
||||
await fiber.dispose()
|
||||
await ctx.fiber.dispose()
|
||||
}
|
||||
})
|
||||
})
|
||||
|
||||
describe('PersistenceCoordinator retirement', () => {
|
||||
it('a retiring unmaterialized owner without buffered events releases its id', async () => {
|
||||
const ctx = new Context()
|
||||
|
||||
Reference in New Issue
Block a user