fix(session-query): qualify persistence revisions by store

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

@@ -6,7 +6,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](../../../docs/rfc/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`) and a monotonic per-log revision live 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).
+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](../../../docs/rfc/implemented/architecture/2026-06-18-session-surface.md)). Out-of-log metadata (`SessionHeader`) 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 repo's `engines.node` is `^22.19.0 || >=24.0.0` (Node 22.19+ or 24+), matching the LTS floor required by the installed Pi adapter dependency; `node:sqlite` itself ships without the `--experimental-sqlite` flag from Node 22.13 (LTS) and 23.4 / 24 (Current) on. The range deliberately excludes Node 23 because that line is non-LTS/EOL and still has flagged runtime features before 23.6. The database opens with `foreign_keys = ON` (so `ON DELETE CASCADE` drops a session's events with its row) and the configured `journal_mode` (default `wal`; pick a rollback-journal mode like `delete` on filesystems where WAL's shared-memory files do not work, e.g. network mounts). The table-layout version is stored in `PRAGMA user_version` and checked on open: a fresh database is stamped with the current `SCHEMA_VERSION`; a database written by any other, incompatible build (a non-current `user_version`, older or newer) is rejected rather than opened against an unknown layout — there is no migration (unreleased software).
 
@@ -14,7 +14,7 @@ The repo's `engines.node` is `^22.19.0 || >=24.0.0` (Node 22.19+ or 24+), matchi
 
 - **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).
-- **Lightweight revisions.** `listSnapshots()` reads the monotonic revision stored beside each session header. Append and mutating load repair increment it in the same transaction as their event changes, so unchanged observations are stable and no full-log count or parse is required.
+- **Lightweight revisions.** `listSnapshots()` combines the database's immutable random store id and physical file identity with the monotonic revision stored beside each session header; an in-memory database uses the store id alone. Append and mutating load repair increment the local counter in the same transaction as their event changes, so unchanged same-file observations are stable, independent stores and file replacements cannot collide on a local counter, and no full-log count or parse is required.
 - **Interrupted-turn close on load.** `load()` reads every stored event ordered by `seq` and finds the longest seq-contiguous, parseable prefix — INCLUDING the real events of an interrupted final turn after the last `turn/end` (the loop only flushes at `turn/end`, so a process killed mid-turn leaves real, fully-written rows past it). A single turn can be huge in a long-horizon task, so those events are **preserved, never truncated**: `load()` CLOSES the orphaned turn by durably appending the minimal synthetic boundary events (an error `tool/result` for every assistant tool call left unanswered, a `step/end` if a step was open, then a `turn/end` carrying `{ kind: 'interrupted' }`), inside one transaction that also DELETEs any never-fully-written torn tail row. `load()` is therefore mutating — after it the stored rows are balanced and the cursor is truthful, so the next `append` continues cleanly. The boundary (last `turn/end`, torn-tail detection) is computed from the `seq`/`type` columns so a malformed `data` in a torn tail row is never parsed (discarded, not unloadable). A parse error or `seq` gap inside the committed region (at or before the last real `turn/end`) makes the session unloadable. A session whose only turn never closed keeps its metadata row and stays present in `list()` — the same as the JSONL backend, whose file likewise survives a first append that never reached `turn/end`.
 
 ## Configuration (schemastery)

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

@@ -19,6 +19,7 @@
 
 import { Context } from 'cordis'
 import z from 'schemastery'
+import { statSync } from 'node:fs'
 import { DatabaseSync } from 'node:sqlite'
 import { mkdir } from 'node:fs/promises'
 import { dirname, resolve } from 'node:path'
@@ -84,6 +85,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>
 
@@ -98,12 +100,29 @@ 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 })
-      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 })
+    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
     }
   }
 
@@ -234,13 +253,13 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
     return rows.map(rowToMeta)
   }
 
-  /** List metadata with an append-only event-count revision per session. */
+  /** 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(`revision:${row.revision}`),
+      revision: SessionPersistenceRevision(`${this.storeIdentity}:revision:${row.revision}`),
     }))
   }
 

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 = 6
 
 /**
  * A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}).
@@ -70,14 +72,25 @@ export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
  * current one (written by a different, incompatible build — older or newer) is
  * REJECTED rather than opened against a layout this build does not understand.
  * There are no migrations: an incompatible layout is rejected. The current
- * sessions row carries every header field plus its monotonic snapshot revision;
- * the events row carries the complete surface metadata.
+ * persistence-state row carries an immutable random store id, the sessions row
+ * carries every header field plus its monotonic snapshot revision, and the
+ * events row carries the complete surface metadata.
  * @param path - the SQLite database file to open (created when absent).
  * @param journalMode - the journal pragma to apply — a closed in-code union, validated by the plugin Config.
- * @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')
   // journalMode is a closed in-code union (validated by the plugin Config), not
   // user-controlled SQL — safe to interpolate (PRAGMA takes no bound params).
@@ -85,7 +98,6 @@ export function openDatabase(path: string, journalMode: JournalMode): DatabaseSy
   // `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) {
@@ -94,6 +106,15 @@ export function openDatabase(path: string, journalMode: JournalMode): DatabaseSy
     // constant (SCHEMA_VERSION is a trusted in-code number, not user input).
     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,
@@ -117,7 +138,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 { mkdtemp, rm } from 'node:fs/promises'
+import { mkdtemp, rm, symlink } from 'node:fs/promises'
 import { tmpdir } from 'node:os'
 import { join } from 'node:path'
 import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
@@ -245,12 +245,12 @@ 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/)
   })
@@ -337,8 +337,46 @@ 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('exposes the schema version constant', () => {
-    expect(SCHEMA_VERSION).toBe(5)
+    expect(SCHEMA_VERSION).toBe(6)
   })
 
   it('keeps the revision stable for an empty repair hook', async () => {
@@ -354,6 +392,17 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
 })
 
 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('append rolls back and rethrows when an event INSERT fails inside the transaction', async () => {
     const path = await freshDbPath()
     const m = meta('rollback-insert')