fix(session-query): close review edge cases (round 4)

This commit is contained in:
Hypatia May
2026-07-17 09:39:39 +08:00
parent 92fd92fa69
commit 75e9958f11
12 changed files with 213 additions and 51 deletions

View File

@@ -627,7 +627,7 @@ export interface Config {
export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist' export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
``` ```
Source: [`packages/session-persistence/session-persistence-sqlite/src/index.ts:39`](../packages/session-persistence/session-persistence-sqlite/src/index.ts) Source: [`packages/session-persistence/session-persistence-sqlite/src/index.ts:40`](../packages/session-persistence/session-persistence-sqlite/src/index.ts)
## `@deepseek-ai/dsh-session-query` ## `@deepseek-ai/dsh-session-query`
@@ -654,9 +654,9 @@ export interface Config {
path: string path: string
/** SQLite journal mode. Defaults to `wal`. */ /** SQLite journal mode. Defaults to `wal`. */
journalMode?: JournalMode journalMode?: JournalMode
/** Page size when a request omits `limit`. Defaults to 20. */ /** Page size when a request omits `limit`. At most `Number.MAX_SAFE_INTEGER - 1`; defaults to 20. */
defaultLimit?: number defaultLimit?: number
/** Largest accepted page size. Defaults to 100. */ /** Largest accepted page size. At most `Number.MAX_SAFE_INTEGER - 1`; defaults to 100. */
maxLimit?: number maxLimit?: number
/** Maximum snippet length in Unicode code points. Defaults to 240. */ /** Maximum snippet length in Unicode code points. Defaults to 240. */
snippetChars?: number snippetChars?: number
@@ -666,7 +666,7 @@ export interface Config {
export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist' export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
``` ```
Source: [`packages/session-query/session-query-sqlite/src/index.ts:67`](../packages/session-query/session-query-sqlite/src/index.ts) Source: [`packages/session-query/session-query-sqlite/src/index.ts:72`](../packages/session-query/session-query-sqlite/src/index.ts)
## `@deepseek-ai/dsh-skill` ## `@deepseek-ai/dsh-skill`

View File

@@ -145,17 +145,21 @@ export class SessionPersistenceJsonl extends SessionPersistence implements Persi
async listSnapshots(): Promise<SessionPersistenceSnapshot[]> { async listSnapshots(): Promise<SessionPersistenceSnapshot[]> {
const snapshots: SessionPersistenceSnapshot[] = [] const snapshots: SessionPersistenceSnapshot[] = []
for (const artifact of await this.listArtifacts()) { for (const artifact of await this.listArtifacts()) {
const identity = await stat(artifact.path, { bigint: true }) try {
snapshots.push({ const identity = await stat(artifact.path, { bigint: true })
header: artifact.header, snapshots.push({
revision: SessionPersistenceRevision([ header: artifact.header,
identity.dev, revision: SessionPersistenceRevision([
identity.ino, identity.dev,
identity.size, identity.ino,
identity.mtimeNs, identity.size,
identity.ctimeNs, identity.mtimeNs,
].join(':')), identity.ctimeNs,
}) ].join(':')),
})
} catch (error: unknown) {
if (!isENOENT(error)) throw error
}
} }
return snapshots return snapshots
} }

View File

@@ -179,6 +179,39 @@ describe('SessionPersistenceJsonl: durability and crash semantics', () => {
await otherCtx.fiber.dispose() await otherCtx.fiber.dispose()
}) })
it('omits a snapshot artifact removed after discovery', async () => {
const m = meta('vanishing-snapshot')
await ctx.sessionPersistence.create(m)
await ctx.sessionPersistence.append(m.id, oneTurnLog())
const persistence = ctx.sessionPersistence as unknown as {
listArtifacts(): Promise<Array<{ header: SessionHeader; path: string }>>
}
const listArtifacts = persistence.listArtifacts.bind(persistence)
const discovery = vi.spyOn(persistence, 'listArtifacts').mockImplementation(async () => {
const artifacts = await listArtifacts()
await rm(artifacts[0]!.path)
return artifacts
})
await expect(ctx.sessionPersistence.listSnapshots()).resolves.toEqual([])
discovery.mockRestore()
})
it('surfaces non-ENOENT snapshot stat failures after discovery', async () => {
const blocker = join(root, 'snapshot-not-a-directory')
await writeFile(blocker, 'x')
const persistence = ctx.sessionPersistence as unknown as {
listArtifacts(): Promise<Array<{ header: SessionHeader; path: string }>>
}
const discovery = vi.spyOn(persistence, 'listArtifacts').mockResolvedValue([{
header: meta('snapshot-stat-failure'),
path: join(blocker, 'session.jsonl'),
}])
await expect(ctx.sessionPersistence.listSnapshots()).rejects.toThrow(/ENOTDIR/)
discovery.mockRestore()
})
it('persists a forked child seed through the existing session write path', async () => { it('persists a forked child seed through the existing session write path', async () => {
const source = ctx.sessions.create(SessionId('persist-parent'), { meta: { cwd: '/workspace' } }) const source = ctx.sessions.create(SessionId('persist-parent'), { meta: { cwd: '/workspace' } })
appendClosedTurn(source) appendClosedTurn(source)

View File

@@ -6,7 +6,7 @@ A SQLite durable session-persistence backend — a second `SessionPersistence` i
## Storage model ## 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 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). 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`), 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. 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.
@@ -15,7 +15,7 @@ The repository's Node range supports unflagged `node:sqlite`. The database enabl
- **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.) - **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). - **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](../../../docs/rfc/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. - **Interrupted-turn close on load.** `load()` implements the shared [crash-recovery contract](../../../docs/rfc/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.
- **Lightweight revisions.** `listSnapshots()` combines an immutable store identity, the database file identity, and a per-session counter incremented in each mutating transaction. This keeps unchanged observations stable without parsing event rows and prevents independent stores from sharing a revision accidentally. - **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) ## Configuration (schemastery)

View File

@@ -7,6 +7,7 @@
import { Context } from 'cordis' import { Context } from 'cordis'
import z from 'schemastery' import z from 'schemastery'
import { randomUUID } from 'node:crypto'
import { statSync } from 'node:fs' import { statSync } from 'node:fs'
import { DatabaseSync } from 'node:sqlite' import { DatabaseSync } from 'node:sqlite'
import { mkdir } from 'node:fs/promises' import { mkdir } from 'node:fs/promises'
@@ -235,7 +236,9 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
const rows = this.db.prepare('SELECT * FROM sessions').all() as unknown as SessionRow[] const rows = this.db.prepare('SELECT * FROM sessions').all() as unknown as SessionRow[]
return rows.map(row => ({ return rows.map(row => ({
header: rowToMeta(row), header: rowToMeta(row),
revision: SessionPersistenceRevision(`${this.storeIdentity}:revision:${row.revision}`), revision: SessionPersistenceRevision(
`${this.storeIdentity}:incarnation:${row.incarnation}:revision:${row.revision}`,
),
})) }))
} }
@@ -259,8 +262,9 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
*/ */
private writeRow(meta: SessionHeader): void { private writeRow(meta: SessionHeader): void {
this.db.prepare(` this.db.prepare(`
INSERT INTO sessions (id, version, created_at, cwd, parent_session, seed_length, revision) INSERT INTO sessions
VALUES (?, ?, ?, ?, ?, ?, 0) (id, version, created_at, cwd, parent_session, seed_length, incarnation, revision)
VALUES (?, ?, ?, ?, ?, ?, ?, 0)
ON CONFLICT(id) DO UPDATE SET ON CONFLICT(id) DO UPDATE SET
version = excluded.version, version = excluded.version,
created_at = excluded.created_at, created_at = excluded.created_at,
@@ -274,6 +278,7 @@ export class SessionPersistenceSqlite extends SessionPersistence implements Pers
meta.cwd ?? null, meta.cwd ?? null,
meta.parentSession ?? null, meta.parentSession ?? null,
meta.seedLength ?? null, meta.seedLength ?? null,
randomUUID(),
) )
} }
} }

View File

@@ -17,7 +17,7 @@ import type { SessionEvent, SessionId, SessionHeader, SurfaceOp } from '@deepsee
* layout; orthogonal to a session's own `version` (which versions the EVENT * layout; orthogonal to a session's own `version` (which versions the EVENT
* vocabulary, stored per session in the `sessions` row). * vocabulary, stored per session in the `sessions` row).
*/ */
export const SCHEMA_VERSION = 6 export const SCHEMA_VERSION = 7
/** /**
* A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}). * A row of the `sessions` table — the out-of-log metadata ({@link SessionHeader}).
@@ -33,6 +33,8 @@ export interface SessionRow {
cwd: string | null cwd: string | null
parent_session: string | null parent_session: string | null
seed_length: number | null seed_length: number | null
/** Stable identity assigned when this log is materialized. */
incarnation: string
/** Monotonic log-change token incremented in each mutating transaction. */ /** Monotonic log-change token incremented in each mutating transaction. */
revision: number revision: number
} }
@@ -108,6 +110,7 @@ function configureDatabase(db: DatabaseSync, path: string, journalMode: JournalM
cwd TEXT, cwd TEXT,
parent_session TEXT, parent_session TEXT,
seed_length INTEGER, seed_length INTEGER,
incarnation TEXT NOT NULL,
revision INTEGER NOT NULL revision INTEGER NOT NULL
) STRICT ) STRICT
`) `)

View File

@@ -378,8 +378,31 @@ describe('SessionPersistenceSqlite: durability and crash semantics', () => {
await b.dispose() 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', () => { it('exposes the schema version constant', () => {
expect(SCHEMA_VERSION).toBe(6) expect(SCHEMA_VERSION).toBe(7)
}) })
it('keeps the revision stable for an empty repair hook', async () => { it('keeps the revision stable for an empty repair hook', async () => {

View File

@@ -24,8 +24,8 @@ The database is disposable but reset is guarded: a recognized incompatible searc
|---|---:|---| |---|---:|---|
| `path` | required | Dedicated derived-index SQLite path; `:memory:` is supported. | | `path` | required | Dedicated derived-index SQLite path; `:memory:` is supported. |
| `journalMode` | `wal` | `wal`, `delete`, `truncate`, or `persist`. | | `journalMode` | `wal` | `wal`, `delete`, `truncate`, or `persist`. |
| `defaultLimit` | `20` | Page size when a request omits `limit`. | | `defaultLimit` | `20` | Page size when a request omits `limit`; at most `Number.MAX_SAFE_INTEGER - 1`. |
| `maxLimit` | `100` | Largest accepted request page size. | | `maxLimit` | `100` | Largest accepted request page size; at most `Number.MAX_SAFE_INTEGER - 1`. |
| `snippetChars` | `240` | Maximum snippet length in Unicode code points. | | `snippetChars` | `240` | Maximum snippet length in Unicode code points. |
## Tokenizer and limits ## Tokenizer and limits

View File

@@ -48,6 +48,7 @@ import {
quoteFtsData, quoteFtsData,
requestFingerprint, requestFingerprint,
sanitizeFtsText, sanitizeFtsText,
SQLITE_MAX_PAGE_LIMIT,
} from './query.ts' } from './query.ts'
export { export {
@@ -63,15 +64,19 @@ export const SESSION_QUERY_SQLITE_MAX_LIMIT = 100
/** Default maximum snippet length in Unicode code points. */ /** Default maximum snippet length in Unicode code points. */
export const SESSION_QUERY_SQLITE_SNIPPET_CHARS = 240 export const SESSION_QUERY_SQLITE_SNIPPET_CHARS = 240
// A serialized search tolerates one transient source change; repeated churn
// fails instead of monopolizing the operation queue.
const STABLE_OBSERVATION_ATTEMPTS = 2
/** SQLite session-search configuration. */ /** SQLite session-search configuration. */
export interface Config { export interface Config {
/** Dedicated derived-index path; `:memory:` is supported for tests. */ /** Dedicated derived-index path; `:memory:` is supported for tests. */
path: string path: string
/** SQLite journal mode. Defaults to `wal`. */ /** SQLite journal mode. Defaults to `wal`. */
journalMode?: JournalMode journalMode?: JournalMode
/** Page size when a request omits `limit`. Defaults to 20. */ /** Page size when a request omits `limit`. At most `Number.MAX_SAFE_INTEGER - 1`; defaults to 20. */
defaultLimit?: number defaultLimit?: number
/** Largest accepted page size. Defaults to 100. */ /** Largest accepted page size. At most `Number.MAX_SAFE_INTEGER - 1`; defaults to 100. */
maxLimit?: number maxLimit?: number
/** Maximum snippet length in Unicode code points. Defaults to 240. */ /** Maximum snippet length in Unicode code points. Defaults to 240. */
snippetChars?: number snippetChars?: number
@@ -154,8 +159,8 @@ export class SessionSearchSqlite extends SessionSearchService {
static Config: z<Config> = z.object({ static Config: z<Config> = z.object({
path: z.string().required(), path: z.string().required(),
journalMode: z.union(['wal', 'delete', 'truncate', 'persist'] as const).default('wal'), journalMode: z.union(['wal', 'delete', 'truncate', 'persist'] as const).default('wal'),
defaultLimit: z.number().step(1).min(1).default(SESSION_QUERY_SQLITE_DEFAULT_LIMIT), defaultLimit: z.number().step(1).min(1).max(SQLITE_MAX_PAGE_LIMIT).default(SESSION_QUERY_SQLITE_DEFAULT_LIMIT),
maxLimit: z.number().step(1).min(1).default(SESSION_QUERY_SQLITE_MAX_LIMIT), maxLimit: z.number().step(1).min(1).max(SQLITE_MAX_PAGE_LIMIT).default(SESSION_QUERY_SQLITE_MAX_LIMIT),
snippetChars: z.number().step(1).min(1).default(SESSION_QUERY_SQLITE_SNIPPET_CHARS), snippetChars: z.number().step(1).min(1).default(SESSION_QUERY_SQLITE_SNIPPET_CHARS),
}) })
@@ -206,14 +211,14 @@ export class SessionSearchSqlite extends SessionSearchService {
const signal = exec?.signal const signal = exec?.signal
return this._serialized(signal, async () => { return this._serialized(signal, async () => {
await this._ensureReady(signal) await this._ensureReady(signal)
await this._reconcile(signal) const persistenceBinding = await this._reconcile(signal)
assertNotAborted(signal) assertNotAborted(signal)
const generation = String(this._globalGeneration) const generation = String(this._globalGeneration)
const fingerprint = requestFingerprint(normalized) const fingerprint = requestFingerprint(normalized)
const offset = normalized.cursor === undefined const offset = normalized.cursor === undefined
? 0 ? 0
: decodeCursor(normalized.cursor, this._instance, 'sessions', fingerprint, generation) : decodeCursor(normalized.cursor, this._instance, 'sessions', fingerprint, generation)
const rows = this._querySessions(normalized, offset) const rows = this._querySessions(normalized, offset, persistenceBinding)
return page(rows, normalized.limit, row => this._sessionHit(row), cursorOffset => encodeCursor({ return page(rows, normalized.limit, row => this._sessionHit(row), cursorOffset => encodeCursor({
version: 1, version: 1,
instance: this._instance, instance: this._instance,
@@ -233,14 +238,14 @@ export class SessionSearchSqlite extends SessionSearchService {
const signal = exec?.signal const signal = exec?.signal
return this._serialized(signal, async () => { return this._serialized(signal, async () => {
await this._ensureReady(signal) await this._ensureReady(signal)
await this._reconcile(signal) const persistenceBinding = await this._reconcile(signal)
assertNotAborted(signal) assertNotAborted(signal)
const generation = this._targetGeneration(normalized.sessionId) const generation = this._targetGeneration(normalized.sessionId, persistenceBinding)
const fingerprint = requestFingerprint(normalized) const fingerprint = requestFingerprint(normalized)
const offset = normalized.cursor === undefined const offset = normalized.cursor === undefined
? 0 ? 0
: decodeCursor(normalized.cursor, this._instance, 'events', fingerprint, generation) : decodeCursor(normalized.cursor, this._instance, 'events', fingerprint, generation)
const rows = this._queryEvents(normalized, offset) const rows = this._queryEvents(normalized, offset, persistenceBinding)
return page(rows, normalized.limit, row => this._eventHit(row), cursorOffset => encodeCursor({ return page(rows, normalized.limit, row => this._eventHit(row), cursorOffset => encodeCursor({
version: 1, version: 1,
instance: this._instance, instance: this._instance,
@@ -316,7 +321,7 @@ export class SessionSearchSqlite extends SessionSearchService {
} }
} }
private async _reconcile(signal: AbortSignal | undefined): Promise<void> { private async _reconcile(signal: AbortSignal | undefined): Promise<PersistenceBinding> {
const db = this._requireDb() const db = this._requireDb()
const persistedRows = db.prepare( const persistedRows = db.prepare(
'SELECT id, revision, generation FROM persisted_sessions', 'SELECT id, revision, generation FROM persisted_sessions',
@@ -392,13 +397,14 @@ export class SessionSearchSqlite extends SessionSearchService {
if (pointerChanged) this._persistenceEpoch += 1 if (pointerChanged) this._persistenceEpoch += 1
this._localGeneration = nextLocalGeneration this._localGeneration = nextLocalGeneration
this._lastPersistenceIdentity = observation.persistenceBinding.identity this._lastPersistenceIdentity = observation.persistenceBinding.identity
return observation.persistenceBinding
} }
private async _observeStable( private async _observeStable(
indexed: ReadonlyMap<SessionId, IndexedPersistedRow>, indexed: ReadonlyMap<SessionId, IndexedPersistedRow>,
signal: AbortSignal | undefined, signal: AbortSignal | undefined,
): Promise<Observation> { ): Promise<Observation> {
for (;;) { for (let attempt = 0; attempt < STABLE_OBSERVATION_ATTEMPTS; attempt += 1) {
assertNotAborted(signal) assertNotAborted(signal)
const persistenceBinding = this._persistenceBinding const persistenceBinding = this._persistenceBinding
const persistence = persistenceBinding.service const persistence = persistenceBinding.service
@@ -446,6 +452,10 @@ export class SessionSearchSqlite extends SessionSearchService {
return { persistenceBinding, persisted, live } return { persistenceBinding, persisted, live }
} }
} }
throw new SessionQueryError(
'session-search persistence observation did not stabilize after one retry',
'SESSION_QUERY_PERSISTENCE_FAILED',
)
} }
private _mainGeneration(): number { private _mainGeneration(): number {
@@ -540,7 +550,11 @@ export class SessionSearchSqlite extends SessionSearchService {
} }
} }
private _querySessions(request: NormalizedSessionRequest, offset: number): SearchRow[] { private _querySessions(
request: NormalizedSessionRequest,
offset: number,
persistenceBinding: PersistenceBinding,
): SearchRow[] {
const selected = selectedDocumentsSql() const selected = selectedDocumentsSql()
const sessionWhere = buildSessionWhere(request.sessionFilters) const sessionWhere = buildSessionWhere(request.sessionFilters)
const eventWhere = buildEventWhere(request.eventFilters) const eventWhere = buildEventWhere(request.eventFilters)
@@ -562,7 +576,7 @@ export class SessionSearchSqlite extends SessionSearchService {
ORDER BY match_count DESC, document_length ASC, time DESC, session_id ASC, seq DESC ORDER BY match_count DESC, document_length ASC, time DESC, session_id ASC, seq DESC
LIMIT ? OFFSET ? LIMIT ? OFFSET ?
`).all( `).all(
...selectedDocumentsParams(request.query, this._persistenceBinding.service !== undefined), ...selectedDocumentsParams(request.query, persistenceBinding.service !== undefined),
...sessionWhere.params, ...sessionWhere.params,
...eventWhere.params, ...eventWhere.params,
request.limit + 1, request.limit + 1,
@@ -570,7 +584,11 @@ export class SessionSearchSqlite extends SessionSearchService {
) as unknown as SearchRow[] ) as unknown as SearchRow[]
} }
private _queryEvents(request: NormalizedEventRequest, offset: number): SearchRow[] { private _queryEvents(
request: NormalizedEventRequest,
offset: number,
persistenceBinding: PersistenceBinding,
): SearchRow[] {
const selected = selectedDocumentsSql() const selected = selectedDocumentsSql()
const eventWhere = buildEventWhere(request.filters) const eventWhere = buildEventWhere(request.filters)
const where = ['session_id = ?', eventWhere.sql].filter(Boolean).join(' AND ') const where = ['session_id = ?', eventWhere.sql].filter(Boolean).join(' AND ')
@@ -581,7 +599,7 @@ export class SessionSearchSqlite extends SessionSearchService {
ORDER BY match_count DESC, document_length ASC, time DESC, seq DESC ORDER BY match_count DESC, document_length ASC, time DESC, seq DESC
LIMIT ? OFFSET ? LIMIT ? OFFSET ?
`).all( `).all(
...selectedDocumentsParams(request.query, this._persistenceBinding.service !== undefined), ...selectedDocumentsParams(request.query, persistenceBinding.service !== undefined),
request.sessionId, request.sessionId,
...eventWhere.params, ...eventWhere.params,
request.limit + 1, request.limit + 1,
@@ -589,13 +607,13 @@ export class SessionSearchSqlite extends SessionSearchService {
) as unknown as SearchRow[] ) as unknown as SearchRow[]
} }
private _targetGeneration(sessionId: SessionId): string { private _targetGeneration(sessionId: SessionId, persistenceBinding: PersistenceBinding): string {
const db = this._requireDb() const db = this._requireDb()
const live = db.prepare( const live = db.prepare(
'SELECT generation FROM temp.live_sessions WHERE id = ?', 'SELECT generation FROM temp.live_sessions WHERE id = ?',
).get(sessionId) as { generation: number } | undefined ).get(sessionId) as { generation: number } | undefined
if (live !== undefined) return `live:${live.generation}` if (live !== undefined) return `live:${live.generation}`
if (this._persistenceBinding.service !== undefined) { if (persistenceBinding.service !== undefined) {
const persisted = db.prepare( const persisted = db.prepare(
'SELECT generation FROM persisted_sessions WHERE id = ?', 'SELECT generation FROM persisted_sessions WHERE id = ?',
).get(sessionId) as { generation: number } | undefined ).get(sessionId) as { generation: number } | undefined
@@ -850,8 +868,8 @@ function resolveConfig(config: Config): ResolvedConfig {
if (typeof resolved.path !== 'string' || resolved.path.trim().length === 0) { if (typeof resolved.path !== 'string' || resolved.path.trim().length === 0) {
throw invalidConfig('path must not be blank') throw invalidConfig('path must not be blank')
} }
assertPositiveInteger('defaultLimit', resolved.defaultLimit) assertPageLimit('defaultLimit', resolved.defaultLimit)
assertPositiveInteger('maxLimit', resolved.maxLimit) assertPageLimit('maxLimit', resolved.maxLimit)
assertPositiveInteger('snippetChars', resolved.snippetChars) assertPositiveInteger('snippetChars', resolved.snippetChars)
if (resolved.defaultLimit > resolved.maxLimit) { if (resolved.defaultLimit > resolved.maxLimit) {
throw invalidConfig('defaultLimit must be less than or equal to maxLimit') throw invalidConfig('defaultLimit must be less than or equal to maxLimit')
@@ -865,6 +883,12 @@ function assertPositiveInteger(name: string, value: number): void {
if (!Number.isInteger(value) || value < 1) throw invalidConfig(`${name} must be a positive integer`) if (!Number.isInteger(value) || value < 1) throw invalidConfig(`${name} must be a positive integer`)
} }
function assertPageLimit(name: string, value: number): void {
if (!Number.isSafeInteger(value) || value < 1 || value > SQLITE_MAX_PAGE_LIMIT) {
throw invalidConfig(`${name} must be an integer between 1 and ${SQLITE_MAX_PAGE_LIMIT}`)
}
}
function invalidConfig(detail: string): SessionQueryError { function invalidConfig(detail: string): SessionQueryError {
return new SessionQueryError( return new SessionQueryError(
`session-search SQLite config: ${detail}`, `session-search SQLite config: ${detail}`,

View File

@@ -20,6 +20,9 @@ export const FTS_HIGHLIGHT_START = '\uFDD0'
/** Collision-free marker inserted after an FTS5 match by `highlight()`. */ /** Collision-free marker inserted after an FTS5 match by `highlight()`. */
export const FTS_HIGHLIGHT_END = '\uFDD1' export const FTS_HIGHLIGHT_END = '\uFDD1'
/** Largest page size whose internal lookahead remains an exact SQLite integer binding. */
export const SQLITE_MAX_PAGE_LIMIT = Number.MAX_SAFE_INTEGER - 1
/** Limit defaults needed to normalize a search request. */ /** Limit defaults needed to normalize a search request. */
export interface QueryLimits { export interface QueryLimits {
/** Page size used when the request omits one. */ /** Page size used when the request omits one. */
@@ -232,14 +235,17 @@ export function makeSnippet(markedText: string, maxChars: number): string {
const characters = Array.from(clean) const characters = Array.from(clean)
if (characters.length <= maxChars) return clean if (characters.length <= maxChars) return clean
if (maxChars === 1) return '…' if (maxChars === 1) return '…'
let start = Math.max(0, matchStart - Math.floor(maxChars / 3)) const matchedIndex = Math.min(matchStart, characters.length - 1)
let prefix = start > 0 ? '…' : '' let start = Math.max(0, matchedIndex - Math.floor(maxChars / 3))
const prefix = start > 0 ? '…' : ''
let suffix = '…' let suffix = '…'
let contentLength = maxChars - prefix.length - suffix.length let contentLength = maxChars - prefix.length - suffix.length
if (contentLength < 1) { if (contentLength < 1) {
start = 0 start = matchedIndex
prefix = '' suffix = ''
contentLength = maxChars - 1 contentLength = maxChars - prefix.length - suffix.length
} else if (matchedIndex >= start + contentLength) {
start = matchedIndex - contentLength + 1
} }
let end = Math.min(characters.length, start + contentLength) let end = Math.min(characters.length, start + contentLength)
if (end === characters.length) { if (end === characters.length) {
@@ -326,9 +332,14 @@ function materializeMetadataFilters(
function normalizeLimit(value: number | undefined, limits: QueryLimits): number { function normalizeLimit(value: number | undefined, limits: QueryLimits): number {
const limit = value ?? limits.defaultLimit const limit = value ?? limits.defaultLimit
if (!Number.isInteger(limit) || limit < 1 || limit > limits.maxLimit) { const maxLimit = Math.min(limits.maxLimit, SQLITE_MAX_PAGE_LIMIT)
if (
!Number.isSafeInteger(limit)
|| limit < 1
|| limit > maxLimit
) {
throw new SessionQueryError( throw new SessionQueryError(
`session-search limit must be an integer between 1 and ${limits.maxLimit}`, `session-search limit must be an integer between 1 and ${maxLimit}`,
'SESSION_QUERY_INVALID_LIMIT', 'SESSION_QUERY_INVALID_LIMIT',
) )
} }

View File

@@ -11,6 +11,7 @@ import {
normalizeSessionRequest, normalizeSessionRequest,
quoteFtsData, quoteFtsData,
requestFingerprint, requestFingerprint,
SQLITE_MAX_PAGE_LIMIT,
type NormalizedEventRequest, type NormalizedEventRequest,
type NormalizedSessionRequest, type NormalizedSessionRequest,
} from '../src/query.ts' } from '../src/query.ts'
@@ -88,6 +89,12 @@ describe('SQLite search request normalization', () => {
expect(() => normalizeEventRequest({ sessionId: SessionId('s'), query: 'x', limit }, limits)) expect(() => normalizeEventRequest({ sessionId: SessionId('s'), query: 'x', limit }, limits))
.toThrow(expectCode('SESSION_QUERY_INVALID_LIMIT')) .toThrow(expectCode('SESSION_QUERY_INVALID_LIMIT'))
} }
expect(() => normalizeEventRequest({
sessionId: SessionId('s'),
query: 'x',
limit: SQLITE_MAX_PAGE_LIMIT + 1,
}, { defaultLimit: 1, maxLimit: SQLITE_MAX_PAGE_LIMIT + 1 }))
.toThrow(expectCode('SESSION_QUERY_INVALID_LIMIT'))
}) })
it('materializes owned filter values during normalization', () => { it('materializes owned filter values during normalization', () => {
@@ -214,7 +221,8 @@ describe('SQLite query identity and presentation', () => {
expect(makeSnippet(`abcde${FTS_HIGHLIGHT_START}f${FTS_HIGHLIGHT_END}`, 1)).toBe('…') expect(makeSnippet(`abcde${FTS_HIGHLIGHT_START}f${FTS_HIGHLIGHT_END}`, 1)).toBe('…')
expect(makeSnippet('abcdefghij', 5)).toBe('abcd…') expect(makeSnippet('abcdefghij', 5)).toBe('abcd…')
expect(makeSnippet(`ab${FTS_HIGHLIGHT_START}c${FTS_HIGHLIGHT_END}defghij`, 5)).toBe('…bcd…') expect(makeSnippet(`ab${FTS_HIGHLIGHT_START}c${FTS_HIGHLIGHT_END}defghij`, 5)).toBe('…bcd…')
expect(makeSnippet(`abcde${FTS_HIGHLIGHT_START}f${FTS_HIGHLIGHT_END}`, 2)).toBe('a…') expect(makeSnippet(`ab${FTS_HIGHLIGHT_START}c${FTS_HIGHLIGHT_END}defghij`, 3)).toBe('…c…')
expect(makeSnippet(`abcde${FTS_HIGHLIGHT_START}f${FTS_HIGHLIGHT_END}`, 2)).toBe('…f')
expect(makeSnippet(`abcde${FTS_HIGHLIGHT_START}f${FTS_HIGHLIGHT_END}`, 5)).toBe('…cdef') expect(makeSnippet(`abcde${FTS_HIGHLIGHT_START}f${FTS_HIGHLIGHT_END}`, 5)).toBe('…cdef')
expect(makeSnippet(` x—${FTS_HIGHLIGHT_START}café${FTS_HIGHLIGHT_END}\n y `, 20)) expect(makeSnippet(` x—${FTS_HIGHLIGHT_START}café${FTS_HIGHLIGHT_END}\n y `, 20))
.toBe('x—café y') .toBe('x—café y')

View File

@@ -387,6 +387,8 @@ describe('SQLite session search', () => {
{ path: '' }, { path: '' },
{ path: ':memory:', defaultLimit: 0 }, { path: ':memory:', defaultLimit: 0 },
{ path: ':memory:', maxLimit: 0 }, { path: ':memory:', maxLimit: 0 },
{ path: ':memory:', defaultLimit: 1e100 },
{ path: ':memory:', maxLimit: 1e100 },
{ path: ':memory:', snippetChars: 0 }, { path: ':memory:', snippetChars: 0 },
{ path: ':memory:', defaultLimit: 3, maxLimit: 2 }, { path: ':memory:', defaultLimit: 3, maxLimit: 2 },
{ path: ':memory:', journalMode: 'memory' }, { path: ':memory:', journalMode: 'memory' },
@@ -462,6 +464,39 @@ describe('SQLite reconciliation and source lifecycle', () => {
.rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND')) .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
}) })
it('uses the reconciled persistence binding through the query boundary', async () => {
const durable = header('post-reconcile-unmount')
TestPersistence.reset([{ meta: durable, events: [
...messageEvents('durable needle', 1),
{ ...messageEvents('durable needle again', 2)[0]!, seq: 1 },
] }])
const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 2 })
const persistence = await ctx.plugin(TestPersistence)
const internals = ctx.sessionSearch as unknown as {
_reconcile(signal: AbortSignal | undefined): Promise<{
identity: symbol
service?: SessionPersistence
}>
}
const reconcile = internals._reconcile.bind(internals)
const boundary = vi.spyOn(internals, '_reconcile').mockImplementation(async (signal) => {
const binding = await reconcile(signal)
await persistence.dispose()
return binding
})
const page = await ctx.sessionSearch.searchEvents({
sessionId: durable.id,
query: 'needle',
limit: 1,
})
expect(page.items).toMatchObject([{ sessionId: durable.id }])
expect(page.nextCursor).toEqual(expect.any(String))
boundary.mockRestore()
await expect(ctx.sessionSearch.searchEvents({ sessionId: durable.id, query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
})
it('discards a stale list rejection when persistence unmounts during observation', async () => { it('discards a stale list rejection when persistence unmounts during observation', async () => {
const durable = header('racing') const durable = header('racing')
TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }]) TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
@@ -561,6 +596,22 @@ describe('SQLite reconciliation and source lifecycle', () => {
expect(TestPersistence.loads.get(added.id)).toBe(1) expect(TestPersistence.loads.get(added.id)).toBe(1)
}) })
it('fails after one retry when persistence snapshots keep changing', async () => {
const durable = header('continuous-mutation')
TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
const ctx = await liveContext()
await ctx.plugin(TestPersistence)
let lists = 0
TestPersistence.snapshotEffect = () => {
lists += 1
TestPersistence.set({ meta: durable, events: messageEvents(`durable needle ${lists}`) })
}
await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
expect(lists).toBe(4)
})
it('retries if the persistence binding changes while live sessions are observed', async () => { it('retries if the persistence binding changes while live sessions are observed', async () => {
const durable = header('live-boundary-retry') const durable = header('live-boundary-retry')
TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }]) TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])