Merge commit 'refs/codex-unblock/20260723/pr570-master-initial' into worktree/pr570-merge-master-20260723

# Conflicts:
#	.agents/notes/implemented/architecture/2026-06-18-shared-persistence-write-coordinator.md
#	docs/core-data-structures/persistence.md
#	packages/session-persistence/session-persistence/README.md

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

@@ -8,7 +8,7 @@ A SQLite durable session-persistence backend — a second `SessionPersistence` i
 
 ## Storage model
 
-Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data, source_event_seqs, surface_op)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). The two `TEXT` columns `source_event_seqs` and `surface_op` are nullable; they store the event's optional surface-metadata fields (see [session surface](../../../.agents/notes/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`) lives in a `sessions` row. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`list` reports exactly the sessions that have a row), so no separate column is needed.
+Each `SessionEvent` maps 1:1 onto a row in an `events` table `(session_id, seq, type, time, data, source_event_seqs, surface_op)` — `data` is the event payload as JSON text, so the row shape is the event verbatim (including `assistant/chunk`, keeping `seq` contiguous). The two `TEXT` columns `source_event_seqs` and `surface_op` are nullable; they store the event's optional surface-metadata fields (see [session surface](../../../.agents/notes/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`), a per-materialization incarnation id, and a monotonic per-log revision live in a `sessions` row; a singleton state row carries the immutable store id. A `sessions` row is written only by the first `append` — its existence is the lazy-materialization signal (`list` reports exactly the sessions that have a row).
 
 The repository's Node range supports unflagged `node:sqlite`. The database enables foreign keys and uses the configured journal mode (`wal` by default; use a rollback mode where WAL shared-memory files are unsuitable). `PRAGMA user_version` stores the table-layout version; databases with any other version are rejected because this unreleased format has no migrations.
 
@@ -19,6 +19,8 @@ On filesystems with POSIX modes, the backend requests mode `0700` for missing di
 - **Append = a transaction.** `append` runs `BEGIN`/`COMMIT` around the batch: it materializes the `sessions` row (if still lazy) and INSERTs every event, asserting the contiguous-seq contract first (the first event's `seq` must equal the stored next-seq). A mid-batch failure (a UNIQUE violation on a duplicated seq) rolls back entirely, so the stored log and the in-memory cursor stay consistent. (`load()` already balanced the stored log, so `append` never has to repair a crash tail.)
 - **Lazy materialization.** `create()` records intent in memory only — no row is written until the first `append`. A created-but-never-appended session has no `sessions` row, so it is absent from `list()` (which reports exactly the sessions that have a row).
 - **Interrupted-turn close on load.** `load()` implements the shared [crash-recovery contract](../../../.agents/notes/implemented/architecture/2026-06-14-session-persistence.md): preserve the valid interrupted turn, append its synthetic closing events in one transaction, and remove only a torn tail row. Committed parse errors or sequence gaps make the session unloadable. Because recovery mutates stored rows, the next append starts from a balanced log and accurate cursor.
+- **Non-mutating inspection.** `inspect()` returns the detached valid row prefix without deleting a torn tail row or appending recovery closers, and leaves the lightweight revision unchanged.
+- **Lightweight revisions.** `listSnapshots()` combines the immutable store and database-file identity, a per-materialization incarnation id, and a per-session counter incremented in each mutating transaction. This keeps unchanged observations stable without parsing event rows and distinguishes independent stores and recreated same-id logs.
 
 ## Configuration (schemastery)
 

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

@@ -8,12 +8,15 @@
 
 import { Context } from 'cordis'
 import z from 'schemastery'
+import { randomUUID } from 'node:crypto'
+import { statSync } from 'node:fs'
 import { DatabaseSync } from 'node:sqlite'
 import { mkdir, open } from 'node:fs/promises'
 import { dirname, resolve } from 'node:path'
 import {
-  SessionPersistence, PersistenceCoordinator,
-  type PersistenceBackend, type SessionLocation, type StoredPrefix,
+  SessionPersistence, SessionPersistenceRevision, PersistenceCoordinator,
+  type PersistenceBackend, type SessionLocation, type SessionPersistenceSnapshot,
+  type StoredPrefix,
 } from '@deepseek-ai/dsh-session-persistence'
 import type { SessionEvent, SurfaceEventType, SessionId, SessionHeader } from '@deepseek-ai/dsh-session'
 import {
@@ -93,6 +96,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
   override readonly name = 'session-persistence-sqlite'
 
   private db!: DatabaseSync
+  private storeIdentity!: string
   private ready: Promise<void>
   private coordinator: PersistenceCoordinator<number>
 
@@ -105,13 +109,32 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
   }
 
   private async openDb(path: string, journalMode: JournalMode): Promise<void> {
-    if (path !== ':memory:') {
-      const abs = resolve(path)
-      await mkdir(dirname(abs), { recursive: true, mode: 0o700 })
-      await createDatabaseFile(abs)
-      this.db = openDatabase(abs, journalMode)
-    } else {
-      this.db = openDatabase(path, journalMode)
+    const actual = path === ':memory:' ? path : resolve(path)
+    if (actual !== ':memory:') {
+      await mkdir(dirname(actual), { recursive: true, mode: 0o700 })
+      await createDatabaseFile(actual)
+    }
+    this.db = openDatabase(actual, journalMode)
+    try {
+      const row = this.db.prepare(
+        'SELECT store_id FROM persistence_state WHERE singleton = 1',
+      ).get() as { store_id: string } | undefined
+      /* v8 ignore next -- openDatabase inserts the singleton before returning. */
+      if (row === undefined) {
+        throw new Error(`session database at "${actual}" has no store identity`)
+      }
+      if (row.store_id.length === 0) {
+        throw new Error(`session database at "${actual}" has no valid store identity`)
+      }
+      if (actual !== ':memory:') {
+        const identity = statSync(actual, { bigint: true })
+        this.storeIdentity = `file:${identity.dev}:${identity.ino}:${identity.birthtimeNs}:store:${row.store_id}`
+      } else {
+        this.storeIdentity = `memory:store:${row.store_id}`
+      }
+    } catch (error: unknown) {
+      this.db.close()
+      throw error
     }
   }
 
@@ -134,6 +157,10 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
     return this.coordinator.load(id)
   }
 
+  inspect(id: SessionId): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
+    return this.coordinator.inspect(id)
+  }
+
   // One method serves both public `list` and the backend hook; delegating it to
   // the coordinator would call this hook recursively.
 
@@ -179,6 +206,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
         const [surfaceSeqs, surfaceOp] = surfaceBindings(event)
         insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp)
       }
+      this.db.prepare('UPDATE sessions SET revision = revision + 1 WHERE id = ?').run(meta.id)
       this.db.exec('COMMIT')
     } catch (error) {
       this.db.exec('ROLLBACK')
@@ -207,6 +235,9 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
           insertEvent.run(meta.id, event.seq, event.type, event.time, JSON.stringify(event.data), surfaceSeqs, surfaceOp)
         }
       }
+      if (tornMarker !== undefined || closers.length > 0) {
+        this.db.prepare('UPDATE sessions SET revision = revision + 1 WHERE id = ?').run(meta.id)
+      }
       this.db.exec('COMMIT')
     } catch (error) {
       // The DELETE+INSERT cannot collide (a row at a closer's seq is preserved or
@@ -228,6 +259,18 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
     return rows.map(rowToMeta)
   }
 
+  /** List metadata with a source-qualified monotonic revision per session. */
+  async listSnapshots(): Promise<SessionPersistenceSnapshot[]> {
+    await this.ready
+    const rows = this.db.prepare('SELECT * FROM sessions').all() as unknown as SessionRow[]
+    return rows.map(row => ({
+      header: rowToMeta(row),
+      revision: SessionPersistenceRevision(
+        `${this.storeIdentity}:incarnation:${row.incarnation}:revision:${row.revision}`,
+      ),
+    }))
+  }
+
   /** Close the database handle (awaited by the coordinator's dispose, post-drain). */
   async close(): Promise<void> {
     await this.ready
@@ -248,8 +291,9 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
    */
   private writeRow(meta: SessionHeader): void {
     this.db.prepare(`
-      INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length, delegation_depth)
-      VALUES (?, ?, ?, ?, ?, ?, ?)
+      INSERT INTO sessions
+        (id, version, created_at, cwd, parent_session, seed_length, delegation_depth, incarnation, revision)
+      VALUES (?, ?, ?, ?, ?, ?, ?, ?, 0)
       ON CONFLICT(id) DO UPDATE SET
         version = excluded.version,
         created_at = excluded.created_at,
@@ -265,6 +309,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
       meta.parentSession ?? null,
       meta.seedLength ?? null,
       meta.delegationDepth ?? null,
+      randomUUID(),
     )
   }
 }

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

@@ -1,12 +1,14 @@
 /**
  * Schema + load-time helpers for the SQLite session-persistence backend: the
- * DDL (a `sessions` metadata table and a 1:1 `events` row per `SessionEvent`),
- * the database open/configure step, and the last-`turn/end` cut that gives the
- * SQLite backend the SAME crash-tail-on-load semantics as the JSONL backend.
+ * DDL (a store-identity row, `sessions` metadata, and a 1:1 `events` row per
+ * `SessionEvent`), the database open/configure step, and the last-`turn/end`
+ * cut that gives the SQLite backend the SAME crash-tail-on-load semantics as
+ * the JSONL backend.
  *
  * @module dsh-session-persistence-sqlite/schema
  */
 
+import { randomUUID } from 'node:crypto'
 import { DatabaseSync } from 'node:sqlite'
 import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepseek-ai/dsh-session'
 
@@ -15,7 +17,7 @@ import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepsee
  * layout; orthogonal to a session's own `version` (which versions the EVENT
  * vocabulary, stored per session in the `sessions` row).
  */
-export const SCHEMA_VERSION = 5
+export const SCHEMA_VERSION = 8
 
 /**
  * A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}).
@@ -31,6 +33,10 @@ export interface SessionRow {
   cwd: string | null
   parent_session: string | null
   seed_length: number | null
+  /** Stable identity assigned when this log is materialized. */
+  incarnation: string
+  /** Monotonic log-change token incremented in each mutating transaction. */
+  revision: number
   delegation_depth: number | null
 }
 
@@ -62,23 +68,41 @@ export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
  * rather than being migrated in place.
  * @param path - the SQLite database file to open (created when absent).
  * @param journalMode - validated journal pragma.
- * @returns the open handle with pragmas applied and both tables ensured.
+ * @returns the open handle with pragmas applied and all three tables ensured.
  */
 export function openDatabase(path: string, journalMode: JournalMode): DatabaseSync {
   const db = new DatabaseSync(path)
+  try {
+    configureDatabase(db, path, journalMode)
+    return db
+  } catch (error: unknown) {
+    db.close()
+    throw error
+  }
+}
+
+function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalMode): void {
   db.exec('PRAGMA foreign_keys = ON')
   // The validated union is safe to interpolate into a non-bindable PRAGMA.
   db.exec(`PRAGMA journal_mode = ${journalMode.toUpperCase()}`)
   // `PRAGMA user_version` always returns exactly one row { user_version }.
   const { user_version: onDisk } = db.prepare('PRAGMA user_version').get() as { user_version: number }
   if (onDisk !== 0 && onDisk !== SCHEMA_VERSION) {
-    db.close()
     throw new Error(`session database at "${path}" has schema version ${onDisk}, incompatible with this build (${SCHEMA_VERSION})`)
   }
   if (onDisk === 0) {
     // Stamp fresh or pre-versioning databases.
     db.exec(`PRAGMA user_version = ${SCHEMA_VERSION}`)
   }
+  db.exec(`
+    CREATE TABLE IF NOT EXISTS persistence_state (
+      singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
+      store_id  TEXT NOT NULL
+    ) STRICT
+  `)
+  db.prepare(
+    'INSERT OR IGNORE INTO persistence_state (singleton, store_id) VALUES (1, ?)',
+  ).run(randomUUID())
   db.exec(`
     CREATE TABLE IF NOT EXISTS sessions (
       id             TEXT PRIMARY KEY,
@@ -87,7 +111,9 @@ export function openDatabase(path: string, journalMode: JournalMode): DatabaseSy
       cwd              TEXT,
       parent_session   TEXT,
       seed_length      INTEGER,
-      delegation_depth INTEGER
+      delegation_depth INTEGER,
+      incarnation      TEXT NOT NULL,
+      revision         INTEGER NOT NULL
     ) STRICT
   `)
   db.exec(`
@@ -102,7 +128,6 @@ export function openDatabase(path: string, journalMode: JournalMode): DatabaseSy
       PRIMARY KEY (session_id, seq)
     ) STRICT
   `)
-  return db
 }
 
 /**

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

@@ -1,7 +1,7 @@
 import { afterEach, describe, expect, it } from 'vitest'
 import { Context } from 'cordis'
 import { existsSync } from 'node:fs'
-import { chmod, mkdtemp, rm, stat, writeFile } from 'node:fs/promises'
+import { chmod, mkdtemp, rm, stat, symlink, writeFile } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { dirname, join } from 'node:path'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
@@ -155,8 +155,8 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     const path = await freshDbPath()
     const m = meta('legacy-header-delta', '/legacy')
     const db = openDatabase(path, 'wal')
-    db.prepare('INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length) VALUES (?, ?, ?, ?, NULL, NULL)')
-      .run(m.id, m.version, m.createdAt, m.cwd ?? null)
+    db.prepare('INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length, delegation_depth, incarnation, revision) VALUES (?, ?, ?, ?, NULL, NULL, NULL, ?, 1)')
+      .run(m.id, m.version, m.createdAt, m.cwd ?? null, 'legacy-header-delta')
     const insert = db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
     insert.run(m.id, 0, 'turn/start', 1, JSON.stringify({ turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } }))
     insert.run(m.id, 1, 'request/header-delta', 2, JSON.stringify({ config: { model: 'legacy' } }))
@@ -172,8 +172,8 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     const path = await freshDbPath()
     const m = meta('legacy-header-fallback', '/legacy')
     const db = openDatabase(path, 'wal')
-    db.prepare('INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length) VALUES (?, ?, ?, ?, NULL, NULL)')
-      .run(m.id, m.version, m.createdAt, m.cwd ?? null)
+    db.prepare('INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length, delegation_depth, incarnation, revision) VALUES (?, ?, ?, ?, NULL, NULL, NULL, ?, 1)')
+      .run(m.id, m.version, m.createdAt, m.cwd ?? null, 'legacy-header-fallback')
     db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
       .run(m.id, 0, 'request/header', 1, JSON.stringify({
         header: { config: { model: 'legacy' } },
@@ -294,22 +294,20 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     dbNewer.close()
     expect(() => openDatabase(path, 'wal')).toThrow(/incompatible with this build/)
 
-    // A stale OLDER version (e.g. a pre-summary-drop v1 DB) is also rejected —
-    // we do not migrate (unreleased software, no backward-compat).
+    // The immediately preceding layout lacks the required store identity and is
+    // rejected rather than migrated (unreleased software, no backward-compat).
     const olderPath = await freshDbPath()
     openDatabase(olderPath, 'wal').close()
     const dbOlder = openDatabase(olderPath, 'wal')
-    dbOlder.exec('PRAGMA user_version = 1')
+    dbOlder.exec(`PRAGMA user_version = ${SCHEMA_VERSION - 1}`)
     dbOlder.close()
     expect(() => openDatabase(olderPath, 'wal')).toThrow(/incompatible with this build/)
   })
 
   it('rejects a sibling v3 database (the merge-collided version) rather than opening it against missing columns', async () => {
-    // Two unmerged branches each shipped a DISTINCT layout under user_version 3 (one added only
-    // `seed_length`, the other only the surface columns). The merged v4 cannot interpret that
-    // ambiguous, incomplete layout and must reject it.
+    // Version 3 identified two incompatible sibling layouts, so it is always rejected.
     const path = await freshDbPath()
-    openDatabase(path, 'wal').close() // creates + stamps user_version = SCHEMA_VERSION (4)
+    openDatabase(path, 'wal').close() // creates + stamps user_version = SCHEMA_VERSION
     const db = openDatabase(path, 'wal')
     db.exec('PRAGMA user_version = 3')
     db.close()
@@ -382,12 +380,95 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
     await fiber2.dispose()
   })
 
+  it('source-qualifies revisions across stores while preserving same-file reopen identity', async () => {
+    const pathA = await freshDbPath()
+    const pathB = await freshDbPath()
+    const m = meta('revision-source')
+    const a = await backend(pathA)
+    await a.ctx.sessionPersistence.create(m)
+    await a.ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const revisionA = (await a.ctx.sessionPersistence.listSnapshots())[0]?.revision
+    await a.dispose()
+
+    const probeA = openDatabase(pathA, 'wal')
+    const storeIdA = (probeA.prepare(
+      'SELECT store_id FROM persistence_state WHERE singleton = 1',
+    ).get() as { store_id: string }).store_id
+    probeA.close()
+
+    const aliasA = `${pathA}.alias`
+    await symlink(pathA, aliasA)
+    const reopenedA = await backend(aliasA)
+    expect((await reopenedA.ctx.sessionPersistence.listSnapshots())[0]?.revision).toBe(revisionA)
+    await reopenedA.dispose()
+
+    const b = await backend(pathB)
+    await b.ctx.sessionPersistence.create(m)
+    await b.ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const revisionB = (await b.ctx.sessionPersistence.listSnapshots())[0]?.revision
+    const probeB = openDatabase(pathB, 'wal')
+    const storeIdB = (probeB.prepare(
+      'SELECT store_id FROM persistence_state WHERE singleton = 1',
+    ).get() as { store_id: string }).store_id
+    probeB.close()
+    expect(storeIdB).not.toBe(storeIdA)
+    expect(revisionB).not.toBe(revisionA)
+    expect(String(revisionA)).toMatch(/:revision:1$/)
+    expect(String(revisionB)).toMatch(/:revision:1$/)
+    await b.dispose()
+  })
+
+  it('changes revisions when a deleted session id is materialized again in the same database', async () => {
+    const path = await freshDbPath()
+    const m = meta('recreated-revision')
+    const first = await backend(path)
+    await first.ctx.sessionPersistence.create(m)
+    await first.ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const before = (await first.ctx.sessionPersistence.listSnapshots())[0]?.revision
+    await first.dispose()
+
+    const cleanup = openDatabase(path, 'wal')
+    cleanup.prepare('DELETE FROM sessions WHERE id = ?').run(m.id)
+    cleanup.close()
+
+    const second = await backend(path)
+    await second.ctx.sessionPersistence.create(m)
+    await second.ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const after = (await second.ctx.sessionPersistence.listSnapshots())[0]?.revision
+    expect(after).not.toBe(before)
+    expect(String(before)).toMatch(/:revision:1$/)
+    expect(String(after)).toMatch(/:revision:1$/)
+    await second.dispose()
+  })
+
   it('exposes the schema version constant', () => {
-    expect(SCHEMA_VERSION).toBe(5)
+    expect(SCHEMA_VERSION).toBe(8)
+  })
+
+  it('keeps the revision stable for an empty repair hook', async () => {
+    const b = await backend()
+    const m = meta('empty-repair')
+    await b.ctx.sessionPersistence.create(m)
+    await b.ctx.sessionPersistence.append(m.id, oneTurnLog())
+    const before = await b.ctx.sessionPersistence.listSnapshots()
+    await (b.ctx.sessionPersistence as SessionPersistenceSqlite).commitRepair(m, undefined, [])
+    expect(await b.ctx.sessionPersistence.listSnapshots()).toEqual(before)
+    await b.dispose()
   })
 })
 
 describe('SessionPersistenceSqlite: edge cases', () => {
+  it('rejects and closes a current-schema database with an invalid store identity', async () => {
+    const path = await freshDbPath()
+    const db = openDatabase(path, 'wal')
+    db.exec("UPDATE persistence_state SET store_id = '' WHERE singleton = 1")
+    db.close()
+
+    const b = await backend(path)
+    await expect(b.ctx.sessionPersistence.listSnapshots()).rejects.toThrow(/no valid store identity/)
+    await expect(b.dispose()).resolves.toBeUndefined()
+  })
+
   it('creates a new database and WAL sidecars with owner-only modes without changing its parent mode', async () => {
     if (process.platform === 'win32') return
     const path = await freshDbPath()