Merge branch 'worktree/web-carrier-chain' into worktree/web-ask-user-question

packages/session-persistence/session-persistence-jsonl/README.md

@@ -19,7 +19,7 @@ The JSONL durable session-persistence backend — a concrete `SessionPersistence
 
 | Key | Type | Notes |
 |---|---|---|
-| `root` | `string` (required) | Root directory for all session files. **No default** — a `process.cwd()` default would scatter files as the process's cwd changes (bash calls, subprocesses). |
+| `root` | `string` (required) | Root directory for all session files. **No default** — a `process.cwd()` default would scatter files as the process's cwd changes (bash calls, subprocesses). An existing root must be a readable directory; an absent root is created on first materialization. |
 | `packChunks` | `boolean` (default `false`) | Write delta-chunk runs as packed rows (~60% smaller logical logs measured on a real coding session). Off, the written logical layout is byte-identical to the pre-packing format; reading packed rows works regardless of this switch. Off by default while the snapshot goldens stay one-event-per-line — recording with packing on rewrites every fixture `session.jsonl`. |
 | `compression` | `'zstd' \| 'none'` | Defaults to `'zstd'`; `'none'` retains newline-delimited UTF-8 text. |
 
@@ -33,6 +33,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the
 
 ## 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 the encoded header and first batch in a temporary file. POSIX publishes it without overwrite via a hard link and `fsync`s the parent directory. Windows publishes it without overwrite via `MoveFileExW(..., MOVEFILE_WRITE_THROUGH)` and creates missing directories through the same write-through pattern. A created-but-never-appended session leaves nothing on disk and is absent from `list`.
 - **Append-only.** Flushed events are never rewritten. Subsequent raw batches append lines; compressed batches append one frame. Both paths `fsync`, and a caught write or sync failure rolls the file back to its prior byte length.
 - **Crash recovery — preserve valid tail work.** `load` validates every complete compressed frame and scans their decompressed JSONL. If the last frame is structurally incomplete, the reader keeps its complete decoded records, truncates from that frame's start, and re-encodes those records with the synthetic tool, step, and turn closers required by the shared [persistence contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md). Raw mode truncates from its first incomplete line. A checksum/decompression failure in a complete frame, or a defect at or before the last committed `turn/end`, is corruption and rejects.
@@ -40,7 +41,7 @@ A root belongs to one encoding. Startup discovery and targeted lookup reject the
 
 ## 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
 
@@ -63,5 +64,5 @@ JSONL storage does not mutate live request prefixes. A resumed loop can reuse pr
 - **Only the configured encoding and current `SESSION_FORMAT_VERSION` (v0) load** — changing compression requires a separate/fresh root or selecting the legacy raw mode; the pre-release format has no migration.
 - **Compressed files are not directly line-readable** — use the backend to load them, or select `compression: 'none'` before writing a fresh root when text fixtures or external line readers are required.
 - **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 POSIX no-overwrite hard link or Windows write-through rename without replacement.
 - **POSIX materialization requires hard-link support** — first append uses `link()` so same-id races fail instead of overwriting a committed log; Windows uses write-through rename without replacement.

packages/session-persistence/session-persistence-jsonl/src/index.ts

@@ -8,8 +8,9 @@
 
 import { Context } from 'cordis'
 import z from 'schemastery'
+import { readdirSync } from 'node:fs'
 import { open, mkdir, readFile, readdir, link, rm, stat as fsStat, 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,
@@ -38,7 +39,9 @@ export interface Config {
   /**
    * Root directory for all session files. Required (no default): a default of
    * `process.cwd()` would scatter session files as the process's cwd changes
-   * (bash calls, subprocesses). Sessions group under per-cwd subdirectories.
+   * (bash calls, subprocesses). Sessions group under per-cwd subdirectories. An
+   * existing root must be a readable directory; an absent root is created on
+   * first materialization.
    */
   root: string
   /**
@@ -101,6 +104,7 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     // the cast records that runtime fact for exactOptionalPropertyTypes.
     this.packChunks = (config as Required<Config>).packChunks
     this.compression = config.compression ?? DEFAULT_COMPRESSION
+    this.assertUsableRoot()
     this.coordinator = new PersistenceCoordinator<JsonlTornMarker>(this.ctx, this)
   }
 
@@ -135,40 +139,32 @@ 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<JsonlTornMarker> | undefined> {
     await this.ensureRootEncoding()
-    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<JsonlTornMarker> | undefined> {
-    await this.ensureRootEncoding()
-    const path = logPath(this.root, cwd, id, this.compression)
-    if (!await this.exists(path)) {
-      await this.rejectOppositeArtifact(cwd, id)
-      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 opaque marker the
    * coordinator can round-trip without knowing the physical encoding.
    */
-  private async readPrefix(path: string): Promise<StoredPrefix<JsonlTornMarker>> {
+  private async readPrefix(path: string, expectedId?: SessionId): Promise<StoredPrefix<JsonlTornMarker>> {
     const buffer = await readFile(path)
-    if (this.compression === 'zstd') return this.readZstdPrefix(buffer)
-    const { meta, events, committedBytes } = scanLog(buffer)
-    return {
-      meta,
-      events,
-      ...committedBytes < buffer.byteLength
-        ? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
-        : {},
+    let prefix: StoredPrefix<JsonlTornMarker>
+    if (this.compression === 'zstd') {
+      prefix = await this.readZstdPrefix(buffer)
+    } else {
+      const { meta, events, committedBytes } = scanLog(buffer)
+      prefix = {
+        meta,
+        events,
+        ...committedBytes < buffer.byteLength
+          ? { tornMarker: { truncateTo: committedBytes, recoveredEvents: [] } }
+          : {},
+      }
     }
+    this.assertStoredIdentity(path, prefix.meta, expectedId)
+    return prefix
   }
 
   /** Decode complete frames and retain complete JSONL records from a torn final frame. */
@@ -245,19 +241,26 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
     if (repairedEvents.length > 0) await this.appendLines(meta, repairedEvents)
   }
 
-  /** 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[]> {
     await this.ensureRootEncoding()
     const metas: SessionHeader[] = []
+    const ids = new Set<SessionId>()
     for (const dir of await this.listCwdDirs()) {
       for (const name of await this.listArtifacts(dir)) {
+        const path = join(dir, name)
         // Read only headers so listing scales with session count, not log size.
         const first = this.compression === 'zstd'
-          ? await this.readFirstZstdLine(`${dir}/${name}`)
-          : await this.readFirstLine(`${dir}/${name}`)
+          ? await this.readFirstZstdLine(path)
+          : 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)
       }
     }
@@ -506,30 +509,54 @@ 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) + logSuffix(this.compression)
+    const oppositeTarget = encodeSegment(id) + logSuffix(this.oppositeCompression())
+    const matches: string[] = []
     for (const dir of await this.listCwdDirs()) {
-      const path = `${dir}/${target}`
-      const opposite = `${dir}/${encodeSegment(id)}${logSuffix(this.oppositeCompression())}`
+      const path = join(dir, target)
+      const opposite = join(dir, oppositeTarget)
       if (await this.exists(opposite)) throw this.encodingMismatch(opposite)
-      if (await this.exists(path)) {
-        // Recover the cwd from the header so the caller has the session's bucket.
-        const { meta } = await this.readPrefix(path)
-        return { path, cwd: meta.cwd }
-      }
+      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]
+  }
+
+  /** Require an existing configured root to be a readable directory. */
+  private assertUsableRoot(): void {
+    try {
+      readdirSync(this.root)
+    } catch (error) {
+      if (isENOENT(error)) return
+      throw error
+    }
+  }
+
+  /** 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, this.compression)
+    } 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 []

packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts

@@ -20,6 +20,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 expectFlushError(promise: Promise<unknown>, message: RegExp): Promise<void> {
   try {
     await promise
@@ -399,6 +408,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 = rawLogPath(root, a.cwd, a.id)
+    const bPath = rawLogPath(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)
@@ -743,6 +777,38 @@ 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(rawLogPath(root, m.cwd, m.id), (header) => { header.cwd = '/elsewhere' })
+
+    await expect(ctx.sessionPersistence.list()).rejects.toThrow(/and cwd belong at/)
+  })
+
+  it('list rejects a session header whose id cannot name a storage path', async () => {
+    const bucket = sessionDir(root, undefined)
+    await mkdir(bucket, { recursive: true })
+    await writeFile(join(bucket, 'invalid-id.jsonl'), JSON.stringify({
+      type: 'session', version: 0, id: '', createdAt: 1, delegationDepth: 0,
+    }) + '\n')
+
+    await expect(ctx.sessionPersistence.list()).rejects.toThrow(/header id cannot name a storage path/)
+  })
+
+  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(rawLogPath(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) => {
@@ -763,18 +829,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, compression: 'none' })
@@ -782,7 +846,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.
@@ -841,21 +905,31 @@ describe('SessionPersistenceJsonl: edge cases', () => {
     await ctx2.fiber.dispose()
   })
 
-  it('list surfaces a non-ENOENT root error (ENOTDIR) instead of reporting no sessions', async () => {
-    // A durable backend must not collapse a storage fault to "no sessions". Making the root a
-    // regular file forces ENOTDIR from `readdir`, which must propagate.
+  it('plugin load rejects an existing root that is not a directory', async () => {
     const filePath = join(root, 'not-a-dir')
     await writeFile(filePath, 'x')
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
-    await ctx2.plugin(SessionPersistenceJsonl, { root: filePath, compression: 'none' })
-    await expect(ctx2.sessionPersistence.list()).rejects.toThrow(/ENOTDIR/)
+    await expect(ctx2.plugin(SessionPersistenceJsonl, { root: filePath, compression: 'none' })).rejects.toThrow(/ENOTDIR/)
     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('list surfaces a root that becomes unusable after plugin load', async () => {
+    await rm(root, { recursive: true })
+    await writeFile(root, 'not a directory')
+
+    await expect(ctx.sessionPersistence.list()).rejects.toThrow(/ENOTDIR/)
+  })
+
+  it('per-id lookup surfaces non-ENOENT storage errors', async () => {
+    const blocker = join(root, 'not-a-directory')
+    await writeFile(blocker, 'x')
+    const backend = ctx.sessionPersistence as unknown as { exists(path: string): Promise<boolean> }
+
+    await expect(backend.exists(join(blocker, 'child.jsonl'))).rejects.toThrow(/ENOTDIR/)
+  })
+
+  it('materialization surfaces a cwd-bucket storage fault', async () => {
     const cwd = '/x'
     const ctx2 = new Context()
     await ctx2.plugin(SessionStore)
@@ -864,8 +938,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()
   })
 

packages/session-persistence/session-persistence-jsonl/tests/zstd.spec.ts

@@ -460,7 +460,7 @@ describe('SessionPersistenceJsonl: encoding selection', () => {
       '',
     ].join('\n'))
     await expect(ctx.sessionPersistence.load(loadHeader.id)).rejects.toThrow(/uses \.jsonl/)
-    await expect((ctx.sessionPersistence as SessionPersistenceJsonl).loadLive(loadHeader.id, loadHeader.cwd))
+    await expect((ctx.sessionPersistence as SessionPersistenceJsonl).loadStored(loadHeader.id))
       .rejects.toThrow(/uses \.jsonl/)
     await expect(ctx.sessionPersistence.list()).rejects.toThrow(/uses \.jsonl/)
   })

packages/session-persistence/session-persistence-sqlite/src/index.ts

@@ -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

packages/session-persistence/session-persistence/README.md

@@ -36,14 +36,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
 

packages/session-persistence/session-persistence/src/coordinator.ts

@@ -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
@@ -264,6 +256,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)
 
@@ -322,6 +315,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 {
@@ -444,6 +444,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))
   }
 
@@ -453,9 +454,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.
    */
@@ -469,14 +471,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)`)
         }
@@ -500,11 +499,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
@@ -533,6 +530,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)) {

packages/session-persistence/session-persistence/tests/persistence.spec.ts

@@ -73,7 +73,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)
   }
@@ -98,18 +98,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.
@@ -147,6 +142,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>
 
@@ -157,10 +153,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)
@@ -172,7 +164,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))
@@ -204,6 +198,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()