fix(session-query): protect live reconciliation

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

@@ -12,7 +12,7 @@ All three surfaces (`current`, `shadowed`, and `log-only`) are searchable by def
 
 ## Source and index lifecycle
 
-The service requires `ctx.sessions` and observes optional `ctx.sessionPersistence` dynamically. One serialized state machine compares source-qualified lightweight durable snapshot revisions, loads only new or changed logs, extracts shared semantic documents, reconciles changes transactionally, and runs the query. Repeated queries and unchanged same-store reopen load no full durable logs; switching stores, or observing new, changed, deleted, or load-repaired sources, reconciles on the next stable observation. Source or transaction failure commits nothing, and the next search retries.
+The service requires `ctx.sessions` and observes optional `ctx.sessionPersistence` dynamically. One serialized state machine compares source-qualified lightweight durable snapshot revisions, loads only new or changed logs, extracts shared semantic documents, reconciles changes transactionally, and runs the query. It never invokes the persistence backend's mutating `load()` for an id currently owned by `ctx.sessions`; the TEMP live row still records persisted availability, and the durable base refreshes after that live owner detaches. Repeated queries and unchanged same-store reopen load no full durable logs; switching stores, or observing new, changed, deleted, or load-repaired sources, reconciles on the next stable observation. Source or transaction failure commits nothing, and the next search retries.
 
 Persisted FTS rows live in a dedicated derived database. Connection-local TEMP tables hold live rows, which shadow the durable base for the same session and reveal it when the live owner disappears. Unmounting persistence hides durable rows without discarding the cache; remounting reconciles it. Closing or reopening the database drops every live overlay while retaining persisted rows.
 

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

@@ -129,6 +129,7 @@ interface IndexedPersistedRow {
 interface IndexedLiveRow {
   id: string
   fingerprint: string
+  persisted: number
   generation: number
 }
 
@@ -338,7 +339,7 @@ export class SessionQuerySqlite extends SessionQueryService {
       'SELECT id, revision, generation FROM persisted_sessions',
     ).all() as unknown as IndexedPersistedRow[]
     const liveRows = db.prepare(
-      'SELECT id, fingerprint, generation FROM temp.live_sessions',
+      'SELECT id, fingerprint, persisted, generation FROM temp.live_sessions',
     ).all() as unknown as IndexedLiveRow[]
     const persistedById = new Map(persistedRows.map(row => [row.id as SessionId, row]))
     const liveById = new Map(liveRows.map(row => [row.id as SessionId, row]))
@@ -350,7 +351,11 @@ export class SessionQuerySqlite extends SessionQueryService {
     const persistentDeletes = observation.persistenceBinding.service === undefined
       ? []
       : persistedRows.filter(row => !observation.persisted.has(row.id as SessionId))
-    const liveChanges = [...observation.live.values()].filter(entry => liveById.get(entry.header.id)?.fingerprint !== entry.fingerprint)
+    const liveChanges = [...observation.live.values()].filter((entry) => {
+      const indexed = liveById.get(entry.header.id)
+      const persisted = observation.persisted.has(entry.header.id) ? 1 : 0
+      return indexed?.fingerprint !== entry.fingerprint || indexed.persisted !== persisted
+    })
     const liveDeletes = liveRows.filter(row => !observation.live.has(row.id as SessionId))
     const pointerChanged = this._lastPersistenceIdentity !== undefined
       && this._lastPersistenceIdentity !== observation.persistenceBinding.identity
@@ -364,7 +369,11 @@ export class SessionQuerySqlite extends SessionQueryService {
     if (persistentChanges.length > 0 || persistentDeletes.length > 0) nextMainGeneration += 1
     const liveReplacements = liveChanges.map((entry) => {
       nextLocalGeneration = Math.max(nextLocalGeneration, nextMainGeneration) + 1
-      return { entry, generation: nextLocalGeneration }
+      return {
+        entry,
+        generation: nextLocalGeneration,
+        persisted: observation.persisted.has(entry.header.id),
+      }
     })
 
     if (hasWrites) {
@@ -382,8 +391,8 @@ export class SessionQuerySqlite extends SessionQueryService {
           db.prepare('UPDATE search_state SET global_generation = ? WHERE singleton = 1').run(nextMainGeneration)
         }
         for (const row of liveDeletes) this._deleteSession('live', row.id as SessionId)
-        for (const { entry, generation } of liveReplacements) {
-          this._replaceLiveSession(entry, generation)
+        for (const { entry, generation, persisted } of liveReplacements) {
+          this._replaceLiveSession(entry, generation, persisted)
         }
         db.exec('COMMIT')
       } catch (error: unknown) {
@@ -419,6 +428,7 @@ export class SessionQuerySqlite extends SessionQueryService {
       assertNotAborted(signal)
       const persistenceBinding = this._persistenceBinding
       const persistence = persistenceBinding.service
+      const initiallyLive = new Set(this.ctx.sessions.list().map(session => session.id))
       let persisted = new Map<SessionId, ObservedPersistedSession>()
       if (persistence !== undefined) {
         try {
@@ -428,6 +438,10 @@ export class SessionQuerySqlite extends SessionQueryService {
           persisted = materializePersistenceSnapshots(before)
           for (const entry of persisted.values()) {
             if (canReuseIndexed && indexed.get(entry.header.id)?.revision === entry.revision) continue
+            // `load()` may durably repair an interrupted tail. Never invoke it
+            // for a session currently owned by the live store: a checkpointed
+            // open turn is active, not crash-interrupted.
+            if (initiallyLive.has(entry.header.id) || this.ctx.sessions.get(entry.header.id) !== undefined) continue
             const loaded = await waitWithAbort(persistence.load(entry.header.id), signal)
             assertSessionHeadersCompatible(entry.header, loaded.meta)
             entry.loaded = observeSession(loaded.meta, loaded.events)
@@ -459,9 +473,8 @@ export class SessionQuerySqlite extends SessionQueryService {
         if (durable !== undefined) assertSessionHeadersCompatible(observed.header, durable.header)
         live.set(session.id, observed)
       }
-      if (this._persistenceBinding === persistenceBinding) {
-        return { persistenceBinding, persisted, live }
-      }
+      if (!sameSessionIds(initiallyLive, live)) continue
+      return { persistenceBinding, persisted, live }
     }
     throw new SessionQueryError(
       'session-search persistence observation did not stabilize after one retry',
@@ -527,13 +540,13 @@ export class SessionQuerySqlite extends SessionQueryService {
     }
   }
 
-  private _replaceLiveSession(entry: ObservedSession, generation: number): void {
+  private _replaceLiveSession(entry: ObservedSession, generation: number, persisted: boolean): void {
     this._deleteSession('live', entry.header.id)
     const db = this._requireDb()
     db.prepare(`
       INSERT INTO temp.live_sessions
-        (id, version, created_at, cwd, parent_session, seed_length, delegation_depth, fingerprint, generation)
-      VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+        (id, version, created_at, cwd, parent_session, seed_length, delegation_depth, fingerprint, persisted, generation)
+      VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
     `).run(
       entry.header.id,
       entry.header.version,
@@ -543,6 +556,7 @@ export class SessionQuerySqlite extends SessionQueryService {
       entry.header.seedLength ?? null,
       entry.header.delegationDepth ?? null,
       entry.fingerprint,
+      persisted ? 1 : 0,
       generation,
     )
     const insert = db.prepare(`
@@ -709,9 +723,7 @@ function selectedDocumentsSql(): { sql: string } {
         ls.seed_length AS seed_length,
         ls.delegation_depth AS delegation_depth,
         1 AS live,
-        CASE WHEN ? = 1 AND EXISTS (
-          SELECT 1 FROM persisted_sessions AS ps WHERE ps.id = ld.session_id
-        ) THEN 1 ELSE 0 END AS persisted,
+        CASE WHEN ? = 1 THEN ls.persisted ELSE 0 END AS persisted,
         CAST(ld.seq AS INTEGER) AS seq,
         ld.type AS type,
         CAST(ld.time AS INTEGER) AS time,
@@ -799,6 +811,17 @@ function samePersistenceSnapshots(
   return true
 }
 
+function sameSessionIds(
+  before: ReadonlySet<SessionId>,
+  after: ReadonlyMap<SessionId, ObservedSession>,
+): boolean {
+  if (before.size !== after.size) return false
+  for (const id of before) {
+    if (!after.has(id)) return false
+  }
+  return true
+}
+
 function sameHeader(a: SessionHeader, b: SessionHeader): boolean {
   return a.version === b.version
     && a.id === b.id

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

@@ -13,6 +13,17 @@ export const SESSION_QUERY_SQLITE_APPLICATION_ID = 0x44534851
 /** Supported SQLite journal modes. */
 export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
 
+const DERIVED_USER_TABLES = new Set([
+  'search_state',
+  'persisted_sessions',
+  'persisted_docs',
+  'persisted_docs_data',
+  'persisted_docs_idx',
+  'persisted_docs_content',
+  'persisted_docs_docsize',
+  'persisted_docs_config',
+])
+
 /**
  * Exclusively create a missing database file with owner-only permissions.
  * Existing files retain their modes, and errors other than `EEXIST` propagate.
@@ -50,7 +61,7 @@ export async function openSearchDatabase(path: string, journalMode: JournalMode)
       throw new Error(`session-search database at "${actual}" is not an empty or recognized derived index`)
     }
     if (applicationId === SESSION_QUERY_SQLITE_APPLICATION_ID && version !== SESSION_QUERY_SQLITE_SCHEMA_VERSION) {
-      resetDerivedSchema(db)
+      resetDerivedSchema(db, actual, userTables)
     }
     // Apply mutating pragmas only after refusing foreign or canonical files.
     // journalMode is a validated closed union, not caller-controlled SQL.
@@ -71,8 +82,14 @@ function listUserTables(db: DatabaseSync): string[] {
   return rows.map(row => row.name)
 }
 
-function resetDerivedSchema(db: DatabaseSync): void {
-  for (const name of listUserTables(db)) {
+function resetDerivedSchema(db: DatabaseSync, path: string, userTables: readonly string[]): void {
+  const unknownTables = userTables.filter(name => !DERIVED_USER_TABLES.has(name))
+  if (unknownTables.length > 0) {
+    throw new Error(
+      `session-search database at "${path}" has unrecognized user tables: ${unknownTables.join(', ')}`,
+    )
+  }
+  for (const name of userTables) {
     db.exec(`DROP TABLE IF EXISTS ${quoteIdentifier(name)}`)
   }
   db.exec('PRAGMA user_version = 0')
@@ -126,6 +143,7 @@ function ensureTemporarySchema(db: DatabaseSync): void {
       seed_length    INTEGER,
       delegation_depth INTEGER,
       fingerprint    TEXT NOT NULL,
+      persisted      INTEGER NOT NULL CHECK (persisted IN (0, 1)),
       generation     INTEGER NOT NULL
     ) STRICT
   `)

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

@@ -10,7 +10,6 @@ import SessionPersistence, { SessionPersistenceRevision } from '@deepseek-ai/dsh
 import type { SessionPersistenceSnapshot } from '@deepseek-ai/dsh-session-persistence'
 import SessionPersistenceSqlite from '@deepseek-ai/dsh-session-persistence-sqlite'
 import SessionQuerySqlite, {
-  SESSION_QUERY_SQLITE_APPLICATION_ID,
   SESSION_QUERY_SQLITE_SCHEMA_VERSION,
 } from '@deepseek-ai/dsh-session-query-sqlite'
 import {
@@ -584,6 +583,63 @@ describe('SQLite reconciliation and source lifecycle', () => {
       .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
   })
 
+  it('does not load a persisted log while the same session is live', async () => {
+    const shared = header('checkpointed-live', 10)
+    TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
+    const ctx = await liveContext()
+    const live = ctx.sessions.prepare(shared.id, {
+      seed: messageEvents('live needle'),
+      meta: { createdAt: shared.createdAt },
+    })
+    const detach = ctx.sessions.enter(live)
+    ctx.sessions.announce(live)
+    const persistence = await ctx.plugin(TestPersistence)
+
+    await expect(ctx.sessionQuery.searchSessions({
+      query: 'live',
+      sessionFilters: [{ kind: 'availability', values: ['persisted'] }],
+    })).resolves.toMatchObject({
+      items: [{ header: shared, live: true, persisted: true }],
+    })
+    expect(TestPersistence.loads.get(shared.id)).toBeUndefined()
+
+    detach()
+    await expect(ctx.sessionQuery.searchSessions({ query: 'persisted' }))
+      .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
+    expect(TestPersistence.loads.get(shared.id)).toBe(1)
+    await persistence.dispose()
+  })
+
+  it('retries when a live owner attaches during persistence observation', async () => {
+    TestPersistence.reset()
+    const ctx = await liveContext()
+    await ctx.plugin(TestPersistence)
+    TestPersistence.snapshotEffect = () => {
+      TestPersistence.snapshotEffect = undefined
+      ctx.sessions.create(SessionId('attached'), { seed: messageEvents('attached needle') })
+    }
+
+    await expect(ctx.sessionQuery.searchSessions({ query: 'attached' }))
+      .resolves.toMatchObject({ items: [{ header: { id: SessionId('attached') } }] })
+  })
+
+  it('retries when one live owner replaces another during persistence observation', async () => {
+    TestPersistence.reset()
+    const ctx = await liveContext()
+    const first = ctx.sessions.prepare(SessionId('first'), { seed: messageEvents('first needle') })
+    const detachFirst = ctx.sessions.enter(first)
+    ctx.sessions.announce(first)
+    await ctx.plugin(TestPersistence)
+    TestPersistence.snapshotEffect = () => {
+      TestPersistence.snapshotEffect = undefined
+      detachFirst()
+      ctx.sessions.create(SessionId('second'), { seed: messageEvents('second needle') })
+    }
+
+    await expect(ctx.sessionQuery.searchSessions({ query: 'second' }))
+      .resolves.toMatchObject({ items: [{ header: { id: SessionId('second') } }] })
+  })
+
   it('uses the reconciled persistence binding through the query boundary', async () => {
     const durable = header('post-reconcile-unmount')
     TestPersistence.reset([{ meta: durable, events: [
@@ -976,12 +1032,12 @@ describe('SQLite schema, cancellation, and real persistence integration', () =>
     expect(ctx.sessionQuery).toBeUndefined()
   })
 
-  it('resets a recognized incompatible derived schema but refuses a foreign database', async () => {
+  it('resets a recognized incompatible schema but refuses unknown or foreign tables', async () => {
     const stalePath = await temporaryPath('stale.db')
+    const staleOwner = await liveContext({ path: stalePath })
+    await (staleOwner.sessionQuery as SessionQuerySqlite).close()
     const stale = new DatabaseSync(stalePath)
-    stale.exec(`PRAGMA application_id = ${SESSION_QUERY_SQLITE_APPLICATION_ID}`)
     stale.exec('PRAGMA user_version = 999')
-    stale.exec('CREATE TABLE stale(value TEXT)')
     stale.close()
     const staleCtx = await liveContext({ path: stalePath })
     staleCtx.sessions.create(SessionId('live'), { seed: messageEvents('needle') })
@@ -990,9 +1046,26 @@ describe('SQLite schema, cancellation, and real persistence integration', () =>
     const rebuilt = new DatabaseSync(stalePath)
     expect((rebuilt.prepare('PRAGMA user_version').get() as { user_version: number }).user_version)
       .toBe(SESSION_QUERY_SQLITE_SCHEMA_VERSION)
-    expect(rebuilt.prepare("SELECT name FROM sqlite_master WHERE name = 'stale'").get()).toBeUndefined()
     rebuilt.close()
 
+    const augmentedPath = await temporaryPath('augmented.db')
+    const augmentedOwner = await liveContext({ path: augmentedPath })
+    await (augmentedOwner.sessionQuery as SessionQuerySqlite).close()
+    const augmented = new DatabaseSync(augmentedPath)
+    augmented.exec('CREATE TABLE unrelated(value TEXT)')
+    augmented.exec("INSERT INTO unrelated VALUES ('safe')")
+    augmented.exec('PRAGMA user_version = 999')
+    augmented.close()
+    const augmentedCtx = new Context()
+    await augmentedCtx.plugin(SessionStore)
+    await expect(augmentedCtx.plugin(SessionQuerySqlite, { path: augmentedPath }))
+      .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
+    expect(augmentedCtx.sessionQuery).toBeUndefined()
+    const stillAugmented = new DatabaseSync(augmentedPath)
+    expect(stillAugmented.prepare('SELECT value FROM unrelated').get()).toEqual({ value: 'safe' })
+    expect(stillAugmented.prepare('PRAGMA user_version').get()).toEqual({ user_version: 999 })
+    stillAugmented.close()
+
     const foreignPath = await temporaryPath('foreign.db')
     const foreign = new DatabaseSync(foreignPath)
     foreign.exec('PRAGMA journal_mode = WAL')