From 73e3f658c609a5b43e2acf292315a0260d87b36a Mon Sep 17 00:00:00 2001 From: Tianyi Cui <53024+tianyicui@users.noreply.github.com> Date: Mon, 20 Jul 2026 17:40:10 +0800 Subject: [PATCH] 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. --- ...18-shared-persistence-write-coordinator.md | 9 +-- ...026-07-20-jsonl-storage-identity.i18n.yaml | 6 ++ .../2026-07-20-jsonl-storage-identity.md | 29 +++++++ .../2026-07-20-jsonl-storage-identity.zh.md | 29 +++++++ .../session-persistence-jsonl/README.md | 7 +- .../session-persistence-jsonl/src/index.ts | 71 ++++++++++-------- .../tests/jsonl.spec.ts | 75 ++++++++++++++++--- .../session-persistence-sqlite/src/index.ts | 5 -- .../session-persistence/README.md | 5 +- .../session-persistence/src/coordinator.ts | 59 ++++++++------- .../tests/persistence.spec.ts | 48 +++++++++--- 11 files changed, 245 insertions(+), 98 deletions(-) create mode 100644 .agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.i18n.yaml create mode 100644 .agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.md create mode 100644 .agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.zh.md diff --git a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md index 7c73cf24a4..8f26aea91b 100644 --- a/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md +++ b/.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md @@ -16,11 +16,10 @@ The coordinator retires each live session from its `session/disposed` notificati ### The hook interface (`PersistenceBackend`) -Six methods (five required + an optional lifecycle hook) — the only seam between the coordinator and storage: +Five required members plus an optional lifecycle hook form the only boundary between the coordinator and storage: - `name` — backend label for the dispose-failure `AggregateError`. -- `loadStored(id)` — read a stored prefix by id, scanning ANY storage scope (every JSONL cwd bucket; SQLite's id is globally unique). Used by resume/load and, via `!== undefined`, the create-collision probe. -- `loadLive(id, cwd)` — read a stored prefix SCOPED to `cwd`. **Deliberately distinct from `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. Collapsing the two reintroduces a cross-cwd adoption bug. SQLite ignores `cwd`. +- `loadStored(id)` — read one stored prefix by id across every storage scope (every JSONL cwd bucket; SQLite's id is globally unique). Resume/load, live adoption, and the create-collision probe share this lookup. The coordinator asserts the returned id and rejects a stored/live cwd mismatch before repair or state publication. - `appendBatch(meta, events, isMaterialized)` — durably append a contiguous batch, lazily materializing the session ATOMICALLY when not yet materialized (the materialize-write and the first event batch must commit together — a crash between them must not leave a materialized-but-empty session; this is why there is no separate `materialize` hook). - `commitRepair(meta, tornMarker, closers)` — make a crash repair durable: truncate the torn tail (iff `tornMarker !== undefined`) and append `closers`. **NOT required to be atomic** — JSONL legitimately truncates-then-appends in two fsync'd steps, SQLite does DELETE+INSERT in one transaction. Used by `load` (truncate + synthetic closers) and live-adoption (truncate only, `closers = []`). - `list()` — list all stored metadata. @@ -37,8 +36,8 @@ The shared `runPersistenceContract` (public-API contract) keeps running for ever ## Alternatives considered - **A base class the backends extend** — rejected for composition: a backend exposes only the hooks, cannot reach the coordinator's private orchestration state, and a third-party backend may still implement the abstract service directly without the coordinator at all. -- **A wider hook surface** — each candidate hook folded away: there is no separate `materialize` hook (the materialize-write must commit atomically with the first event batch inside `appendBatch`), no separate create-collision probe (it is `loadStored(id) !== undefined`), and no coordinator pass-through for `list()` (listing needs none of the orchestration). +- **A wider hook surface** — each candidate hook folds away: there is no scope-specific live lookup because `loadStored` plus the coordinator's cwd check preserves the collision boundary, no storage-locator generic because validated JSONL metadata reproduces its path while SQLite is already id-bound, no separate `materialize` hook because the first batch must commit atomically with materialization, no separate create-collision probe because it is `loadStored(id) !== undefined`, and no coordinator pass-through for `list()` because listing needs none of the orchestration. ## Consequences -The coordinator adds one indirection, an opaque torn marker, and detached session-retirement tasks, but centralizes correctness-heavy orchestration previously duplicated by every backend. Session disposal remains an observe-only event, so the session owner does not await persistence retirement; the coordinator contains failures, preserves uncommitted buffers, and makes backend teardown the quiescence boundary. Its hook surface stays narrow: collision checks reuse `loadStored`, materialization stays atomic inside `appendBatch`, and listing bypasses the coordinator. New backends implement storage primitives rather than copy the event-buffer-flush lifecycle. +The coordinator adds one indirection, an opaque torn marker, and detached session-retirement tasks, but centralizes correctness-heavy orchestration previously duplicated by every backend. Session disposal remains an observe-only event, so the session owner does not await persistence retirement; the coordinator contains failures, preserves uncommitted buffers, and makes backend teardown the quiescence boundary. Its hook surface stays narrow: identity, adoption, and collision checks reuse `loadStored`; materialization stays atomic inside `appendBatch`; and listing bypasses the coordinator. New backends implement storage primitives rather than copy the event-buffer-flush lifecycle. diff --git a/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.i18n.yaml b/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.i18n.yaml new file mode 100644 index 0000000000..2feb8bdf82 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.i18n.yaml @@ -0,0 +1,6 @@ +# Bilingual-pair consistency record (docs/i18n/README.md): the git blob hash of each +# side as of the last confirmed-consistent state. Both languages carry equal authority; +# after editing either side, bring the other along and re-record with: +# pnpm run verify-translation-pairing --write +2026-07-20-jsonl-storage-identity.md: c22377834244e5749993952a5fe8018b89130c1c +2026-07-20-jsonl-storage-identity.zh.md: 8b9a291772ba7e079c06e0d9e3ab9f285ac1ad7e diff --git a/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.md b/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.md new file mode 100644 index 0000000000..c223778342 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.md @@ -0,0 +1,29 @@ +# Agent Note: Bind JSONL session identity before mutation + +Status: implemented + +English | [中文](2026-07-20-jsonl-storage-identity.zh.md) + +## Problem + +JSONL lookup selects a physical log from the requested session id across cwd buckets, while the parsed `SessionHeader` supplies the metadata used by later repair and append operations. Without binding those two facts, a log selected for session A can declare session B's id or cwd and redirect a repair or later append to B's path. The bucket scan also needs a defined result when the same encoded id exists in more than one bucket. SQLite does not share this ambiguity because its primary-key query binds metadata and events to the requested id. + +## Decision + +`loadStored(id)` is the coordinator's single stored-prefix lookup. The JSONL backend scans every cwd bucket, requires at most one matching encoded filename, parses that file, then validates both `header.id === id` and `selectedPath === logPath(root, header.cwd, header.id)` before returning metadata. `list()` applies the same path validation and rejects duplicate ids across buckets. + +The coordinator independently asserts the returned id and compares the stored cwd with a live session's cwd before repair, state publication, or suffix persistence. It keeps a detached copy of validated metadata; JSONL append and repair derive their path from that copy. The `PersistenceBackend` interface therefore needs neither a scope-specific live lookup nor a storage-locator type. + +The backend supports one live writer per session; another backend instance or process must not mutate that session until the owner finishes disposal and all writes stop. + +## Alternatives considered + +**Flatten storage by session id.** A flat namespace makes duplicate publication collide on one path, but path validation and duplicate rejection close the identity defect without changing the project-grouped cwd layout or its consumers. + +**Carry an opaque storage locator through the coordinator.** A locator binds JSONL mutations directly to a selected path, but JSONL can reproduce that path from metadata it has already validated. Adding another generic and argument to SQLite, test backends, append, and repair makes every implementation carry a concept only the file backend needs. + +**Coordinate multiple live writers.** A dedicated coordination service, process-global registry, or cross-process lock would define a new deployment topology rather than repair identity validation. The supported topology has one live writer; no-overwrite hard-link publication still arbitrates an initial same-id creation race. + +## Consequences + +Mismatched, misplaced, and duplicate JSONL logs fail before repair or coordinator state mutation. The cwd-bucket format stays unchanged and needs no migration. Lookup remains proportional to the number of buckets, and one-live-writer ownership remains an explicit limitation. Coordinator and JSONL tests pin rejection before repair, unchanged bytes for both affected logs, path validation during listing, duplicate-id rejection, and cwd collision handling. diff --git a/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.zh.md b/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.zh.md new file mode 100644 index 0000000000..8b9a291772 --- /dev/null +++ b/.agents/notes/implemented/bug-fix/2026-07-20-jsonl-storage-identity.zh.md @@ -0,0 +1,29 @@ +# Agent Note: 在变更前绑定 JSONL 会话身份 + +Status: implemented + +[English](2026-07-20-jsonl-storage-identity.md) | 中文 + +## 问题 + +JSONL 查找会根据请求的会话 id 在各个 cwd 分桶目录中选出物理日志，而解析得到的 `SessionHeader` 会提供后续修复和追加操作使用的元数据。如果这两个事实没有绑定，为会话 A 选中的日志就能声明会话 B 的 id 或 cwd，并将修复或后续追加重定向到 B 的路径。当同一个编码后 id 出现在多个分桶目录中时，分桶扫描也必须给出确定的结果。SQLite 不存在这种歧义，因为主键查询会将元数据和事件绑定到请求的 id。 + +## 决策 + +`loadStored(id)` 是协调器唯一的已存前缀查找操作。JSONL 后端扫描所有 cwd 分桶目录，要求匹配编码文件名的日志至多有一个，解析该文件，然后在返回元数据前同时验证 `header.id === id` 和 `selectedPath === logPath(root, header.cwd, header.id)`。`list()` 执行相同的路径验证，并拒绝跨分桶目录重复的 id。 + +协调器会独立断言返回的 id，并在修复、发布状态或持久化后缀之前比较已存 cwd 和活动会话的 cwd。协调器保留一份已验证元数据的独立副本；JSONL 的追加和修复操作根据该副本派生路径。因此，`PersistenceBackend` 接口既不需要限定范围的活动会话查找，也不需要存储定位器类型。 + +后端对每个会话只支持一个活动写入方；在所有者完成资源释放且所有写入停止之前，另一个后端实例或进程不得变更该会话。 + +## 考虑过的替代方案 + +**按会话 id 扁平化存储。** 扁平命名空间会让重复发布在同一路径上冲突，但路径验证和重复项拒绝无需改变按项目分组的 cwd 布局及其消费方，也能消除身份缺陷。 + +**通过协调器传递不透明存储定位器。** 定位器可以将 JSONL 变更直接绑定到选定路径，但 JSONL 可以根据已经验证的元数据重新得到该路径。为 SQLite、测试后端、追加和修复操作增加一个泛型和参数，会让每个实现都承担只有文件后端需要的概念。 + +**协调多个活动写入方。** 专用协调服务、进程级全局注册表或跨进程锁会定义新的部署拓扑，而不是修复身份验证。受支持的拓扑只有一个活动写入方；禁止覆盖的硬链接发布仍会裁决初始的同 id 创建竞争。 + +## 后果 + +JSONL 日志的身份不匹配、位置错误和重复会在修复或协调器状态变更前失败。cwd 分桶格式保持不变，无需迁移。查找开销仍与分桶目录数量成正比，单一活动写入方的所有权仍是明确限制。协调器和 JSONL 测试固定了修复前拒绝、两个受影响日志的字节均保持不变、列出时的路径验证、重复 id 拒绝以及 cwd 冲突处理。 diff --git a/packages/session-persistence/session-persistence-jsonl/README.md b/packages/session-persistence/session-persistence-jsonl/README.md index cf733b85d4..a97fbfd206 100644 --- a/packages/session-persistence/session-persistence-jsonl/README.md +++ b/packages/session-persistence/session-persistence-jsonl/README.md @@ -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. diff --git a/packages/session-persistence/session-persistence-jsonl/src/index.ts b/packages/session-persistence/session-persistence-jsonl/src/index.ts index 4e52cb0b9e..84db4ac541 100644 --- a/packages/session-persistence/session-persistence-jsonl/src/index.ts +++ b/packages/session-persistence/session-persistence-jsonl/src/index.ts @@ -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 | 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 | 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> { + private async readPrefix(path: string, expectedId: SessionId): Promise> { 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 { const metas: SessionHeader[] = [] + const ids = new Set() 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 { 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 { 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 [] diff --git a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts index e7dc469132..ab62bce938 100644 --- a/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts +++ b/packages/session-persistence/session-persistence-jsonl/tests/jsonl.spec.ts @@ -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) => void): Promise { + const lines = (await readFile(path, 'utf8')).split('\n') + const header = JSON.parse(lines[0] as string) as Record + update(header) + lines[0] = JSON.stringify(header) + await writeFile(path, lines.join('\n')) +} + async function expectParallelFlushError(promise: Promise, message: RegExp): Promise { 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() }) diff --git a/packages/session-persistence/session-persistence-sqlite/src/index.ts b/packages/session-persistence/session-persistence-sqlite/src/index.ts index 4661b41309..c7d7770642 100644 --- a/packages/session-persistence/session-persistence-sqlite/src/index.ts +++ b/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 | 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 diff --git a/packages/session-persistence/session-persistence/README.md b/packages/session-persistence/session-persistence/README.md index ca04cd1ab8..7ed713a3d6 100644 --- a/packages/session-persistence/session-persistence/README.md +++ b/packages/session-persistence/session-persistence/README.md @@ -34,14 +34,13 @@ The `PersistenceBackend` 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 diff --git a/packages/session-persistence/session-persistence/src/coordinator.ts b/packages/session-persistence/session-persistence/src/coordinator.ts index beb5fca483..11e003147c 100644 --- a/packages/session-persistence/session-persistence/src/coordinator.ts +++ b/packages/session-persistence/session-persistence/src/coordinator.ts @@ -35,22 +35,14 @@ export interface PersistenceBackend { 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 | 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 | 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 { 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 { } } + /** 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 { 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 { * 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 { 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 { } } - // 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 { */ private async adoptLivePrefix(session: Session, seed: readonly SessionEvent[], stored: StoredPrefix): Promise { 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)) { diff --git a/packages/session-persistence/session-persistence/tests/persistence.spec.ts b/packages/session-persistence/session-persistence/tests/persistence.spec.ts index e083ac3543..21d878a6a7 100644 --- a/packages/session-persistence/session-persistence/tests/persistence.spec.ts +++ b/packages/session-persistence/session-persistence/tests/persistence.spec.ts @@ -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() this.coordinator = new PersistenceCoordinator(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 | 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 | undefined> { - return this.loadStored(id) - } - async appendBatch(m: SessionHeader, events: readonly SessionEvent[], _isMaterialized: boolean): Promise { // 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 { readonly lifecycle: string[] = [] appendAttempts = 0 loadAttempts = 0 + repairAttempts = 0 beforeAppend?: (attempt: number) => Promise beforeLoadStored?: (attempt: number) => Promise @@ -147,10 +143,6 @@ class ControlledBackend implements PersistenceBackend { return { meta: structuredClone(entry.meta), events: structuredClone(entry.events) } } - loadLive(id: SessionId, _cwd: string | undefined): Promise | undefined> { - return this.loadStored(id) - } - async appendBatch(m: SessionHeader, events: readonly SessionEvent[], _isMaterialized: boolean): Promise { const attempt = ++this.appendAttempts await this.beforeAppend?.(attempt) @@ -162,7 +154,9 @@ class ControlledBackend implements PersistenceBackend { } } - async commitRepair(_m: SessionHeader, _tornMarker: undefined, _closers: readonly SessionEvent[]): Promise {} + async commitRepair(_m: SessionHeader, _tornMarker: undefined, _closers: readonly SessionEvent[]): Promise { + this.repairAttempts += 1 + } async list(): Promise { return [...this.store.values()].map(entry => structuredClone(entry.meta)) @@ -194,6 +188,36 @@ runCoordinatorContract('memory', async (): Promise => { } }) +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 + 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()