feat(session-query): add SQLite full-text search

This commit is contained in:
Hypatia May
2026-07-15 10:51:38 +08:00
parent e9f0c37745
commit ecf90ff382
38 changed files with 3181 additions and 120 deletions

View File

@@ -25,7 +25,7 @@ Packages are grouped by modular role at `packages/<group>/<pkg>/`. The group dir
| [`cordis/`](cordis/README.md) | Self-referential runtime toolset: inspect the live runtime's plugins and services, mount/unmount model-written plugins ([design](../docs/rfc/implemented/feature/2026-07-08-self-referential-cordis-toolset.md)) | Product — stable surface |
| [`hooks/`](hooks/README.md) | Hook bridges + the shared Claude Code / Codex wire-protocol library | Product — stable surface |
| [`session-persistence/`](session-persistence/README.md) | Persistence capability family: the seam + JSONL/SQLite backends | Product — stable surface |
| [`session-query/`](session-query/README.md) | Session retrieval family: logical corpus, surface records, and bounded exact reads | Product — stable surface |
| [`session-query/`](session-query/README.md) | Session retrieval family: exact reads, semantic filtering, and SQLite full-text search | Product — stable surface |
| [`ui/`](ui/README.md) | Editor/client integration surfaces: ACP bridge, JSON-RPC SDK server, app packages, user-approval and user-interaction seams, ask-user tool | Product — stable surface |
| [`support/`](support/README.md) | Dev/test/example infrastructure (invariants, replay adapter, subagent mock) | Support — lower compatibility expectations |
| [`util/`](util/README.md) | Low-level zero-dependency utilities shared across groups (the `Branded<B>` primitive) | Support — small, stable, harness-dep-free |

View File

@@ -157,6 +157,7 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
methods: [
'listSessions(): Promise<SessionRecord[]>',
'async listEvents(sessionId: SessionId): Promise<SessionEventRecord[]>',
'async filterEvents( sessionId: SessionId, filters: readonly SessionEventResultFilter[], ): Promise<SessionEventSearchDocument[]>',
'async readEvent(request: SessionEventReadRequest): Promise<SessionEventWindow>',
],
},
@@ -174,6 +175,14 @@ export const SERVICE_API: readonly ServiceApiEntry[] = [
'fork(source: SessionForkSource, boundary?: number, childSessionId?: SessionId): Session',
],
},
{
key: 'sessionSearch',
summary: 'Abstract full-text search service implemented by one concrete backend.',
methods: [
'abstract searchSessions( request: SessionSearchRequest, exec?: SessionSearchExecContext, ): Promise<SessionSearchPage<SessionSearchHit>>',
'abstract searchEvents( request: SessionEventSearchRequest, exec?: SessionSearchExecContext, ): Promise<SessionSearchPage<SessionEventSearchHit>>',
],
},
{
key: 'skills',
summary: 'Registry of skill providers.',
@@ -787,6 +796,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'SendOptions',
declaration: 'export interface SendOptions {\n source?: MessageSource;\n}',
},
{
name: 'SessionAvailability',
declaration: 'export type SessionAvailability = \'live\' | \'persisted\';',
},
{
name: 'SessionEvent',
declaration: 'export type SessionEvent<T extends SessionEventType = SessionEventType> = {\n [K in SessionEventType]: {\n type: K;\n seq: number;\n time: number;\n data: SessionEventMap[K];\n } & (K extends SurfaceEventType ? {\n sourceEventSeqs?: number[];\n surfaceOp?: SurfaceOp;\n } : object);\n}[T];',
@@ -795,6 +808,10 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'SessionEventMap',
declaration: 'export interface SessionEventMap {\n \'turn/start\': {\n turn: number;\n trigger: TurnTrigger;\n };\n \'turn/end\': {\n turn: number;\n reason: TurnEndReason;\n };\n \'step/start\': {\n turn: number;\n step: number;\n };\n \'step/end\': {\n turn: number;\n step: number;\n };\n \'user/message\': {\n content: ContentBlock[];\n source: MessageSource;\n };\n \'prompt/blocked\': {\n content: ContentBlock[];\n source: MessageSource;\n reason: string;\n };\n \'context/message\': {\n content: ContentBlock[];\n source: MessageSource;\n };\n \'assistant/chunk\': {\n turn: number;\n step: number;\n chunk: StreamChunk;\n };\n \'assistant/message\': {\n turn: number;\n step: number;\n content: ContentBlock[];\n usage?: TokenUsage;\n };\n \'tool/call\': {\n turn: number;\n step: number;\n callId: CallId;\n name: string;\n arguments: string;\n };\n \'tool/result\': {\n turn: number;\n step: number;\n callId: CallId;\n content: ContentBlock[];\n isError: boolean;\n error?: {\n name: string;\n code: string;\n };\n meta?: unknown;\n };\n \'steering/message\': {\n turn: number;\n content: ContentBlock[];\n source: MessageSource;\n };\n \'todo/write\': {\n todos: TodoItem[];\n };\n \'request/header\': {\n header: E /* …truncated — full shape in source */',
},
{
name: 'SessionEventMetadataFilter',
declaration: 'export type SessionEventMetadataFilter = Exclude<SessionEventResultFilter, {\n kind: \'text\';\n}>;',
},
{
name: 'SessionEventReadRequest',
declaration: 'export interface SessionEventReadRequest {\n sessionId: SessionId;\n seq: number;\n before?: number;\n after?: number;\n}',
@@ -803,6 +820,22 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'SessionEventRecord',
declaration: 'export interface SessionEventRecord {\n sessionId: SessionId;\n seq: number;\n type: SessionEventType;\n time: number;\n surface: SessionEventSurface;\n}',
},
{
name: 'SessionEventResultFilter',
declaration: 'export type SessionEventResultFilter = ({\n kind: \'seq\';\n} & SessionResultRange) | ({\n kind: \'time\';\n} & SessionResultRange) | {\n kind: \'type\';\n values: readonly SessionEventType[];\n} | {\n kind: \'surface\';\n values: readonly SessionEventSurface[];\n} | {\n kind: \'text\';\n text: string;\n};',
},
{
name: 'SessionEventSearchDocument',
declaration: 'export interface SessionEventSearchDocument extends SessionEventRecord {\n text: string;\n}',
},
{
name: 'SessionEventSearchHit',
declaration: 'export interface SessionEventSearchHit extends SessionEventRecord {\n snippet: string;\n}',
},
{
name: 'SessionEventSearchRequest',
declaration: 'export interface SessionEventSearchRequest {\n sessionId: SessionId;\n query: string;\n filters?: readonly SessionEventMetadataFilter[];\n limit?: number;\n cursor?: string;\n}',
},
{
name: 'SessionEventSurface',
declaration: 'export type SessionEventSurface = \'current\' | \'shadowed\' | \'log-only\';',
@@ -831,6 +864,30 @@ export const TYPE_API: readonly TypeApiEntry[] = [
name: 'SessionRecord',
declaration: 'export interface SessionRecord {\n header: SessionHeader;\n live: boolean;\n persisted: boolean;\n}',
},
{
name: 'SessionResultFilter',
declaration: 'export type SessionResultFilter = {\n kind: \'id\';\n values: readonly SessionId[];\n} | {\n kind: \'cwd\';\n values: readonly (string | null)[];\n} | ({\n kind: \'created-at\';\n} & SessionResultRange) | {\n kind: \'parent\';\n values: readonly (SessionId | null)[];\n} | {\n kind: \'availability\';\n values: readonly SessionAvailability[];\n};',
},
{
name: 'SessionResultRange',
declaration: 'export interface SessionResultRange {\n from?: number;\n to?: number;\n}',
},
{
name: 'SessionSearchExecContext',
declaration: 'export interface SessionSearchExecContext {\n signal?: AbortSignal;\n}',
},
{
name: 'SessionSearchHit',
declaration: 'export interface SessionSearchHit extends SessionRecord {\n bestMatch: SessionEventSearchHit;\n}',
},
{
name: 'SessionSearchPage',
declaration: 'export interface SessionSearchPage<T> {\n items: readonly T[];\n nextCursor?: string;\n}',
},
{
name: 'SessionSearchRequest',
declaration: 'export interface SessionSearchRequest {\n query: string;\n sessionFilters?: readonly SessionResultFilter[];\n eventFilters?: readonly SessionEventMetadataFilter[];\n limit?: number;\n cursor?: string;\n}',
},
{
name: 'SkillCandidate',
declaration: 'export interface SkillCandidate extends SkillSummary {\n readonly rank: number;\n readonly locator: unknown;\n readonly path?: string;\n readonly metadata?: Readonly<Record<string, unknown>>;\n}',

View File

@@ -1,9 +1,10 @@
# session-query/ — session retrieval capability family
Trusted exact reads over live and durable session logs. Phase one contains one interface package that owns `ctx.sessionQuery`, logical-corpus precedence, surface classification, and bounded event reads.
Trusted exact reads, provider-independent semantic filtering, and SQLite full-text search over live and durable session logs.
| Package | Role | ctx key |
|---|---|---|
| [`session-query/`](session-query/README.md) | Logical-corpus and exact-event read service | `ctx.sessionQuery` |
| [`session-query/`](session-query/README.md) | Logical-corpus reads, semantic extraction/filtering, and the abstract search seam | `ctx.sessionQuery`, `ctx.sessionSearch` |
| [`session-query-sqlite/`](session-query-sqlite/README.md) | SQLite FTS5 search with persistent bases and live overlays | `ctx.sessionSearch` |
The family is independent of compaction: it reads the canonical session log but does not participate in compaction policy or execution. Full-text search remains proposed as a phase-two SQLite package rather than a speculative provider seam in this interface package.
The family is independent of compaction: it reads the canonical session log but does not participate in compaction policy or execution. Search uses one abstract service and one concrete owner, not a provider registry or coordinator.

View File

@@ -0,0 +1,35 @@
# @deepseek-ai/dsh-session-query-sqlite
SQLite FTS5 implementation of `ctx.sessionSearch`. The service searches the live-preferred logical session corpus, groups cross-session results by their strongest event, and keeps provider-specific BM25 scores private.
## Search contract
`searchSessions(request, exec?)` returns `SessionSearchHit` pages across the corpus; `searchEvents(request, exec?)` returns `SessionEventSearchHit` pages within one session. Queries are required, trimmed, whitespace-normalized literal phrases. FTS5 syntax such as quotes, `OR`, `NEAR`, and `*` is treated as data rather than executable MATCH syntax. Metadata filters are parameterized SQL predicates applied before ranking.
Ordering is deterministic: relevance first, then event time, session id where applicable, and seq. Cross-session results expose the selected event as `bestMatch`; both scopes return plain-text snippets bounded in Unicode code points. Cursors are opaque, bind to the normalized request and service instance, and fail when the relevant generation changes. A within-session cursor survives unrelated-session changes; a cross-session cursor does not.
All three surfaces (`current`, `shadowed`, and `log-only`) are searchable by default. Pass a surface filter to narrow them.
## Source and index lifecycle
The service requires `ctx.sessions` and observes optional `ctx.sessionPersistence` dynamically. One serialized state machine observes complete sources, extracts shared semantic documents, reconciles changes transactionally, and runs the query. Stable fingerprints preserve unchanged persisted rows and generations; new, changed, and deleted durable sessions reconcile on the next search. 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.
The database is disposable but reset is guarded: a recognized incompatible search schema rebuilds in place, while an unrelated or canonical database is refused. Never point `path` at the session-persistence database.
## Configuration
| Key | Default | Contract |
|---|---:|---|
| `path` | required | Dedicated derived-index SQLite path; `:memory:` is supported. |
| `journalMode` | `wal` | `wal`, `delete`, `truncate`, or `persist`. |
| `defaultLimit` | `20` | Page size when a request omits `limit`. |
| `maxLimit` | `100` | Largest accepted request page size. |
| `snippetChars` | `240` | Maximum snippet length in Unicode code points. |
## Tokenizer and limits
The index uses FTS5 `unicode61`. In the implementation experiment it supported the two-character query `AI` and produced an index about 2.1× smaller than the trigram alternative. The trade-off is token/phrase recall rather than arbitrary substring recall: `AI` does not match the token `BRAID`. Use `ctx.sessionQuery.filterEvents()` with a `text` clause when a literal whitespace-flexible substring scan is required.
Abort signals stop queued work and caller waits around asynchronous source observation. Node's synchronous `DatabaseSync` API cannot interrupt a MATCH statement already executing on the JavaScript thread; the signal is checked immediately before and after the serialized observation/reconciliation boundary.

View File

@@ -0,0 +1,46 @@
{
"name": "@deepseek-ai/dsh-session-query-sqlite",
"description": "SQLite FTS5 implementation of ctx.sessionSearch",
"version": "0.0.1",
"private": true,
"type": "module",
"main": "lib/index.js",
"types": "lib/types/index.d.ts",
"exports": {
".": {
"types": "./lib/types/index.d.ts",
"default": "./lib/index.js"
},
"./src/*": "./src/*",
"./package.json": "./package.json"
},
"files": [
"lib/index.js",
"lib/types/**/*.d.ts",
"lib/types/**/*.d.ts.map",
"src"
],
"license": "BSD-3-Clause",
"peerDependencies": {
"@deepseek-ai/dsh-session": "^0.0.1",
"@deepseek-ai/dsh-session-persistence": "^0.0.1",
"@deepseek-ai/dsh-session-query": "^0.0.1",
"cordis": "^4.0.0-rc.6"
},
"peerDependenciesMeta": {
"@deepseek-ai/dsh-session-persistence": {
"optional": true
}
},
"dependencies": {
"schemastery": "^3.18.0"
},
"devDependencies": {
"@cordisjs/plugin-loader": "workspace:^",
"@deepseek-ai/dsh-session": "workspace:^",
"@deepseek-ai/dsh-session-persistence": "workspace:^",
"@deepseek-ai/dsh-session-persistence-sqlite": "workspace:^",
"@deepseek-ai/dsh-session-query": "workspace:^",
"cordis": "^4.0.0-rc.6"
}
}

View File

@@ -0,0 +1,765 @@
/**
* SQLite FTS5 search over the live-preferred logical session corpus.
*
* @module @deepseek-ai/dsh-session-query-sqlite
*/
import { createHash, randomUUID } from 'node:crypto'
import { DatabaseSync } from 'node:sqlite'
import { Context } from 'cordis'
import z from 'schemastery'
import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-ai/dsh-session'
import type SessionPersistence from '@deepseek-ai/dsh-session-persistence'
import {
SessionQueryError,
SessionSearchService,
assertSessionHeadersCompatible,
buildSessionEventSearchDocuments,
} from '@deepseek-ai/dsh-session-query'
import type {
SessionEventSearchDocument,
SessionEventSearchHit,
SessionEventSearchRequest,
SessionSearchExecContext,
SessionSearchHit,
SessionSearchPage,
SessionSearchRequest,
} from '@deepseek-ai/dsh-session-query'
import {
type JournalMode,
openSearchDatabase,
} from './schema.ts'
import {
type NormalizedEventRequest,
type NormalizedSessionRequest,
buildEventWhere,
buildSessionWhere,
makeSnippet,
normalizeEventRequest,
normalizeSessionRequest,
quoteFtsData,
requestFingerprint,
} from './query.ts'
export {
SESSION_QUERY_SQLITE_APPLICATION_ID,
SESSION_QUERY_SQLITE_SCHEMA_VERSION,
type JournalMode,
} from './schema.ts'
/** Default result page size. */
export const SESSION_QUERY_SQLITE_DEFAULT_LIMIT = 20
/** Maximum accepted result page size. */
export const SESSION_QUERY_SQLITE_MAX_LIMIT = 100
/** Default maximum snippet length in Unicode code points. */
export const SESSION_QUERY_SQLITE_SNIPPET_CHARS = 240
/** SQLite session-search configuration. */
export interface Config {
/** Dedicated derived-index path; `:memory:` is supported for tests. */
path: string
/** SQLite journal mode. Defaults to `wal`. */
journalMode?: JournalMode
/** Page size when a request omits `limit`. Defaults to 20. */
defaultLimit?: number
/** Largest accepted page size. Defaults to 100. */
maxLimit?: number
/** Maximum snippet length in Unicode code points. Defaults to 240. */
snippetChars?: number
}
interface ResolvedConfig {
path: string
journalMode: JournalMode
defaultLimit: number
maxLimit: number
snippetChars: number
}
interface ObservedSession {
header: SessionHeader
events: SessionEvent[]
documents: SessionEventSearchDocument[]
fingerprint: string
}
interface Observation {
persistence: SessionPersistence | undefined
persistenceRevision: number
persisted: Map<SessionId, ObservedSession>
live: Map<SessionId, ObservedSession>
}
interface IndexedRow {
id: string
fingerprint: string
generation: number
}
interface SearchRow {
session_id: string
version: number
created_at: number
cwd: string | null
parent_session: string | null
seed_length: number | null
live: number
persisted: number
seq: number
type: string
time: number
surface: string
text: string
score: number
}
interface CursorPayload {
version: 1
instance: string
scope: 'sessions' | 'events'
fingerprint: string
generation: string
offset: number
}
/** Concrete SQLite owner of `ctx.sessionSearch`. */
export class SessionSearchSqlite extends SessionSearchService {
static inject = ['sessions']
static Config: z<Config> = z.object({
path: z.string().required(),
journalMode: z.union(['wal', 'delete', 'truncate', 'persist'] as const).default('wal'),
defaultLimit: z.number().step(1).min(1).default(SESSION_QUERY_SQLITE_DEFAULT_LIMIT),
maxLimit: z.number().step(1).min(1).default(SESSION_QUERY_SQLITE_MAX_LIMIT),
snippetChars: z.number().step(1).min(1).default(SESSION_QUERY_SQLITE_SNIPPET_CHARS),
})
/** Validated and defaulted backend configuration. */
readonly config: ResolvedConfig
private readonly _instance = randomUUID()
private readonly _ready: Promise<void>
private _db: DatabaseSync | undefined
private _persistence: SessionPersistence | undefined
private _persistenceBinding: object | undefined
private _persistenceRevision = 0
private _lastPersistenceRevision: number | undefined
private _persistenceEpoch = 0
private _globalGeneration = 0
private _localGeneration = 0
private _tail: Promise<void> = Promise.resolve()
private _closed = false
constructor(ctx: Context, config: Config) {
super(ctx)
this.config = resolveConfig(config)
this._ready = this._open()
ctx.effect(() => {
const fiber = ctx.inject(['sessionPersistence'], (childCtx: Context) => {
const service = childCtx.sessionPersistence
const binding = {}
this._persistenceBinding = binding
this._persistence = service
this._persistenceRevision += 1
childCtx.effect(() => () => {
/* v8 ignore next -- a stale optional-service disposer cannot clear a replacement */
if (this._persistenceBinding !== binding) return
this._persistenceBinding = undefined
this._persistence = undefined
this._persistenceRevision += 1
}, 'sessionSearchSqlite.persistenceBinding')
})
return () => void fiber.dispose()
}, 'sessionSearchSqlite.optionalPersistence')
ctx.effect(() => async () => this.close(), 'sessionSearchSqlite.close')
}
override async searchSessions(
request: SessionSearchRequest,
exec?: SessionSearchExecContext,
): Promise<SessionSearchPage<SessionSearchHit>> {
const normalized = normalizeSessionRequest(request, this.config)
return this._serialized(exec?.signal, async () => {
await this._ensureReady(exec?.signal)
await this._reconcile(exec?.signal)
assertNotAborted(exec?.signal)
const generation = String(this._globalGeneration)
const fingerprint = requestFingerprint(normalized)
const offset = normalized.cursor === undefined
? 0
: decodeCursor(normalized.cursor, this._instance, 'sessions', fingerprint, generation)
const rows = this._querySessions(normalized, offset)
return page(rows, normalized.limit, row => this._sessionHit(row, normalized.query), cursorOffset => encodeCursor({
version: 1,
instance: this._instance,
scope: 'sessions',
fingerprint,
generation,
offset: cursorOffset,
}), offset)
})
}
override async searchEvents(
request: SessionEventSearchRequest,
exec?: SessionSearchExecContext,
): Promise<SessionSearchPage<SessionEventSearchHit>> {
const normalized = normalizeEventRequest(request, this.config)
return this._serialized(exec?.signal, async () => {
await this._ensureReady(exec?.signal)
await this._reconcile(exec?.signal)
assertNotAborted(exec?.signal)
const generation = this._targetGeneration(normalized.sessionId)
const fingerprint = requestFingerprint(normalized)
const offset = normalized.cursor === undefined
? 0
: decodeCursor(normalized.cursor, this._instance, 'events', fingerprint, generation)
const rows = this._queryEvents(normalized, offset)
return page(rows, normalized.limit, row => this._eventHit(row, normalized.query), cursorOffset => encodeCursor({
version: 1,
instance: this._instance,
scope: 'events',
fingerprint,
generation,
offset: cursorOffset,
}), offset)
})
}
/** Close the database after every accepted operation reaches quiescence. */
async close(): Promise<void> {
if (this._closed) return
this._closed = true
await this._tail
try {
await this._ready
} catch {
// Opening already closed a partially-created handle; disposal only waits.
}
this._db?.close()
this._db = undefined
}
private async _open(): Promise<void> {
this._db = await openSearchDatabase(this.config.path, this.config.journalMode)
const state = this._db.prepare(
'SELECT global_generation FROM search_state WHERE singleton = 1',
).get() as { global_generation: number }
this._globalGeneration = state.global_generation
this._localGeneration = state.global_generation
}
private async _ensureReady(signal: AbortSignal | undefined): Promise<void> {
try {
await waitWithAbort(this._ready, signal)
} catch (error: unknown) {
if (isAbort(error)) throw error
throw new SessionQueryError(
`session-search SQLite index failed to open: ${errorMessage(error)}`,
'SESSION_QUERY_INDEX_FAILED',
{ cause: error },
)
}
}
private async _serialized<T>(signal: AbortSignal | undefined, operation: () => Promise<T>): Promise<T> {
if (this._isClosed()) throw indexClosed()
let release!: () => void
const gate = new Promise<void>((resolve) => { release = resolve })
const prior = this._tail
this._tail = prior.then(() => gate)
try {
await waitWithAbort(prior, signal)
} catch (error: unknown) {
release()
throw error
}
if (this._isClosed()) {
release()
throw indexClosed()
}
try {
assertNotAborted(signal)
return await operation()
} finally {
release()
}
}
private async _reconcile(signal: AbortSignal | undefined): Promise<void> {
const observation = await this._observeStable(signal)
assertNotAborted(signal)
const db = this._requireDb()
const persistedRows = db.prepare(
'SELECT id, fingerprint, generation FROM persisted_sessions',
).all() as unknown as IndexedRow[]
const liveRows = db.prepare(
'SELECT id, fingerprint, generation FROM temp.live_sessions',
).all() as unknown as IndexedRow[]
const persistedById = new Map(persistedRows.map(row => [row.id as SessionId, row]))
const liveById = new Map(liveRows.map(row => [row.id as SessionId, row]))
const persistentChanges = observation.persistence === undefined
? []
: [...observation.persisted.values()].filter(entry => persistedById.get(entry.header.id)?.fingerprint !== entry.fingerprint)
const persistentDeletes = observation.persistence === 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 liveDeletes = liveRows.filter(row => !observation.live.has(row.id as SessionId))
const pointerChanged = this._lastPersistenceRevision !== undefined
&& this._lastPersistenceRevision !== observation.persistenceRevision
const hasWrites = persistentChanges.length > 0
|| persistentDeletes.length > 0
|| liveChanges.length > 0
|| liveDeletes.length > 0
let nextMainGeneration = this._mainGeneration()
let nextLocalGeneration = this._localGeneration
if (persistentChanges.length > 0 || persistentDeletes.length > 0) nextMainGeneration += 1
const liveReplacements = liveChanges.map((entry) => {
nextLocalGeneration = Math.max(nextLocalGeneration, nextMainGeneration) + 1
return { entry, generation: nextLocalGeneration }
})
if (hasWrites) {
let began = false
try {
db.exec('BEGIN IMMEDIATE')
began = true
for (const row of persistentDeletes) this._deleteSession('persisted', row.id as SessionId)
for (const entry of persistentChanges) this._replaceSession('persisted', entry, nextMainGeneration)
if (persistentChanges.length > 0 || persistentDeletes.length > 0) {
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._replaceSession('live', entry, generation)
}
db.exec('COMMIT')
} catch (error: unknown) {
/* v8 ignore next -- a BEGIN failure has no transaction to roll back; the common wrapper still reports it. */
if (began) {
/* v8 ignore next 5 -- ROLLBACK failure requires a SQLite double fault; the original failure remains actionable. */
try {
db.exec('ROLLBACK')
} catch {
// The original SQLite failure remains the actionable cause.
}
}
throw new SessionQueryError(
`session-search reconciliation failed: ${errorMessage(error)}`,
'SESSION_QUERY_INDEX_FAILED',
{ cause: error },
)
}
}
if (hasWrites || pointerChanged) this._globalGeneration += 1
if (pointerChanged) this._persistenceEpoch += 1
this._localGeneration = nextLocalGeneration
this._lastPersistenceRevision = observation.persistenceRevision
}
private async _observeStable(signal: AbortSignal | undefined): Promise<Observation> {
for (;;) {
assertNotAborted(signal)
const persistence = this._persistence
const persistenceRevision = this._persistenceRevision
const persisted = new Map<SessionId, ObservedSession>()
if (persistence !== undefined) {
try {
const headers = await waitWithAbort(persistence.list(), signal)
for (const listed of headers) {
const loaded = await waitWithAbort(persistence.load(listed.id), signal)
assertSessionHeadersCompatible(listed, loaded.meta)
persisted.set(listed.id, observeSession(loaded.meta, loaded.events))
}
} catch (error: unknown) {
if (error instanceof SessionQueryError) throw error
throw new SessionQueryError(
`session-search persistence observation failed: ${errorMessage(error)}`,
'SESSION_QUERY_PERSISTENCE_FAILED',
{ cause: error },
)
}
}
const live = new Map<SessionId, ObservedSession>()
for (const session of this.ctx.sessions.list()) {
const observed = observeLive(session)
const durable = persisted.get(session.id)
if (durable !== undefined) assertSessionHeadersCompatible(observed.header, durable.header)
live.set(session.id, observed)
}
if (this._persistenceRevision === persistenceRevision) {
return { persistence, persistenceRevision, persisted, live }
}
}
}
private _mainGeneration(): number {
const row = this._requireDb().prepare(
'SELECT global_generation FROM search_state WHERE singleton = 1',
).get() as { global_generation: number }
return row.global_generation
}
private _deleteSession(source: 'persisted' | 'live', id: SessionId): void {
const db = this._requireDb()
if (source === 'persisted') {
db.prepare('DELETE FROM persisted_docs WHERE session_id = ?').run(id)
db.prepare('DELETE FROM persisted_sessions WHERE id = ?').run(id)
} else {
db.prepare('DELETE FROM temp.live_docs WHERE session_id = ?').run(id)
db.prepare('DELETE FROM temp.live_sessions WHERE id = ?').run(id)
}
}
private _replaceSession(source: 'persisted' | 'live', entry: ObservedSession, generation: number): void {
this._deleteSession(source, entry.header.id)
const db = this._requireDb()
const sessionTable = source === 'persisted' ? 'persisted_sessions' : 'temp.live_sessions'
const docsTable = source === 'persisted' ? 'persisted_docs' : 'temp.live_docs'
db.prepare(`
INSERT INTO ${sessionTable}
(id, version, created_at, cwd, parent_session, seed_length, fingerprint, generation)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
`).run(
entry.header.id,
entry.header.version,
entry.header.createdAt,
entry.header.cwd ?? null,
entry.header.parentSession ?? null,
entry.header.seedLength ?? null,
entry.fingerprint,
generation,
)
const insert = db.prepare(`
INSERT INTO ${docsTable} (text, session_id, seq, type, time, surface)
VALUES (?, ?, ?, ?, ?, ?)
`)
for (const document of entry.documents) {
insert.run(document.text, document.sessionId, document.seq, document.type, document.time, document.surface)
}
}
private _querySessions(request: NormalizedSessionRequest, offset: number): SearchRow[] {
const selected = selectedDocumentsSql()
const sessionWhere = buildSessionWhere(request.sessionFilters)
const eventWhere = buildEventWhere(request.eventFilters)
const where = [sessionWhere.sql, eventWhere.sql].filter(Boolean).join(' AND ')
return this._requireDb().prepare(`
${selected.sql},
filtered AS (
SELECT * FROM matched ${where.length === 0 ? '' : `WHERE ${where}`}
),
ranked AS (
SELECT *, ROW_NUMBER() OVER (
PARTITION BY session_id
ORDER BY score ASC, time DESC, seq DESC
) AS event_rank
FROM filtered
)
SELECT * FROM ranked
WHERE event_rank = 1
ORDER BY score ASC, time DESC, session_id ASC, seq DESC
LIMIT ? OFFSET ?
`).all(
quoteFtsData(request.query),
this._persistence === undefined ? 0 : 1,
this._persistence === undefined ? 0 : 1,
quoteFtsData(request.query),
...sessionWhere.params,
...eventWhere.params,
request.limit + 1,
offset,
) as unknown as SearchRow[]
}
private _queryEvents(request: NormalizedEventRequest, offset: number): SearchRow[] {
const selected = selectedDocumentsSql()
const eventWhere = buildEventWhere(request.filters)
const where = ['session_id = ?', eventWhere.sql].filter(Boolean).join(' AND ')
return this._requireDb().prepare(`
${selected.sql}
SELECT * FROM matched
WHERE ${where}
ORDER BY score ASC, time DESC, seq DESC
LIMIT ? OFFSET ?
`).all(
quoteFtsData(request.query),
this._persistence === undefined ? 0 : 1,
this._persistence === undefined ? 0 : 1,
quoteFtsData(request.query),
request.sessionId,
...eventWhere.params,
request.limit + 1,
offset,
) as unknown as SearchRow[]
}
private _targetGeneration(sessionId: SessionId): string {
const db = this._requireDb()
const live = db.prepare(
'SELECT generation FROM temp.live_sessions WHERE id = ?',
).get(sessionId) as { generation: number } | undefined
if (live !== undefined) return `live:${live.generation}`
if (this._persistence !== undefined) {
const persisted = db.prepare(
'SELECT generation FROM persisted_sessions WHERE id = ?',
).get(sessionId) as { generation: number } | undefined
if (persisted !== undefined) return `persisted:${this._persistenceEpoch}:${persisted.generation}`
}
throw new SessionQueryError(
`session "${sessionId}" not found`,
'SESSION_QUERY_SESSION_NOT_FOUND',
)
}
private _sessionHit(row: SearchRow, query: string): SessionSearchHit {
return {
header: rowHeader(row),
live: row.live === 1,
persisted: row.persisted === 1,
bestMatch: this._eventHit(row, query),
}
}
private _eventHit(row: SearchRow, query: string): SessionEventSearchHit {
return {
sessionId: row.session_id as SessionId,
seq: row.seq,
type: row.type as SessionEventSearchHit['type'],
time: row.time,
surface: row.surface as SessionEventSearchHit['surface'],
snippet: makeSnippet(row.text, query, this.config.snippetChars),
}
}
private _requireDb(): DatabaseSync {
/* v8 ignore next -- callers await `_ready`; this guards lifecycle misuse */
if (this._db === undefined) throw indexClosed()
return this._db
}
private _isClosed(): boolean {
return this._closed
}
}
function selectedDocumentsSql(): { sql: string } {
return {
sql: `WITH matched AS (
SELECT
pd.session_id AS session_id,
ps.version AS version,
ps.created_at AS created_at,
ps.cwd AS cwd,
ps.parent_session AS parent_session,
ps.seed_length AS seed_length,
0 AS live,
1 AS persisted,
CAST(pd.seq AS INTEGER) AS seq,
pd.type AS type,
CAST(pd.time AS INTEGER) AS time,
pd.surface AS surface,
pd.text AS text,
bm25(persisted_docs) AS score
FROM persisted_docs AS pd
JOIN persisted_sessions AS ps ON ps.id = pd.session_id
WHERE persisted_docs MATCH ?
AND ? = 1
AND NOT EXISTS (SELECT 1 FROM temp.live_sessions AS ls WHERE ls.id = pd.session_id)
UNION ALL
SELECT
ld.session_id AS session_id,
ls.version AS version,
ls.created_at AS created_at,
ls.cwd AS cwd,
ls.parent_session AS parent_session,
ls.seed_length AS seed_length,
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,
CAST(ld.seq AS INTEGER) AS seq,
ld.type AS type,
CAST(ld.time AS INTEGER) AS time,
ld.surface AS surface,
ld.text AS text,
bm25(live_docs) AS score
FROM temp.live_docs AS ld
JOIN temp.live_sessions AS ls ON ls.id = ld.session_id
WHERE live_docs MATCH ?
)`,
}
}
function observeLive(session: Session): ObservedSession {
return observeSession(
structuredClone(session.header),
session.events.map(event => structuredClone(event)),
)
}
function observeSession(header: SessionHeader, events: readonly SessionEvent[]): ObservedSession {
const detachedHeader = structuredClone(header)
const detachedEvents = events.map(event => structuredClone(event))
return {
header: detachedHeader,
events: detachedEvents,
documents: buildSessionEventSearchDocuments(detachedHeader.id, detachedEvents),
fingerprint: createHash('sha256')
.update(JSON.stringify({ header: detachedHeader, events: detachedEvents }))
.digest('base64url'),
}
}
function rowHeader(row: SearchRow): SessionHeader {
return {
version: row.version,
id: row.session_id as SessionId,
createdAt: row.created_at,
...row.cwd === null ? {} : { cwd: row.cwd },
...row.parent_session === null ? {} : { parentSession: row.parent_session as SessionId },
...row.seed_length === null ? {} : { seedLength: row.seed_length },
}
}
function page<Row, Item>(
rows: readonly Row[],
limit: number,
convert: (row: Row) => Item,
nextCursor: (offset: number) => string,
offset: number,
): SessionSearchPage<Item> {
const hasMore = rows.length > limit
return {
items: rows.slice(0, limit).map(convert),
...hasMore ? { nextCursor: nextCursor(offset + limit) } : {},
}
}
function encodeCursor(payload: CursorPayload): string {
return Buffer.from(JSON.stringify(payload), 'utf8').toString('base64url')
}
function decodeCursor(
cursor: string,
instance: string,
scope: CursorPayload['scope'],
fingerprint: string,
generation: string,
): number {
let decoded: Partial<CursorPayload>
try {
decoded = JSON.parse(Buffer.from(cursor, 'base64url').toString('utf8')) as Partial<CursorPayload>
} catch (error: unknown) {
throw invalidCursor(error)
}
if (
decoded.version !== 1
|| decoded.instance !== instance
|| decoded.scope !== scope
|| decoded.fingerprint !== fingerprint
|| !Number.isInteger(decoded.offset)
|| decoded.offset === undefined
|| decoded.offset < 0
) {
throw invalidCursor(new Error('cursor does not belong to this normalized request'))
}
if (decoded.generation !== generation) {
throw new SessionQueryError(
'session-search cursor is stale because its relevant corpus changed',
'SESSION_QUERY_STALE_CURSOR',
)
}
return decoded.offset
}
function invalidCursor(cause: unknown): SessionQueryError {
return new SessionQueryError(
'session-search cursor is invalid',
'SESSION_QUERY_INVALID_CURSOR',
{ cause },
)
}
function resolveConfig(config: Config): ResolvedConfig {
const resolved: ResolvedConfig = {
path: config.path,
journalMode: config.journalMode ?? 'wal',
defaultLimit: config.defaultLimit ?? SESSION_QUERY_SQLITE_DEFAULT_LIMIT,
maxLimit: config.maxLimit ?? SESSION_QUERY_SQLITE_MAX_LIMIT,
snippetChars: config.snippetChars ?? SESSION_QUERY_SQLITE_SNIPPET_CHARS,
}
if (typeof resolved.path !== 'string' || resolved.path.trim().length === 0) {
throw invalidConfig('path must not be blank')
}
assertPositiveInteger('defaultLimit', resolved.defaultLimit)
assertPositiveInteger('maxLimit', resolved.maxLimit)
assertPositiveInteger('snippetChars', resolved.snippetChars)
if (resolved.defaultLimit > resolved.maxLimit) {
throw invalidConfig('defaultLimit must be less than or equal to maxLimit')
}
const journalModes: readonly string[] = ['wal', 'delete', 'truncate', 'persist']
if (!journalModes.includes(resolved.journalMode)) throw invalidConfig('journalMode is not supported')
return resolved
}
function assertPositiveInteger(name: string, value: number): void {
if (!Number.isInteger(value) || value < 1) throw invalidConfig(`${name} must be a positive integer`)
}
function invalidConfig(detail: string): SessionQueryError {
return new SessionQueryError(
`session-search SQLite config: ${detail}`,
'SESSION_QUERY_INVALID_CONFIG',
)
}
function indexClosed(): SessionQueryError {
return new SessionQueryError('session-search SQLite index is closed', 'SESSION_QUERY_INDEX_FAILED')
}
function assertNotAborted(signal: AbortSignal | undefined): void {
if (signal?.aborted) {
throw new SessionQueryError('session-search aborted', 'SESSION_QUERY_ABORTED')
}
}
function waitWithAbort<T>(promise: Promise<T>, signal: AbortSignal | undefined): Promise<T> {
if (signal === undefined) return promise
if (signal.aborted) return Promise.reject(new SessionQueryError('session-search aborted', 'SESSION_QUERY_ABORTED'))
return new Promise<T>((resolve, reject) => {
const onAbort = () => {
reject(new SessionQueryError('session-search aborted', 'SESSION_QUERY_ABORTED'))
}
signal.addEventListener('abort', onAbort, { once: true })
promise.then(
(value) => {
signal.removeEventListener('abort', onAbort)
resolve(value)
},
(error: unknown) => {
signal.removeEventListener('abort', onAbort)
reject(asError(error))
},
)
})
}
function isAbort(error: unknown): boolean {
return error instanceof SessionQueryError && error.code === 'SESSION_QUERY_ABORTED'
}
function asError(error: unknown): Error {
return error instanceof Error
? error
: new Error('session-search dependency rejected with a non-Error value', { cause: error })
}
function errorMessage(error: unknown): string {
return error instanceof Error ? error.message : 'unknown error'
}
export default SessionSearchSqlite

View File

@@ -0,0 +1,312 @@
/** Request normalization, parameterized predicates, and result presentation. */
import {
SessionQueryError,
filterSessionEventDocuments,
filterSessionResults,
} from '@deepseek-ai/dsh-session-query'
import type {
SessionEventMetadataFilter,
SessionEventSearchRequest,
SessionResultFilter,
SessionSearchRequest,
} from '@deepseek-ai/dsh-session-query'
/** Limit defaults needed to normalize a search request. */
export interface QueryLimits {
/** Page size used when the request omits one. */
defaultLimit: number
/** Largest accepted page size. */
maxLimit: number
}
/** Normalized cross-session request. */
export interface NormalizedSessionRequest {
query: string
sessionFilters: readonly SessionResultFilter[]
eventFilters: readonly SessionEventMetadataFilter[]
limit: number
cursor?: string
}
/** Normalized within-session request. */
export interface NormalizedEventRequest {
sessionId: SessionEventSearchRequest['sessionId']
query: string
filters: readonly SessionEventMetadataFilter[]
limit: number
cursor?: string
}
/** Parameterized SQL predicate fragment. */
export interface SqlWhere {
/** SQL without the leading `WHERE`. */
sql: string
/** Bindings in placeholder order. */
params: Array<string | number>
}
/**
* Validate and canonicalize a cross-session request.
* @param request - caller-provided query, filters, limit, and cursor.
* @param limits - configured default and maximum page sizes.
* @returns normalized request with explicit arrays and limit.
*/
export function normalizeSessionRequest(
request: SessionSearchRequest,
limits: QueryLimits,
): NormalizedSessionRequest {
const sessionFilters = request.sessionFilters ?? []
const eventFilters = request.eventFilters ?? []
filterSessionResults([], sessionFilters)
filterSessionEventDocuments([], eventFilters)
return {
query: normalizeQuery(request.query),
sessionFilters,
eventFilters,
limit: normalizeLimit(request.limit, limits),
...request.cursor === undefined ? {} : { cursor: request.cursor },
}
}
/**
* Validate and canonicalize a within-session request.
* @param request - caller-provided target, query, filters, limit, and cursor.
* @param limits - configured default and maximum page sizes.
* @returns normalized request with an explicit filter array and limit.
*/
export function normalizeEventRequest(
request: SessionEventSearchRequest,
limits: QueryLimits,
): NormalizedEventRequest {
const filters = request.filters ?? []
filterSessionEventDocuments([], filters)
return {
sessionId: request.sessionId,
query: normalizeQuery(request.query),
filters,
limit: normalizeLimit(request.limit, limits),
...request.cursor === undefined ? {} : { cursor: request.cursor },
}
}
/**
* Compile logical-session predicates against selected-document columns.
* @param filters - validated ANDed logical-session clauses.
* @returns parameterized SQL fragment and ordered bindings.
*/
export function buildSessionWhere(filters: readonly SessionResultFilter[]): SqlWhere {
const clauses: string[] = []
const params: Array<string | number> = []
for (const filter of filters) {
switch (filter.kind) {
case 'id':
addList(clauses, params, 'session_id', filter.values)
break
case 'cwd':
addNullableList(clauses, params, 'cwd', filter.values)
break
case 'created-at':
addRange(clauses, params, 'created_at', filter)
break
case 'parent':
addNullableList(clauses, params, 'parent_session', filter.values)
break
case 'availability': {
const availability = [...new Set(filter.values)]
if (availability.length === 0) clauses.push('0')
else if (availability.length === 1) clauses.push(`${availability[0]} = 1`)
break
}
}
}
return { sql: clauses.join(' AND '), params }
}
/**
* Compile event metadata predicates against selected-document columns.
* @param filters - validated ANDed event metadata clauses.
* @returns parameterized SQL fragment and ordered bindings.
*/
export function buildEventWhere(filters: readonly SessionEventMetadataFilter[]): SqlWhere {
const clauses: string[] = []
const params: Array<string | number> = []
for (const filter of filters) {
switch (filter.kind) {
case 'seq':
addRange(clauses, params, 'seq', filter)
break
case 'time':
addRange(clauses, params, 'time', filter)
break
case 'type':
addList(clauses, params, 'type', filter.values)
break
case 'surface':
addList(clauses, params, 'surface', filter.values)
break
}
}
return { sql: clauses.join(' AND '), params }
}
/**
* Quote caller text as one FTS5 phrase so query syntax remains inert data.
* @param query - normalized caller query.
* @returns FTS5 expression containing one escaped literal phrase.
*/
export function quoteFtsData(query: string): string {
return `"${query.replaceAll('"', '""')}"`
}
/**
* Build the stable normalized request identity stored in opaque cursors.
* @param request - normalized request whose filter ordering is canonicalized.
* @returns deterministic JSON identity for cursor binding.
*/
export function requestFingerprint(request: NormalizedSessionRequest | NormalizedEventRequest): string {
if ('sessionId' in request) {
return JSON.stringify({
scope: 'events',
sessionId: request.sessionId,
query: request.query,
filters: canonicalFilters(request.filters),
limit: request.limit,
})
}
return JSON.stringify({
scope: 'sessions',
query: request.query,
sessionFilters: canonicalFilters(request.sessionFilters),
eventFilters: canonicalFilters(request.eventFilters),
limit: request.limit,
})
}
/**
* Build a whitespace-normalized excerpt no longer than `maxChars`.
* @param text - complete extracted semantic document.
* @param query - normalized literal query used to position the excerpt.
* @param maxChars - maximum result length in Unicode code points.
* @returns bounded plain-text snippet.
*/
export function makeSnippet(text: string, query: string, maxChars: number): string {
const clean = text.replace(/\s+/gu, ' ').trim()
const characters = Array.from(clean)
if (characters.length <= maxChars) return clean
if (maxChars === 1) return '…'
const foundUnits = clean.toLowerCase().indexOf(query.toLowerCase())
const found = foundUnits < 0 ? -1 : Array.from(clean.slice(0, foundUnits)).length
let start = found < 0 ? 0 : Math.max(0, found - Math.floor(maxChars / 3))
let prefix = start > 0 ? '…' : ''
let suffix = '…'
let contentLength = maxChars - prefix.length - suffix.length
if (contentLength < 1) {
start = 0
prefix = ''
contentLength = maxChars - 1
}
let end = Math.min(characters.length, start + contentLength)
if (end === characters.length) {
suffix = ''
contentLength = maxChars - prefix.length
start = Math.max(0, end - contentLength)
}
end = Math.min(characters.length, start + contentLength)
return `${prefix}${characters.slice(start, end).join('')}${suffix}`
}
function normalizeQuery(value: string): string {
if (typeof value !== 'string') {
throw new SessionQueryError('session-search query must be text', 'SESSION_QUERY_INVALID_QUERY')
}
const query = value.trim().replace(/\s+/gu, ' ')
if (query.length === 0) {
throw new SessionQueryError(
'session-search query must contain non-whitespace text',
'SESSION_QUERY_INVALID_QUERY',
)
}
return query
}
function normalizeLimit(value: number | undefined, limits: QueryLimits): number {
const limit = value ?? limits.defaultLimit
if (!Number.isInteger(limit) || limit < 1 || limit > limits.maxLimit) {
throw new SessionQueryError(
`session-search limit must be an integer between 1 and ${limits.maxLimit}`,
'SESSION_QUERY_INVALID_LIMIT',
)
}
return limit
}
function addList(
clauses: string[],
params: Array<string | number>,
column: string,
values: readonly (string | number)[],
): void {
if (values.length === 0) {
clauses.push('0')
return
}
clauses.push(`${column} IN (${values.map(() => '?').join(', ')})`)
params.push(...values)
}
function addNullableList(
clauses: string[],
params: Array<string | number>,
column: string,
values: readonly (string | null)[],
): void {
if (values.length === 0) {
clauses.push('0')
return
}
const concrete = values.filter((value): value is string => value !== null)
const parts: string[] = []
if (concrete.length > 0) {
parts.push(`${column} IN (${concrete.map(() => '?').join(', ')})`)
params.push(...concrete)
}
if (values.includes(null)) parts.push(`${column} IS NULL`)
clauses.push(`(${parts.join(' OR ')})`)
}
function addRange(
clauses: string[],
params: Array<string | number>,
column: string,
range: { from?: number; to?: number },
): void {
if (range.from !== undefined) {
clauses.push(`CAST(${column} AS INTEGER) >= ?`)
params.push(range.from)
}
if (range.to !== undefined) {
clauses.push(`CAST(${column} AS INTEGER) <= ?`)
params.push(range.to)
}
}
function canonicalFilters(filters: readonly (SessionResultFilter | SessionEventMetadataFilter)[]): unknown[] {
return filters.map((filter) => {
if ('values' in filter) {
return { ...filter, values: [...filter.values].sort(compareNullable) }
}
return {
kind: filter.kind,
from: filter.from ?? null,
to: filter.to ?? null,
}
}).sort((a, b) => JSON.stringify(a).localeCompare(JSON.stringify(b)))
}
function compareNullable(a: string | null, b: string | null): number {
if (a === b) return 0
if (a === null) return -1
if (b === null) return 1
return a.localeCompare(b)
}

View File

@@ -0,0 +1,127 @@
/** SQLite schema for the disposable session full-text read model. */
import { DatabaseSync } from 'node:sqlite'
import { mkdir } from 'node:fs/promises'
import { dirname, resolve } from 'node:path'
/** Current derived-index schema version. Incompatible versions reset in place. */
export const SESSION_QUERY_SQLITE_SCHEMA_VERSION = 1
/** SQLite application id protecting unrelated databases from derived resets. */
export const SESSION_QUERY_SQLITE_APPLICATION_ID = 0x44534851
/** Supported SQLite journal modes. */
export type JournalMode = 'wal' | 'delete' | 'truncate' | 'persist'
/**
* Open, validate, and initialize persistent and connection-local schemas.
* @param path - dedicated derived-index path or `:memory:`.
* @param journalMode - validated SQLite journal mode.
* @returns initialized database handle owned by the search service.
*/
export async function openSearchDatabase(path: string, journalMode: JournalMode): Promise<DatabaseSync> {
const actual = path === ':memory:' ? path : resolve(path)
if (actual !== ':memory:') await mkdir(dirname(actual), { recursive: true, mode: 0o700 })
const db = new DatabaseSync(actual)
try {
// journalMode is a validated closed union, not caller-controlled SQL.
db.exec(`PRAGMA journal_mode = ${journalMode.toUpperCase()}`)
const { application_id: applicationId } = db.prepare('PRAGMA application_id').get() as { application_id: number }
const { user_version: version } = db.prepare('PRAGMA user_version').get() as { user_version: number }
const userTables = listUserTables(db)
if (applicationId !== 0 && applicationId !== SESSION_QUERY_SQLITE_APPLICATION_ID) {
throw new Error(`session-search database at "${actual}" belongs to another application`)
}
if (applicationId === 0 && userTables.length > 0) {
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)
}
ensurePersistentSchema(db)
ensureTemporarySchema(db)
return db
} catch (error: unknown) {
db.close()
throw error
}
}
function listUserTables(db: DatabaseSync): string[] {
const rows = db.prepare(
"SELECT name FROM sqlite_master WHERE type = 'table' AND name NOT LIKE 'sqlite_%' ORDER BY name",
).all() as Array<{ name: string }>
return rows.map(row => row.name)
}
function resetDerivedSchema(db: DatabaseSync): void {
for (const name of listUserTables(db)) {
db.exec(`DROP TABLE IF EXISTS ${quoteIdentifier(name)}`)
}
db.exec('PRAGMA user_version = 0')
}
function ensurePersistentSchema(db: DatabaseSync): void {
db.exec(`PRAGMA application_id = ${SESSION_QUERY_SQLITE_APPLICATION_ID}`)
db.exec(`
CREATE TABLE IF NOT EXISTS search_state (
singleton INTEGER PRIMARY KEY CHECK (singleton = 1),
global_generation INTEGER NOT NULL
) STRICT
`)
db.exec('INSERT OR IGNORE INTO search_state (singleton, global_generation) VALUES (1, 0)')
db.exec(`
CREATE TABLE IF NOT EXISTS persisted_sessions (
id TEXT PRIMARY KEY,
version INTEGER NOT NULL,
created_at INTEGER NOT NULL,
cwd TEXT,
parent_session TEXT,
seed_length INTEGER,
fingerprint TEXT NOT NULL,
generation INTEGER NOT NULL
) STRICT
`)
db.exec(`
CREATE VIRTUAL TABLE IF NOT EXISTS persisted_docs USING fts5(
text,
session_id UNINDEXED,
seq UNINDEXED,
type UNINDEXED,
time UNINDEXED,
surface UNINDEXED,
tokenize = 'unicode61'
)
`)
db.exec(`PRAGMA user_version = ${SESSION_QUERY_SQLITE_SCHEMA_VERSION}`)
}
function ensureTemporarySchema(db: DatabaseSync): void {
db.exec(`
CREATE TEMP TABLE IF NOT EXISTS live_sessions (
id TEXT PRIMARY KEY,
version INTEGER NOT NULL,
created_at INTEGER NOT NULL,
cwd TEXT,
parent_session TEXT,
seed_length INTEGER,
fingerprint TEXT NOT NULL,
generation INTEGER NOT NULL
) STRICT
`)
db.exec(`
CREATE VIRTUAL TABLE IF NOT EXISTS temp.live_docs USING fts5(
text,
session_id UNINDEXED,
seq UNINDEXED,
type UNINDEXED,
time UNINDEXED,
surface UNINDEXED,
tokenize = 'unicode61'
)
`)
}
function quoteIdentifier(value: string): string {
return `"${value.replaceAll('"', '""')}"`
}

View File

@@ -0,0 +1,60 @@
/**
* Keyless real-Loader-path smoke for the SQLite session-search service.
*
* @module @deepseek-ai/dsh-session-query-sqlite/tests/load-path
*/
import { afterEach, describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import Loader from '@cordisjs/plugin-loader'
import { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
import SessionStore from '@deepseek-ai/dsh-session'
import SessionPersistenceSqlite from '@deepseek-ai/dsh-session-persistence-sqlite'
import SessionSearchSqlite, * as searchModule from '@deepseek-ai/dsh-session-query-sqlite'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
const temporaryDirectories: string[] = []
afterEach(async () => {
for (const directory of temporaryDirectories.splice(0)) {
await rm(directory, { recursive: true, force: true })
}
})
async function temporaryPath(name: string): Promise<string> {
const directory = await mkdtemp(join(tmpdir(), 'dsh-session-search-loader-'))
temporaryDirectories.push(directory)
return join(directory, name)
}
describe('dsh-session-query-sqlite real Loader path', () => {
it('unwraps, mounts, and searches the real persistence backend', async () => {
const persistencePath = await temporaryPath('canonical.db')
const searchPath = await temporaryPath('derived.db')
const ctx = new Context()
await ctx.plugin(SessionStore)
const persistence = await ctx.plugin(SessionPersistenceSqlite, { path: persistencePath })
const loader = Object.create(Loader.prototype) as Loader
const unwrapped = loader.unwrapExports(searchModule) as Parameters<Context['plugin']>[0]
expect(unwrapped).toBe(SessionSearchSqlite)
const search = await ctx.plugin(unwrapped, { path: searchPath })
const id = SessionId('loader-path')
await ctx.sessionPersistence.create({ version: SESSION_FORMAT_VERSION, id, createdAt: 10 })
await ctx.sessionPersistence.append(id, [{
type: 'user/message',
seq: 0,
time: 10,
data: { content: [{ type: 'text', text: 'real Loader needle' }], source: { kind: 'user' } },
surfaceOp: 'append',
}])
await expect(ctx.sessionSearch.searchSessions({ query: 'Loader needle' }))
.resolves.toMatchObject({ items: [{ header: { id }, persisted: true, live: false }] })
await search.dispose()
await persistence.dispose()
})
})

View File

@@ -0,0 +1,179 @@
import { describe, expect, it } from 'vitest'
import { SessionId } from '@deepseek-ai/dsh-session'
import type { SessionQueryErrorCode } from '@deepseek-ai/dsh-session-query'
import {
buildEventWhere,
buildSessionWhere,
makeSnippet,
normalizeEventRequest,
normalizeSessionRequest,
quoteFtsData,
requestFingerprint,
type NormalizedEventRequest,
type NormalizedSessionRequest,
} from '../src/query.ts'
const limits = { defaultLimit: 2, maxLimit: 3 }
function expectCode(code: SessionQueryErrorCode): Error {
return expect.objectContaining({ code }) as Error
}
describe('SQLite search request normalization', () => {
it('normalizes both scopes, defaults arrays and limits, and preserves cursors', () => {
expect(normalizeSessionRequest({ query: ' alpha\n beta ' }, limits)).toEqual({
query: 'alpha beta',
sessionFilters: [],
eventFilters: [],
limit: 2,
})
expect(normalizeSessionRequest({
query: 'needle',
sessionFilters: [{ kind: 'availability', values: ['live'] }],
eventFilters: [{ kind: 'surface', values: ['current'] }],
limit: 3,
cursor: 'next',
}, limits)).toEqual({
query: 'needle',
sessionFilters: [{ kind: 'availability', values: ['live'] }],
eventFilters: [{ kind: 'surface', values: ['current'] }],
limit: 3,
cursor: 'next',
})
expect(normalizeEventRequest({ sessionId: SessionId('s'), query: 'needle' }, limits)).toEqual({
sessionId: SessionId('s'),
query: 'needle',
filters: [],
limit: 2,
})
expect(normalizeEventRequest({
sessionId: SessionId('s'),
query: 'needle',
filters: [{ kind: 'seq', from: 1 }],
cursor: 'next',
}, limits)).toEqual({
sessionId: SessionId('s'),
query: 'needle',
filters: [{ kind: 'seq', from: 1 }],
limit: 2,
cursor: 'next',
})
})
it('rejects non-text, blank, non-integer, non-positive, and oversized requests', () => {
expect(() => normalizeSessionRequest({ query: 1 as never }, limits))
.toThrow(expectCode('SESSION_QUERY_INVALID_QUERY'))
expect(() => normalizeSessionRequest({ query: ' \n ' }, limits))
.toThrow(expectCode('SESSION_QUERY_INVALID_QUERY'))
for (const limit of [1.5, 0, 4]) {
expect(() => normalizeEventRequest({ sessionId: SessionId('s'), query: 'x', limit }, limits))
.toThrow(expectCode('SESSION_QUERY_INVALID_LIMIT'))
}
})
})
describe('SQLite search predicate compilation', () => {
it('compiles all logical-session clauses including empty and nullable values', () => {
expect(buildSessionWhere([])).toEqual({ sql: '', params: [] })
expect(buildSessionWhere([{ kind: 'id', values: [] }])).toEqual({ sql: '0', params: [] })
expect(buildSessionWhere([{ kind: 'id', values: [SessionId('a'), SessionId('b')] }])).toEqual({
sql: 'session_id IN (?, ?)',
params: [SessionId('a'), SessionId('b')],
})
expect(buildSessionWhere([{ kind: 'cwd', values: [] }])).toEqual({ sql: '0', params: [] })
expect(buildSessionWhere([{ kind: 'cwd', values: [null] }])).toEqual({
sql: '(cwd IS NULL)',
params: [],
})
expect(buildSessionWhere([{ kind: 'cwd', values: ['/a'] }])).toEqual({
sql: '(cwd IN (?))',
params: ['/a'],
})
expect(buildSessionWhere([{ kind: 'parent', values: [SessionId('p'), null] }])).toEqual({
sql: '(parent_session IN (?) OR parent_session IS NULL)',
params: [SessionId('p')],
})
expect(buildSessionWhere([
{ kind: 'created-at', from: 1, to: 2 },
{ kind: 'availability', values: [] },
{ kind: 'availability', values: ['live', 'live'] },
{ kind: 'availability', values: ['live', 'persisted'] },
])).toEqual({
sql: 'CAST(created_at AS INTEGER) >= ? AND CAST(created_at AS INTEGER) <= ? AND 0 AND live = 1',
params: [1, 2],
})
expect(buildSessionWhere([{ kind: 'created-at' }])).toEqual({ sql: '', params: [] })
})
it('compiles every event clause and empty lists', () => {
expect(buildEventWhere([
{ kind: 'seq', from: 1 },
{ kind: 'time', to: 9 },
{ kind: 'type', values: ['user/message'] },
{ kind: 'surface', values: ['current', 'log-only'] },
])).toEqual({
sql: 'CAST(seq AS INTEGER) >= ? AND CAST(time AS INTEGER) <= ? AND type IN (?) AND surface IN (?, ?)',
params: [1, 9, 'user/message', 'current', 'log-only'],
})
expect(buildEventWhere([
{ kind: 'type', values: [] },
{ kind: 'surface', values: [] },
])).toEqual({ sql: '0 AND 0', params: [] })
})
})
describe('SQLite query identity and presentation', () => {
it('quotes all caller MATCH syntax as data', () => {
expect(quoteFtsData('say "needle" OR *')).toBe('"say ""needle"" OR *"')
})
it('canonicalizes request and filter ordering in both scopes', () => {
const sessionA: NormalizedSessionRequest = {
query: 'needle',
limit: 2,
sessionFilters: [
{ kind: 'cwd', values: ['/b', '/a'] },
{ kind: 'parent', values: [null, SessionId('p')] },
{ kind: 'id', values: [SessionId('same'), SessionId('same')] },
{ kind: 'created-at', from: 1 },
],
eventFilters: [{ kind: 'time', to: 9 }],
}
const sessionB: NormalizedSessionRequest = {
query: 'needle',
limit: 2,
sessionFilters: [
{ kind: 'created-at', from: 1 },
{ kind: 'id', values: [SessionId('same'), SessionId('same')] },
{ kind: 'parent', values: [SessionId('p'), null] },
{ kind: 'cwd', values: ['/a', '/b'] },
],
eventFilters: [{ kind: 'time', to: 9 }],
}
expect(requestFingerprint(sessionA)).toBe(requestFingerprint(sessionB))
const eventA: NormalizedEventRequest = {
sessionId: SessionId('s'),
query: 'needle',
limit: 2,
filters: [{ kind: 'seq' }, { kind: 'surface', values: ['shadowed', 'current'] }],
}
const eventB: NormalizedEventRequest = {
sessionId: SessionId('s'),
query: 'needle',
limit: 2,
filters: [{ kind: 'surface', values: ['current', 'shadowed'] }, { kind: 'seq' }],
}
expect(requestFingerprint(eventA)).toBe(requestFingerprint(eventB))
expect(requestFingerprint(eventA)).not.toBe(requestFingerprint({ ...eventB, sessionId: SessionId('other') }))
})
it('normalizes, bounds, and positions snippets by Unicode code point', () => {
expect(makeSnippet(' short\ntext ', 'absent', 20)).toBe('short text')
expect(makeSnippet('abcdef', 'f', 1)).toBe('…')
expect(makeSnippet('abcdefghij', 'absent', 5)).toBe('abcd…')
expect(makeSnippet('abcdefghij', 'c', 5)).toBe('…bcd…')
expect(makeSnippet('abcdef', 'f', 2)).toBe('a…')
expect(makeSnippet('abcdef', 'f', 5)).toBe('…cdef')
})
})

View File

@@ -0,0 +1,594 @@
import { afterEach, describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { DatabaseSync } from 'node:sqlite'
import { mkdtemp, rm } from 'node:fs/promises'
import { tmpdir } from 'node:os'
import { join } from 'node:path'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader, SessionId as SessionIdType } from '@deepseek-ai/dsh-session'
import SessionPersistence from '@deepseek-ai/dsh-session-persistence'
import SessionPersistenceSqlite from '@deepseek-ai/dsh-session-persistence-sqlite'
import SessionSearchSqlite, {
SESSION_QUERY_SQLITE_APPLICATION_ID,
SESSION_QUERY_SQLITE_SCHEMA_VERSION,
} from '@deepseek-ai/dsh-session-query-sqlite'
import type { SessionQueryErrorCode } from '@deepseek-ai/dsh-session-query'
const temporaryDirectories: string[] = []
afterEach(async () => {
for (const directory of temporaryDirectories.splice(0)) {
await rm(directory, { recursive: true, force: true })
}
})
async function temporaryPath(name = 'search.db'): Promise<string> {
const directory = await mkdtemp(join(tmpdir(), 'dsh-session-search-'))
temporaryDirectories.push(directory)
return join(directory, name)
}
function header(id: string, createdAt = 1, extra: Partial<SessionHeader> = {}): SessionHeader {
return { version: SESSION_FORMAT_VERSION, id: SessionId(id), createdAt, ...extra }
}
function messageEvents(text: string, time = 1): SessionEvent[] {
return [{
type: 'user/message',
seq: 0,
time,
data: { content: [{ type: 'text', text }], source: { kind: 'user' } },
surfaceOp: 'append',
}]
}
function expectCode(code: SessionQueryErrorCode): Error {
return expect.objectContaining({ code }) as Error
}
class TestPersistence extends SessionPersistence {
static entries = new Map<SessionIdType, { meta: SessionHeader; events: SessionEvent[] }>()
static listGate: Promise<void> | undefined
static listStarted: (() => void) | undefined
static failure: unknown
static reset(entries: readonly { meta: SessionHeader; events: SessionEvent[] }[] = []): void {
this.entries = new Map(entries.map(entry => [entry.meta.id, structuredClone(entry)]))
this.listGate = undefined
this.listStarted = undefined
this.failure = undefined
}
create(meta: SessionHeader): Promise<void> {
TestPersistence.entries.set(meta.id, { meta: structuredClone(meta), events: [] })
return Promise.resolve()
}
append(id: SessionIdType, events: readonly SessionEvent[]): Promise<void> {
const entry = TestPersistence.entries.get(id)
if (entry === undefined) return Promise.reject(new Error('missing test session'))
entry.events.push(...structuredClone(events))
return Promise.resolve()
}
async load(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
if (TestPersistence.failure !== undefined) throw TestPersistence.failure
const entry = TestPersistence.entries.get(id)
if (entry === undefined) throw new Error('missing test session')
return structuredClone(entry)
}
async list(): Promise<SessionHeader[]> {
TestPersistence.listStarted?.()
await TestPersistence.listGate
if (TestPersistence.failure !== undefined) throw TestPersistence.failure
return [...TestPersistence.entries.values()].map(entry => structuredClone(entry.meta))
}
}
async function liveContext(config: ConstructorParameters<typeof SessionSearchSqlite>[1] = { path: ':memory:' }): Promise<Context> {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionSearchSqlite, config)
return ctx
}
describe('SQLite session search', () => {
it('searches two-character Unicode61 tokens in live-only sessions', async () => {
const ctx = await liveContext({ path: ':memory:', snippetChars: 20 })
const session = ctx.sessions.create(SessionId('live'), { meta: { cwd: '/work', createdAt: 10, seedLength: 1 } })
session.append(
'user/message',
{ content: [{ type: 'text', text: 'An AI helper' }], source: { kind: 'user' } },
{ surfaceOp: 'append' },
)
await expect(ctx.sessionSearch.searchEvents({ sessionId: session.id, query: 'AI' }))
.resolves.toMatchObject({ items: [{ sessionId: session.id, seq: 0, snippet: 'An AI helper' }] })
await expect(ctx.sessionSearch.searchSessions({ query: 'AI' }))
.resolves.toMatchObject({ items: [{ header: { ...session.header, seedLength: 1 }, live: true, persisted: false }] })
})
it('searches all surfaces by default and applies metadata before ranking', async () => {
const ctx = await liveContext({ path: ':memory:', defaultLimit: 10, maxLimit: 20 })
const parent = SessionId('parent')
const events: SessionEvent[] = [
{ type: 'user/message', seq: 0, time: 10, data: { content: [{ type: 'text', text: 'needle original' }], source: { kind: 'user' } }, surfaceOp: 'append' },
{ type: 'assistant/chunk', seq: 1, time: 11, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'needle raw' } } },
{ type: 'user/message', seq: 2, time: 12, data: { content: [{ type: 'text', text: 'needle summary' }], source: { kind: 'plugin', plugin: 'test' } }, surfaceOp: { op: 'replace', start: 0, end: 0 } },
{ type: 'turn/end', seq: 3, time: 13, data: { turn: 1, reason: { kind: 'error', step: 1, message: 'needle failure' } } },
]
ctx.sessions.create(SessionId('a'), { seed: events, meta: { cwd: '/a', parentSession: parent, createdAt: 20 } })
ctx.sessions.create(SessionId('b'), { seed: messageEvents('needle peer', 12), meta: { createdAt: 20 } })
const all = await ctx.sessionSearch.searchEvents({ sessionId: SessionId('a'), query: 'needle' })
expect(new Set(all.items.map(item => item.surface))).toEqual(new Set(['current', 'shadowed', 'log-only']))
await expect(ctx.sessionSearch.searchEvents({
sessionId: SessionId('a'),
query: 'needle',
filters: [
{ kind: 'seq', from: 2, to: 2 },
{ kind: 'time', from: 12, to: 12 },
{ kind: 'type', values: ['user/message'] },
{ kind: 'surface', values: ['current'] },
],
})).resolves.toMatchObject({ items: [{ seq: 2, surface: 'current' }] })
const grouped = await ctx.sessionSearch.searchSessions({
query: 'needle',
sessionFilters: [
{ kind: 'id', values: [SessionId('a')] },
{ kind: 'cwd', values: ['/a'] },
{ kind: 'created-at', from: 20, to: 20 },
{ kind: 'parent', values: [parent] },
{ kind: 'availability', values: ['live'] },
],
eventFilters: [{ kind: 'surface', values: ['shadowed'] }],
})
expect(grouped.items).toHaveLength(1)
expect(grouped.items[0]).toMatchObject({
header: { id: SessionId('a'), cwd: '/a', parentSession: parent },
live: true,
persisted: false,
bestMatch: { seq: 0, surface: 'shadowed' },
})
})
it('uses literal phrase tokens, stable ties, and bounded Unicode snippets', async () => {
const ctx = await liveContext({ path: ':memory:', defaultLimit: 10, maxLimit: 10, snippetChars: 5 })
ctx.sessions.create(SessionId('a'), { seed: messageEvents('😀😀 alpha beta BRAID 😀😀', 10), meta: { createdAt: 1 } })
ctx.sessions.create(SessionId('b'), { seed: messageEvents('alpha beta', 10), meta: { createdAt: 1 } })
ctx.sessions.create(SessionId('c'), { seed: messageEvents('alpha middle beta', 10), meta: { createdAt: 1 } })
ctx.sessions.create(SessionId('d'), { seed: messageEvents('alpha beta', 10), meta: { createdAt: 1 } })
ctx.sessions.create(SessionId('operator'), { seed: messageEvents('needle OR absent', 10), meta: { createdAt: 1 } })
ctx.sessions.create(SessionId('only'), { seed: messageEvents('needle only', 10), meta: { createdAt: 1 } })
ctx.sessions.create(SessionId('quote'), { seed: messageEvents('say "needle" exactly', 10), meta: { createdAt: 1 } })
const phrase = await ctx.sessionSearch.searchSessions({ query: 'alpha beta' })
expect(phrase.items.map(item => item.header.id)).toEqual([SessionId('b'), SessionId('d'), SessionId('a')])
expect(phrase.items.every(item => Array.from(item.bestMatch.snippet).length <= 5)).toBe(true)
await expect(ctx.sessionSearch.searchSessions({ query: 'AI' })).resolves.toEqual({ items: [] })
await expect(ctx.sessionSearch.searchSessions({ query: 'needle OR absent' }))
.resolves.toMatchObject({ items: [{ header: { id: SessionId('operator') } }] })
await expect(ctx.sessionSearch.searchSessions({ query: 'say "needle"' }))
.resolves.toMatchObject({ items: [{ header: { id: SessionId('quote') } }] })
await expect(ctx.sessionSearch.searchSessions({ query: '*' })).resolves.toEqual({ items: [] })
})
it('binds cursors to requests and only invalidates within-session pages for target changes', async () => {
const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 5 })
const target = ctx.sessions.create(SessionId('target'), {
seed: [
...messageEvents('needle one', 10),
{ ...messageEvents('needle two', 11)[0]!, seq: 1 },
{ ...messageEvents('needle three', 12)[0]!, seq: 2 },
],
})
ctx.sessions.create(SessionId('other'), { seed: messageEvents('needle other', 10) })
const eventPage = await ctx.sessionSearch.searchEvents({ sessionId: target.id, query: 'needle', limit: 1 })
const sessionPage = await ctx.sessionSearch.searchSessions({ query: 'needle', limit: 1 })
expect(eventPage.nextCursor).toEqual(expect.any(String))
expect(sessionPage.nextCursor).toEqual(expect.any(String))
if (eventPage.nextCursor === undefined || sessionPage.nextCursor === undefined) throw new Error('expected cursors')
const eventKeys = eventPage.items.map(item => `${item.sessionId}:${item.seq}`)
let eventCursor: string | undefined = eventPage.nextCursor
while (eventCursor !== undefined) {
const next = await ctx.sessionSearch.searchEvents({
sessionId: target.id,
query: 'needle',
limit: 1,
cursor: eventCursor,
})
eventKeys.push(...next.items.map(item => `${item.sessionId}:${item.seq}`))
eventCursor = next.nextCursor
}
expect(eventKeys).toHaveLength(3)
expect(new Set(eventKeys).size).toBe(eventKeys.length)
const sessionIds = sessionPage.items.map(item => item.header.id)
let sessionCursor: string | undefined = sessionPage.nextCursor
while (sessionCursor !== undefined) {
const next = await ctx.sessionSearch.searchSessions({ query: 'needle', limit: 1, cursor: sessionCursor })
sessionIds.push(...next.items.map(item => item.header.id))
sessionCursor = next.nextCursor
}
expect(sessionIds).toHaveLength(2)
expect(new Set(sessionIds).size).toBe(sessionIds.length)
ctx.sessions.create(SessionId('unrelated'), { seed: messageEvents('needle unrelated', 20) })
await expect(ctx.sessionSearch.searchEvents({
sessionId: target.id,
query: 'needle',
limit: 1,
cursor: eventPage.nextCursor,
})).resolves.toMatchObject({ items: [{ sessionId: target.id }] })
await expect(ctx.sessionSearch.searchSessions({ query: 'needle', limit: 1, cursor: sessionPage.nextCursor }))
.rejects.toThrow(expectCode('SESSION_QUERY_STALE_CURSOR'))
await expect(ctx.sessionSearch.searchEvents({
sessionId: target.id,
query: 'different',
limit: 1,
cursor: eventPage.nextCursor,
})).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CURSOR'))
target.append('user/message', { content: [{ type: 'text', text: 'needle four' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
await expect(ctx.sessionSearch.searchEvents({
sessionId: target.id,
query: 'needle',
limit: 1,
cursor: eventPage.nextCursor,
})).rejects.toThrow(expectCode('SESSION_QUERY_STALE_CURSOR'))
})
it('rejects invalid requests, filters, cursors, and direct config', async () => {
const ctx = await liveContext({ path: ':memory:', defaultLimit: 2, maxLimit: 3 })
const session = ctx.sessions.create(SessionId('valid'), { seed: messageEvents('needle') })
for (const request of [
{ sessionId: session.id, query: '' },
{ sessionId: session.id, query: 'needle', limit: 0 },
{ sessionId: session.id, query: 'needle', limit: 4 },
{ sessionId: session.id, query: 'needle', filters: [{ kind: 'seq', from: 2, to: 1 }] },
{ sessionId: session.id, query: 'needle', filters: [{ kind: 'surface', values: ['future'] }] },
] as const) {
await expect(ctx.sessionSearch.searchEvents(request as never)).rejects.toBeInstanceOf(Error)
}
await expect(ctx.sessionSearch.searchSessions({
query: 'needle',
sessionFilters: [{ kind: 'availability', values: ['remote' as never] }],
})).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
await expect(ctx.sessionSearch.searchEvents({ sessionId: session.id, query: 'needle', cursor: 'not-json' }))
.rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CURSOR'))
await expect(ctx.sessionSearch.searchEvents({ sessionId: SessionId('absent'), query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
for (const config of [
{ path: '' },
{ path: ':memory:', defaultLimit: 0 },
{ path: ':memory:', maxLimit: 0 },
{ path: ':memory:', snippetChars: 0 },
{ path: ':memory:', defaultLimit: 3, maxLimit: 2 },
{ path: ':memory:', journalMode: 'memory' },
]) {
const direct = new Context()
await direct.plugin(SessionStore)
expect(() => new SessionSearchSqlite(direct, config as never))
.toThrow(expectCode('SESSION_QUERY_INVALID_CONFIG'))
}
})
})
describe('SQLite reconciliation and source lifecycle', () => {
it('mounts persistence dynamically, shadows with TEMP live rows, reveals, and hides on unmount', async () => {
const shared = header('shared', 10, { cwd: '/work' })
const durable = header('durable', 5)
TestPersistence.reset([
{ meta: shared, events: messageEvents('persisted needle') },
{ meta: durable, events: messageEvents('durable needle') },
])
const ctx = await liveContext()
await expect(ctx.sessionSearch.searchSessions({ query: 'durable' })).resolves.toEqual({ items: [] })
const persistenceFiber = await ctx.plugin(TestPersistence)
await expect(ctx.sessionSearch.searchSessions({ query: 'durable' }))
.resolves.toMatchObject({ items: [{ header: durable, live: false, persisted: true }] })
const live = ctx.sessions.prepare(shared.id, { meta: { createdAt: 10, cwd: '/work' } })
live.append('user/message', { content: [{ type: 'text', text: 'live needle' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
const detach = ctx.sessions.enter(live)
ctx.sessions.announce(live)
await expect(ctx.sessionSearch.searchSessions({ query: 'persisted' })).resolves.toEqual({ items: [] })
await expect(ctx.sessionSearch.searchSessions({ query: 'live' }))
.resolves.toMatchObject({ items: [{ header: shared, live: true, persisted: true }] })
detach()
await expect(ctx.sessionSearch.searchSessions({ query: 'persisted' }))
.resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
await persistenceFiber.dispose()
await expect(ctx.sessionSearch.searchSessions({ query: 'durable' })).resolves.toEqual({ items: [] })
await expect(ctx.sessionSearch.searchEvents({ sessionId: durable.id, query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
})
it('restarts observation when persistence unmounts during an asynchronous list', async () => {
const durable = header('racing')
TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
const ctx = await liveContext()
const persistenceFiber = await ctx.plugin(TestPersistence)
let release!: () => void
TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
let markStarted!: () => void
const started = new Promise<void>((resolve) => { markStarted = resolve })
TestPersistence.listStarted = () => {
TestPersistence.listStarted = undefined
markStarted()
}
const search = ctx.sessionSearch.searchSessions({ query: 'needle' })
await started
await persistenceFiber.dispose()
release()
await expect(search).resolves.toEqual({ items: [] })
})
it('rejects immutable header conflicts between live and persisted sources', async () => {
const shared = header('conflict', 10)
TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
const ctx = await liveContext()
await ctx.plugin(TestPersistence)
ctx.sessions.create(shared.id, { seed: messageEvents('live needle'), meta: { createdAt: 11 } })
await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_SOURCE_CONFLICT'))
})
it('preserves unchanged persisted generations while reconciling new, changed, and deleted rows', async () => {
const path = await temporaryPath()
const unchanged = header('unchanged')
const changed = header('changed')
const deleted = header('deleted')
TestPersistence.reset([
{ meta: unchanged, events: messageEvents('unchanged needle') },
{ meta: changed, events: messageEvents('old needle') },
{ meta: deleted, events: messageEvents('deleted needle') },
])
const first = new Context()
await first.plugin(SessionStore)
const firstPersistence = await first.plugin(TestPersistence)
const firstSearch = await first.plugin(SessionSearchSqlite, { path })
await first.sessionSearch.searchSessions({ query: 'needle' })
await firstSearch.dispose()
await firstPersistence.dispose()
const beforeDb = new DatabaseSync(path)
const beforeRows = beforeDb.prepare('SELECT id, generation FROM persisted_sessions ORDER BY id').all() as Array<{ id: string; generation: number }>
beforeDb.close()
const before = new Map(beforeRows.map(row => [row.id, row.generation]))
const added = header('added')
TestPersistence.entries.delete(deleted.id)
TestPersistence.entries.set(changed.id, { meta: changed, events: messageEvents('changed needle') })
TestPersistence.entries.set(added.id, { meta: added, events: messageEvents('added needle') })
const second = new Context()
await second.plugin(SessionStore)
const secondPersistence = await second.plugin(TestPersistence)
const secondSearch = await second.plugin(SessionSearchSqlite, { path })
const result = await second.sessionSearch.searchSessions({ query: 'needle' })
expect(result.items.map(item => item.header.id).sort()).toEqual([added.id, changed.id, unchanged.id].sort())
await secondSearch.dispose()
await secondPersistence.dispose()
const afterDb = new DatabaseSync(path)
const afterRows = afterDb.prepare('SELECT id, generation FROM persisted_sessions ORDER BY id').all() as Array<{ id: string; generation: number }>
afterDb.close()
const after = new Map(afterRows.map(row => [row.id, row.generation]))
expect(after.get(unchanged.id)).toBe(before.get(unchanged.id))
expect(after.get(changed.id)).toBeGreaterThan(before.get(changed.id)!)
expect(after.has(deleted.id)).toBe(false)
expect(after.has(added.id)).toBe(true)
})
it('drops connection-local live overlays on reopen and retains persistent bases', async () => {
const path = await temporaryPath()
const shared = header('shared', 10)
TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
const first = new Context()
await first.plugin(SessionStore)
const persistence = await first.plugin(TestPersistence)
const live = first.sessions.create(shared.id, { seed: messageEvents('live needle'), meta: { createdAt: 10 } })
const search = await first.plugin(SessionSearchSqlite, { path })
await expect(first.sessionSearch.searchEvents({ sessionId: live.id, query: 'live' })).resolves.toMatchObject({ items: [{}] })
await search.dispose()
await persistence.dispose()
const second = new Context()
await second.plugin(SessionStore)
const persistenceAgain = await second.plugin(TestPersistence)
const searchAgain = await second.plugin(SessionSearchSqlite, { path })
await expect(second.sessionSearch.searchSessions({ query: 'live' })).resolves.toEqual({ items: [] })
await expect(second.sessionSearch.searchSessions({ query: 'persisted' }))
.resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
await searchAgain.dispose()
await persistenceAgain.dispose()
})
it('recovers on the next search after source and SQLite transaction failures', async () => {
TestPersistence.reset([{ meta: header('durable'), events: messageEvents('durable needle') }])
const ctx = await liveContext()
await ctx.plugin(TestPersistence)
TestPersistence.failure = 'offline'
await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
const signal = new AbortController().signal
await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal }))
.rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
TestPersistence.failure = new Error('still offline')
await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal }))
.rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
TestPersistence.failure = undefined
await expect(ctx.sessionSearch.searchSessions({ query: 'needle' })).resolves.toMatchObject({ items: [{}] })
const live = ctx.sessions.create(SessionId('live'), { seed: messageEvents('base') })
await ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'base' })
const db = (ctx.sessionSearch as unknown as { _db: DatabaseSync })._db
db.exec('PRAGMA query_only = ON')
live.append('user/message', { content: [{ type: 'text', text: 'retry needle' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
await expect(ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
db.exec('PRAGMA query_only = OFF')
await expect(ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'needle' }))
.resolves.toMatchObject({ items: [{ seq: 1 }] })
})
})
describe('SQLite schema, cancellation, and real persistence integration', () => {
it('resets a recognized incompatible derived schema but refuses a foreign database', async () => {
const stalePath = await temporaryPath('stale.db')
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') })
await staleCtx.sessionSearch.searchSessions({ query: 'needle' })
await (staleCtx.sessionSearch as SessionSearchSqlite).close()
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 foreignPath = await temporaryPath('foreign.db')
const foreign = new DatabaseSync(foreignPath)
foreign.exec('CREATE TABLE canonical(value TEXT)')
foreign.exec("INSERT INTO canonical VALUES ('safe')")
foreign.close()
const foreignCtx = await liveContext({ path: foreignPath })
await expect(foreignCtx.sessionSearch.searchSessions({ query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
const stillForeign = new DatabaseSync(foreignPath)
expect(stillForeign.prepare('SELECT value FROM canonical').get()).toEqual({ value: 'safe' })
stillForeign.close()
const otherAppPath = await temporaryPath('other-app.db')
const otherApp = new DatabaseSync(otherAppPath)
otherApp.exec('PRAGMA application_id = 123')
otherApp.close()
const otherAppCtx = await liveContext({ path: otherAppPath })
await expect(otherAppCtx.sessionSearch.searchSessions({ query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
})
it('cancels both queued and in-flight source waits without committing them', async () => {
TestPersistence.reset()
const ctx = await liveContext()
await ctx.plugin(TestPersistence)
const boundaryController = new AbortController()
const boundary = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: boundaryController.signal })
queueMicrotask(() => { boundaryController.abort() })
await expect(boundary).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
const readyController = new AbortController()
readyController.abort()
const internals = ctx.sessionSearch as unknown as {
_ensureReady(signal: AbortSignal): Promise<void>
}
await expect(internals._ensureReady(readyController.signal))
.rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
let releaseBlocking!: () => void
TestPersistence.listGate = new Promise<void>((resolve) => { releaseBlocking = resolve })
let markBlockingStarted!: () => void
const blockingStarted = new Promise<void>((resolve) => { markBlockingStarted = resolve })
TestPersistence.listStarted = () => {
TestPersistence.listStarted = undefined
markBlockingStarted()
}
const blocking = ctx.sessionSearch.searchSessions({ query: 'needle' })
await blockingStarted
const queuedController = new AbortController()
const queued = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: queuedController.signal })
queuedController.abort()
await expect(queued).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
releaseBlocking()
await expect(blocking).resolves.toEqual({ items: [] })
TestPersistence.entries.set(SessionId('uncommitted'), {
meta: header('uncommitted'),
events: messageEvents('durable needle'),
})
let releaseActive!: () => void
TestPersistence.listGate = new Promise<void>((resolve) => { releaseActive = resolve })
let markActiveStarted!: () => void
const activeStarted = new Promise<void>((resolve) => { markActiveStarted = resolve })
TestPersistence.listStarted = () => {
TestPersistence.listStarted = undefined
markActiveStarted()
}
const activeController = new AbortController()
const active = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: activeController.signal })
await activeStarted
activeController.abort()
await expect(active).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
releaseActive()
const db = (ctx.sessionSearch as unknown as { _db: DatabaseSync })._db
expect(db.prepare('SELECT COUNT(*) AS count FROM persisted_sessions').get()).toEqual({ count: 0 })
await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
.resolves.toMatchObject({ items: [{ header: { id: SessionId('uncommitted') } }] })
})
it('rejects queued and future work when close waits for an accepted operation', async () => {
TestPersistence.reset()
let release!: () => void
TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
let markStarted!: () => void
const started = new Promise<void>((resolve) => { markStarted = resolve })
TestPersistence.listStarted = () => {
TestPersistence.listStarted = undefined
markStarted()
}
const ctx = await liveContext()
await ctx.plugin(TestPersistence)
const search = ctx.sessionSearch as SessionSearchSqlite
const accepted = search.searchSessions({ query: 'needle' })
await started
const queued = search.searchSessions({ query: 'needle' })
const closing = search.close()
release()
await expect(accepted).resolves.toEqual({ items: [] })
await expect(queued).rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
await closing
await expect(search.searchSessions({ query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
await search.close()
})
it('combines the real SQLite persistence backend with the real search service keylessly', async () => {
const persistencePath = await temporaryPath('canonical.db')
const searchPath = await temporaryPath('derived.db')
const ctx = new Context()
await ctx.plugin(SessionStore)
const persistence = await ctx.plugin(SessionPersistenceSqlite, { path: persistencePath })
const search = await ctx.plugin(SessionSearchSqlite, { path: searchPath })
const meta = header('real', 10, { cwd: '/work' })
await ctx.sessionPersistence.create(meta)
await ctx.sessionPersistence.append(meta.id, messageEvents('real SQLite needle'))
await expect(ctx.sessionSearch.searchSessions({ query: 'SQLite needle' }))
.resolves.toMatchObject({ items: [{ header: meta, persisted: true, live: false }] })
await expect(ctx.sessionSearch.searchEvents({ sessionId: meta.id, query: 'SQLite needle' }))
.resolves.toMatchObject({ items: [{ sessionId: meta.id, seq: 0 }] })
await expect(ctx.sessionSearch.searchEvents({ sessionId: SessionId('absent'), query: 'needle' }))
.rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
await search.dispose()
await expect(ctx.sessionPersistence.load(meta.id)).resolves.toMatchObject({ meta, events: [{ seq: 0 }] })
await persistence.dispose()
})
})

View File

@@ -0,0 +1,30 @@
{
"extends": "../../../tsconfig.base.json",
"compilerOptions": {
"rootDir": "src",
"outDir": "lib/types"
},
"include": [
"src"
],
"references": [
{
"path": "../../../vendor/cosmokit"
},
{
"path": "../../../vendor/cordis"
},
{
"path": "../../../vendor/schemastery"
},
{
"path": "../../core/session"
},
{
"path": "../../session-persistence/session-persistence"
},
{
"path": "../session-query"
}
]
}

View File

@@ -1,6 +1,6 @@
# @deepseek-ai/dsh-session-query
Exact session-history retrieval through `ctx.sessionQuery`. The service presents live `ctx.sessions` and an optional, dynamically mounted `ctx.sessionPersistence` as one logical corpus. Matching ids produce one record: live events win, while `live` and `persisted` report both source availabilities. Conflicting immutable headers fail with `SESSION_QUERY_SOURCE_CONFLICT`.
Session-history query contracts and provider-independent helpers. The concrete `ctx.sessionQuery` service presents live `ctx.sessions` and an optional, dynamically mounted `ctx.sessionPersistence` as one logical corpus for exact reads and semantic scans. The abstract `ctx.sessionSearch` service defines full-text search without introducing a provider registry.
This is trusted context-wide infrastructure. It performs no caller authorization; a future model tool or UI must constrain which sessions its caller may inspect.
@@ -8,11 +8,24 @@ This is trusted context-wide infrastructure. It performs no caller authorization
- `listSessions()` reads current persistence metadata, merges live records with live precedence, and returns cloned records in deterministic newest-first order.
- `listEvents(sessionId)` loads the live-preferred raw log and classifies each event as `current`, `shadowed`, or `log-only` with the shared `dsh-session` surface fold.
- `filterEvents(sessionId, filters)` extracts first-party semantic documents and applies provider-independent metadata and literal-text predicates in ascending seq order.
- `readEvent(request)` returns a cloned header, the full target event, and a bounded raw-seq window. `before` and `after` default to zero and may not exceed `readWindowMax`.
Persistence is optional and may mount or unmount dynamically. A cross-corpus list fails with `SESSION_QUERY_PERSISTENCE_FAILED` while mounted persistence is unreadable. A read targeting a known live session does not consult persistence, so durable backend health cannot make current in-memory history unreadable. Persisted exact reads list before loading, and reject a metadata mismatch rather than combining inconsistent observations.
`SessionQueryError.code` is a closed union: `SESSION_QUERY_EVENT_NOT_FOUND`, `SESSION_QUERY_INVALID_CONFIG`, `SESSION_QUERY_INVALID_SURFACE`, `SESSION_QUERY_INVALID_WINDOW`, `SESSION_QUERY_PERSISTENCE_FAILED`, `SESSION_QUERY_SESSION_NOT_FOUND`, and `SESSION_QUERY_SOURCE_CONFLICT`.
## Filtering and extraction
`SessionResultFilter` covers id, nullable cwd, created-at range, nullable parent, and source availability. `SessionEventResultFilter` covers seq/time ranges, event type, surface, and semantic text. Filter arrays are ANDed; values within one list clause are ORed. Empty list values match nothing, ranges are inclusive, and malformed ranges or closed-union values fail with `SESSION_QUERY_INVALID_FILTER`.
The text clause is deliberately independent of FTS providers: caller text is escaped into a Unicode, case-insensitive regular expression, and each whitespace run matches one or more whitespace characters. It is a literal semantic-text scan, not a full-text query. `extractSessionEventText()` and `buildSessionEventSearchDocuments()` define the shared first-party document projection; structural boundaries, stream chunks, request headers, and unknown declaration-merged variants produce no document.
## Full-text seam
`SessionSearchService` owns the independent `ctx.sessionSearch` key. `searchSessions(request, exec?)` groups the logical corpus by strongest matching event; `searchEvents(request, exec?)` searches one logical session. Both return opaque cursor pages, accept optional cancellation, and expose snippets without provider-specific numeric scores. Search requests accept only metadata event filters, because literal-text filtering is the scan path described above.
The package has no provider coordinator or registration protocol. A concrete backend owns observation, reconciliation, ranking, cursor generations, and query execution as one lifecycle; the first implementation is [`@deepseek-ai/dsh-session-query-sqlite`](../session-query-sqlite/README.md).
`SessionQueryError.code` is a closed union covering request validation, missing targets, malformed surfaces, source conflicts, persistence/index failures, cancellation, and invalid or stale cursors; the exact literals are defined in [`src/config.ts`](src/config.ts).
## Configuration
@@ -20,4 +33,4 @@ Persistence is optional and may mount or unmount dynamically. A cross-corpus lis
|---|---:|---|
| `readWindowMax` | `50` | Maximum `before` or `after` raw-event count. |
This phase deliberately has no filters, lineage/provenance traversal, extraction registry, search-provider protocol, index synchronization, or model-facing tool. Full-text search belongs beside its first real implementation; the proposed SQLite package and its single transaction/reconciliation owner are described in the [phase-two RFC](../../../docs/rfc/proposed/feature/2026-07-10-sqlite-session-query-provider.md).
The package deliberately has no lineage/provenance traversal, extractor registry, search-provider registry, index synchronization, caller authorization, or model-facing tool. The SQLite ownership and tokenizer decisions are recorded in the [implemented search RFC](../../../docs/rfc/implemented/feature/2026-07-10-sqlite-session-query-provider.md).

View File

@@ -1,4 +1,4 @@
/** Public configuration and typed failures for session-query. */
/** Public configuration and typed failures for session-query and search. */
import { HarnessError } from '@deepseek-ai/dsh-llm'
@@ -11,14 +11,21 @@ export interface Config {
readWindowMax?: number
}
/** Stable machine-routable failure taxonomy for exact session reads. */
/** Stable machine-routable failure taxonomy for session reads and search. */
export type SessionQueryErrorCode =
| 'SESSION_QUERY_ABORTED'
| 'SESSION_QUERY_EVENT_NOT_FOUND'
| 'SESSION_QUERY_INDEX_FAILED'
| 'SESSION_QUERY_INVALID_CONFIG'
| 'SESSION_QUERY_INVALID_CURSOR'
| 'SESSION_QUERY_INVALID_FILTER'
| 'SESSION_QUERY_INVALID_LIMIT'
| 'SESSION_QUERY_INVALID_QUERY'
| 'SESSION_QUERY_INVALID_SURFACE'
| 'SESSION_QUERY_INVALID_WINDOW'
| 'SESSION_QUERY_PERSISTENCE_FAILED'
| 'SESSION_QUERY_SESSION_NOT_FOUND'
| 'SESSION_QUERY_STALE_CURSOR'
| 'SESSION_QUERY_SOURCE_CONFLICT'
/** Typed session-query failure whose `code` is one closed taxonomy member. */

View File

@@ -5,6 +5,7 @@ import type { Session, SessionEvent, SessionHeader, SessionId } from '@deepseek-
import type SessionPersistence from '@deepseek-ai/dsh-session-persistence'
import type { SessionRecord } from './types.ts'
import { SessionQueryError } from './config.ts'
import { assertSessionHeadersCompatible } from './sources.ts'
/** Detached source selected for one exact read. */
export interface LogicalSession {
@@ -45,7 +46,7 @@ export class SessionCorpus {
}
for (const session of this._ctx.sessions.list()) {
const durable = records.get(session.id)
if (durable !== undefined) assertCompatibleHeaders(session.header, durable.header)
if (durable !== undefined) assertSessionHeadersCompatible(session.header, durable.header)
records.set(session.id, {
header: structuredClone(session.header),
live: true,
@@ -80,7 +81,7 @@ export class SessionCorpus {
{ cause: error },
)
}
assertCompatibleHeaders(loaded.meta, listed)
assertSessionHeadersCompatible(loaded.meta, listed)
return {
header: structuredClone(loaded.meta),
events: loaded.events.map(event => structuredClone(event)),
@@ -107,22 +108,6 @@ function snapshotLive(session: Session): LogicalSession {
}
}
function assertCompatibleHeaders(a: SessionHeader, b: SessionHeader): void {
if (
a.version !== b.version
|| a.id !== b.id
|| a.createdAt !== b.createdAt
|| a.cwd !== b.cwd
|| a.parentSession !== b.parentSession
|| a.seedLength !== b.seedLength
) {
throw new SessionQueryError(
`live and persisted headers conflict for session "${a.id}"`,
'SESSION_QUERY_SOURCE_CONFLICT',
)
}
}
function compareSessions(a: SessionRecord, b: SessionRecord): number {
return b.header.createdAt - a.header.createdAt || a.header.id.localeCompare(b.header.id)
}

View File

@@ -0,0 +1,74 @@
/** Shared event metadata and semantic-document projection. */
import { foldSurface } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEventRecord, SessionEventSearchDocument, SessionEventSurface } from './types.ts'
import { SessionQueryError } from './config.ts'
import { extractSessionEventText } from './extraction.ts'
/**
* Project a raw log into lightweight surface-aware event records.
* @param sessionId - session that owns the log.
* @param events - complete contiguous raw event log.
* @returns one record per event in ascending seq order.
*/
export function buildSessionEventRecords(
sessionId: SessionId,
events: readonly SessionEvent[],
): SessionEventRecord[] {
const surfaceBySeq = classifySurface(events)
return events.map(event => ({
sessionId,
seq: event.seq,
type: event.type,
time: event.time,
surface: surfaceBySeq.get(event.seq) ?? 'log-only',
}))
}
/**
* Build first-party semantic documents for one complete raw event log.
* @param sessionId - session that owns the log.
* @param events - complete contiguous raw event log.
* @returns searchable documents in ascending seq order; structural events are omitted.
*/
export function buildSessionEventSearchDocuments(
sessionId: SessionId,
events: readonly SessionEvent[],
): SessionEventSearchDocument[] {
const surfaceBySeq = classifySurface(events)
const documents: SessionEventSearchDocument[] = []
for (const event of events) {
const text = extractSessionEventText(event)
if (text.length === 0) continue
documents.push({
sessionId,
seq: event.seq,
type: event.type,
time: event.time,
surface: surfaceBySeq.get(event.seq) ?? 'log-only',
text,
})
}
return documents
}
function classifySurface(events: readonly SessionEvent[]): Map<number, SessionEventSurface> {
let folded: ReturnType<typeof foldSurface>
try {
folded = foldSurface(events)
} catch (error: unknown) {
throw new SessionQueryError(
/* v8 ignore next -- foldSurface throws Error instances */
`invalid session surface: ${error instanceof Error ? error.message : 'unknown error'}`,
'SESSION_QUERY_INVALID_SURFACE',
{ cause: error },
)
}
const result = new Map<number, SessionEventSurface>()
for (const node of folded.nodes) result.set(node.seq, 'current')
for (const replacement of folded.replacements) {
for (const seq of replacement.shadowedSeqs) result.set(seq, 'shadowed')
}
return result
}

View File

@@ -0,0 +1,93 @@
/** First-party semantic text extraction for session-query consumers. */
import type { SessionEvent } from '@deepseek-ai/dsh-session'
/**
* Extract searchable semantic text from one first-party session event.
*
* Structural boundaries, raw stream chunks, request envelopes, and unknown
* declaration-merged events contribute no text.
* @param event - event to inspect.
* @returns newline-joined semantic text, or an empty string when non-searchable.
*/
export function extractSessionEventText(event: SessionEvent): string {
switch (event.type) {
case 'user/message':
case 'assistant/message':
case 'context/message':
case 'steering/message':
return contentText(event.data.content)
case 'prompt/blocked':
return joinText([contentText(event.data.content), event.data.reason])
case 'tool/call':
return joinText([event.data.name, event.data.arguments])
case 'tool/result':
return joinText([
contentText(event.data.content),
event.data.error?.name ?? '',
event.data.error?.code ?? '',
])
case 'todo/write':
return joinText(event.data.todos.flatMap(todo => [todo.status, todo.content]))
case 'turn/end':
return turnEndText(event.data.reason)
case 'turn/start':
case 'step/start':
case 'step/end':
case 'assistant/chunk':
case 'request/header':
case 'request/header-delta':
return ''
// SessionEventMap is merge-extensible. Unknown events remain
// non-searchable until a concrete first-party consumer defines semantics.
default:
return ''
}
}
function turnEndText(reason: SessionEvent<'turn/end'>['data']['reason']): string {
switch (reason.kind) {
case 'error':
return joinText(['error', reason.message, reason.code ?? ''])
case 'aborted':
return joinText(['aborted', reason.reason ?? ''])
case 'rejected':
return joinText(['rejected', reason.reason])
case 'disposed':
case 'max-tokens':
case 'interrupted':
return reason.kind
case 'completed':
return ''
// TurnEndReasonMap is merge-extensible. Unknown outcomes stay out until
// their owner defines which detail is semantic rather than structural.
default:
return ''
}
}
type SessionContentBlock = SessionEvent<'user/message'>['data']['content'][number]
function contentText(content: readonly SessionContentBlock[]): string {
return joinText(content.flatMap(blockText))
}
function blockText(block: SessionContentBlock): string[] {
switch (block.type) {
case 'text':
case 'reasoning':
return [block.text]
case 'tool-call':
return [block.name, block.arguments]
case 'tool-result':
return block.content.flatMap(blockText)
// ContentBlockMap is merge-extensible. Unknown blocks do not become
// searchable merely because their payload happens to contain strings.
default:
return []
}
}
function joinText(parts: readonly string[]): string {
return parts.map(part => part.trim()).filter(Boolean).join('\n')
}

View File

@@ -0,0 +1,132 @@
/** Pure provider-independent predicates for logical sessions and event text. */
import type { SessionRecord, SessionEventSearchDocument, SessionEventResultFilter, SessionResultFilter, SessionResultRange } from './types.ts'
import { SessionQueryError } from './config.ts'
/**
* Apply ANDed logical-session filters while preserving input order.
* @param records - detached logical-session records to inspect.
* @param filters - clauses whose list values are ORed within each clause.
* @returns records accepted by every clause.
*/
export function filterSessionResults<T extends SessionRecord>(
records: readonly T[],
filters: readonly SessionResultFilter[] = [],
): T[] {
const predicates = filters.map(sessionPredicate)
return records.filter(record => predicates.every(predicate => predicate(record)))
}
/**
* Apply ANDed event filters to extracted semantic documents.
* @param documents - semantic documents produced by {@link buildSessionEventSearchDocuments}.
* @param filters - metadata and literal-text predicates.
* @returns documents accepted by every clause, in input order.
*/
export function filterSessionEventDocuments<T extends SessionEventSearchDocument>(
documents: readonly T[],
filters: readonly SessionEventResultFilter[] = [],
): T[] {
const predicates = filters.map(eventPredicate)
return documents.filter(document => predicates.every(predicate => predicate(document)))
}
/**
* Compile a literal case-insensitive, whitespace-flexible semantic-text match.
* @param text - caller-provided literal text.
* @returns Unicode-aware regular expression safe from regex injection.
*/
export function compileSessionTextFilter(text: string): RegExp {
const trimmed = text.trim()
if (trimmed.length === 0) {
throw new SessionQueryError(
'session text filter must contain non-whitespace text',
'SESSION_QUERY_INVALID_FILTER',
)
}
const pattern = trimmed
.split(/\s+/u)
.map(part => part.replace(/[.*+?^${}()|[\]\\]/gu, '\\$&'))
.join('\\s+')
return new RegExp(pattern, 'iu')
}
function sessionPredicate(filter: SessionResultFilter): (record: SessionRecord) => boolean {
switch (filter.kind) {
case 'id':
return record => filter.values.includes(record.header.id)
case 'cwd':
return record => filter.values.includes(record.header.cwd ?? null)
case 'created-at': {
const range = validateRange(filter.kind, filter)
return record => matchesRange(record.header.createdAt, range)
}
case 'parent':
return record => filter.values.includes(record.header.parentSession ?? null)
case 'availability':
assertAllowedValues(filter.kind, filter.values, ['live', 'persisted'])
return record => filter.values.some(value => value === 'live' ? record.live : record.persisted)
}
}
function eventPredicate(filter: SessionEventResultFilter): (document: SessionEventSearchDocument) => boolean {
switch (filter.kind) {
case 'seq': {
const range = validateRange(filter.kind, filter)
return document => matchesRange(document.seq, range)
}
case 'time': {
const range = validateRange(filter.kind, filter)
return document => matchesRange(document.time, range)
}
case 'type':
return document => filter.values.includes(document.type)
case 'surface':
assertAllowedValues(filter.kind, filter.values, ['current', 'shadowed', 'log-only'])
return document => filter.values.includes(document.surface)
case 'text': {
const pattern = compileSessionTextFilter(filter.text)
return document => pattern.test(document.text)
}
}
}
function assertAllowedValues(
name: string,
values: readonly string[],
allowed: readonly string[],
): void {
for (const value of values) {
if (!allowed.includes(value)) {
throw new SessionQueryError(
`session ${name} filter contains unknown value "${value}"`,
'SESSION_QUERY_INVALID_FILTER',
)
}
}
}
function validateRange(name: string, range: SessionResultRange): SessionResultRange {
if (range.from !== undefined && !Number.isFinite(range.from)) {
throw invalidRange(name, 'from must be finite')
}
if (range.to !== undefined && !Number.isFinite(range.to)) {
throw invalidRange(name, 'to must be finite')
}
if (range.from !== undefined && range.to !== undefined && range.from > range.to) {
throw invalidRange(name, 'from must be less than or equal to to')
}
return range
}
function matchesRange(value: number, range: SessionResultRange): boolean {
return (range.from === undefined || value >= range.from)
&& (range.to === undefined || value <= range.to)
}
function invalidRange(name: string, detail: string): SessionQueryError {
return new SessionQueryError(
`session ${name} filter ${detail}`,
'SESSION_QUERY_INVALID_FILTER',
)
}

View File

@@ -6,13 +6,20 @@
import { Context, Service } from 'cordis'
import z from 'schemastery'
import { foldSurface } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionId } from '@deepseek-ai/dsh-session'
import type {
SessionEventResultFilter,
SessionEventReadRequest,
SessionEventRecord,
SessionEventSearchHit,
SessionEventSearchDocument,
SessionEventSearchRequest,
SessionEventWindow,
SessionRecord,
SessionSearchExecContext,
SessionSearchHit,
SessionSearchPage,
SessionSearchRequest,
} from './types.ts'
import {
SESSION_QUERY_READ_WINDOW_MAX,
@@ -20,17 +27,58 @@ import {
type Config,
} from './config.ts'
import { SessionCorpus } from './corpus.ts'
import { buildSessionEventRecords, buildSessionEventSearchDocuments } from './documents.ts'
import { filterSessionEventDocuments } from './filters.ts'
export type * from './types.ts'
export type { Config, SessionQueryErrorCode } from './config.ts'
export { SESSION_QUERY_READ_WINDOW_MAX, SessionQueryError } from './config.ts'
export { extractSessionEventText } from './extraction.ts'
export { buildSessionEventRecords, buildSessionEventSearchDocuments } from './documents.ts'
export { compileSessionTextFilter, filterSessionEventDocuments, filterSessionResults } from './filters.ts'
export { assertSessionHeadersCompatible } from './sources.ts'
declare module 'cordis' {
interface Context {
sessionQuery: SessionQueryService
sessionSearch: SessionSearchService
}
}
/**
* Abstract full-text search service implemented by one concrete backend.
*
* The implementation owns source observation, reconciliation, cursor
* generations, ranking, and query execution as one lifecycle.
*/
export abstract class SessionSearchService extends Service {
constructor(ctx: Context) {
super(ctx, 'sessionSearch')
}
/**
* Search the live-preferred logical corpus and group by session.
* @param request - query text, metadata filters, page size, and cursor.
* @param exec - optional cancellation control.
* @returns session hits ranked by their strongest matching event.
*/
abstract searchSessions(
request: SessionSearchRequest,
exec?: SessionSearchExecContext,
): Promise<SessionSearchPage<SessionSearchHit>>
/**
* Search events within one live-preferred logical session.
* @param request - target session, query text, filters, page size, and cursor.
* @param exec - optional cancellation control.
* @returns matching event hits in deterministic relevance order.
*/
abstract searchEvents(
request: SessionEventSearchRequest,
exec?: SessionSearchExecContext,
): Promise<SessionSearchPage<SessionEventSearchHit>>
}
/** Live-preferred logical-corpus and exact-event read service. */
export class SessionQueryService extends Service {
static inject = ['sessions']
@@ -68,7 +116,22 @@ export class SessionQueryService extends Service {
*/
async listEvents(sessionId: SessionId): Promise<SessionEventRecord[]> {
const loaded = await this._corpus.load(sessionId)
return eventRecords(sessionId, loaded.events)
return buildSessionEventRecords(sessionId, loaded.events)
}
/**
* Scan first-party semantic event documents with provider-independent filters.
* @param sessionId - live-preferred session id to scan.
* @param filters - ANDed metadata and literal-text predicates.
* @returns matching semantic documents in ascending seq order.
*/
async filterEvents(
sessionId: SessionId,
filters: readonly SessionEventResultFilter[],
): Promise<SessionEventSearchDocument[]> {
const loaded = await this._corpus.load(sessionId)
const documents = buildSessionEventSearchDocuments(sessionId, loaded.events)
return filterSessionEventDocuments(documents, filters)
}
/**
@@ -110,27 +173,4 @@ export class SessionQueryService extends Service {
}
}
function eventRecords(sessionId: SessionId, events: readonly SessionEvent[]): SessionEventRecord[] {
let folded: ReturnType<typeof foldSurface>
try {
folded = foldSurface(events)
} catch (error: unknown) {
throw new SessionQueryError(
/* v8 ignore next -- foldSurface throws Error instances */
`invalid session surface: ${error instanceof Error ? error.message : 'unknown error'}`,
'SESSION_QUERY_INVALID_SURFACE',
{ cause: error },
)
}
const current = new Set(folded.nodes.map(node => node.seq))
const shadowed = new Set(folded.replacements.flatMap(replacement => replacement.shadowedSeqs))
return events.map(event => ({
sessionId,
seq: event.seq,
type: event.type,
time: event.time,
surface: current.has(event.seq) ? 'current' : shadowed.has(event.seq) ? 'shadowed' : 'log-only',
}))
}
export default SessionQueryService

View File

@@ -0,0 +1,25 @@
/** Shared immutable-header checks for logical session source observers. */
import type { SessionHeader } from '@deepseek-ai/dsh-session'
import { SessionQueryError } from './config.ts'
/**
* Reject incompatible observations of one logical session source.
* @param a - first live, listed, or loaded header observation.
* @param b - second header observation expected to identify the same source.
*/
export function assertSessionHeadersCompatible(a: SessionHeader, b: SessionHeader): void {
if (
a.version !== b.version
|| a.id !== b.id
|| a.createdAt !== b.createdAt
|| a.cwd !== b.cwd
|| a.parentSession !== b.parentSession
|| a.seedLength !== b.seedLength
) {
throw new SessionQueryError(
`session source headers conflict for session "${a.id}"`,
'SESSION_QUERY_SOURCE_CONFLICT',
)
}
}

View File

@@ -58,3 +58,99 @@ export interface SessionEventWindow {
/** Last seq included in `events`. */
endSeq: number
}
/** Inclusive numeric interval used by time and sequence filters. */
export interface SessionResultRange {
/** Inclusive lower bound. */
from?: number
/** Inclusive upper bound. */
to?: number
}
/** Source availability predicates understood by logical-session filters. */
export type SessionAvailability = 'live' | 'persisted'
/**
* One logical-session predicate. A filter array is ANDed; `values` within a
* clause are ORed.
*/
export type SessionResultFilter =
| { kind: 'id'; values: readonly SessionId[] }
| { kind: 'cwd'; values: readonly (string | null)[] }
| ({ kind: 'created-at' } & SessionResultRange)
| { kind: 'parent'; values: readonly (SessionId | null)[] }
| { kind: 'availability'; values: readonly SessionAvailability[] }
/**
* One event predicate. A filter array is ANDed; list-valued clauses are ORed.
* Text is a literal, case-insensitive, whitespace-flexible semantic-text scan.
*/
export type SessionEventResultFilter =
| ({ kind: 'seq' } & SessionResultRange)
| ({ kind: 'time' } & SessionResultRange)
| { kind: 'type'; values: readonly SessionEventType[] }
| { kind: 'surface'; values: readonly SessionEventSurface[] }
| { kind: 'text'; text: string }
/** Event predicates a full-text provider can apply before relevance ranking. */
export type SessionEventMetadataFilter = Exclude<SessionEventResultFilter, { kind: 'text' }>
/** Searchable semantic document derived from one session event. */
export interface SessionEventSearchDocument extends SessionEventRecord {
/** First-party semantic text used by scan filters and full-text indexes. */
text: string
}
/** One cursor-paginated result page. */
export interface SessionSearchPage<T> {
/** Results for this page in contract-defined order. */
items: readonly T[]
/** Opaque continuation cursor, absent on the final page. */
nextCursor?: string
}
/** Controls shared by cross-session and within-session search calls. */
export interface SessionSearchExecContext {
/** Abort caller waiting and interrupt provider work where supported. */
signal?: AbortSignal
}
/** Cross-session full-text search request. */
export interface SessionSearchRequest {
/** Full-text query interpreted as data, never executable FTS syntax. */
query: string
/** Logical-session predicates applied before event ranking. */
sessionFilters?: readonly SessionResultFilter[]
/** Event predicates applied before event ranking. */
eventFilters?: readonly SessionEventMetadataFilter[]
/** Maximum sessions in this page. */
limit?: number
/** Opaque cursor returned for the identical normalized request. */
cursor?: string
}
/** Within-session full-text search request. */
export interface SessionEventSearchRequest {
/** Session whose live-preferred logical log is searched. */
sessionId: SessionId
/** Full-text query interpreted as data, never executable FTS syntax. */
query: string
/** Event predicates applied before ranking. */
filters?: readonly SessionEventMetadataFilter[]
/** Maximum events in this page. */
limit?: number
/** Opaque cursor returned for the identical normalized request. */
cursor?: string
}
/** One event full-text search hit with a bounded plain-text excerpt. */
export interface SessionEventSearchHit extends SessionEventRecord {
/** Plain text excerpt selected around the match. */
snippet: string
}
/** One grouped cross-session hit, ranked by its strongest matching event. */
export interface SessionSearchHit extends SessionRecord {
/** Strongest matching event for this session. */
bestMatch: SessionEventSearchHit
}

View File

@@ -0,0 +1,209 @@
import { describe, expect, it } from 'vitest'
import { Context } from 'cordis'
import { CallId } from '@deepseek-ai/dsh-llm'
import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
import type { SessionEvent, SessionHeader } from '@deepseek-ai/dsh-session'
import SessionQueryService, {
buildSessionEventRecords,
buildSessionEventSearchDocuments,
compileSessionTextFilter,
extractSessionEventText,
filterSessionEventDocuments,
filterSessionResults,
SessionSearchService,
type SessionEventSearchHit,
type SessionEventSearchRequest,
type SessionQueryErrorCode,
type SessionSearchExecContext,
type SessionSearchHit,
type SessionSearchPage,
type SessionSearchRequest,
} from '@deepseek-ai/dsh-session-query'
const id = SessionId('session')
function header(value: string, extra: Partial<SessionHeader> = {}): SessionHeader {
return { version: SESSION_FORMAT_VERSION, id: SessionId(value), createdAt: 10, ...extra }
}
function expectCode(code: SessionQueryErrorCode): Error {
return expect.objectContaining({ code }) as Error
}
describe('session-query semantic extraction', () => {
it('extracts first-party message, tool, todo, and failure detail', () => {
const callId = CallId('call')
const messageContent: SessionEvent<'user/message'>['data']['content'] = [
{ type: 'text', text: ' visible ' },
{ type: 'reasoning', text: 'thought' },
{ type: 'tool-call', id: callId, name: 'read', arguments: '{"path":"a"}' },
{
type: 'tool-result',
toolCallId: callId,
content: [{ type: 'text', text: 'nested' }],
isError: false,
},
{ type: 'future-content', payload: 'hidden' } as never,
]
const events: SessionEvent[] = [
{ type: 'user/message', seq: 0, time: 1, data: { content: messageContent, source: { kind: 'user' } }, surfaceOp: 'append' },
{ type: 'assistant/message', seq: 1, time: 2, data: { turn: 1, step: 1, content: messageContent }, surfaceOp: 'append' },
{ type: 'context/message', seq: 2, time: 3, data: { content: messageContent, source: { kind: 'plugin', plugin: 'test' } }, surfaceOp: 'append' },
{ type: 'steering/message', seq: 3, time: 4, data: { turn: 1, content: messageContent, source: { kind: 'user' } }, surfaceOp: 'append' },
{ type: 'prompt/blocked', seq: 4, time: 5, data: { content: [{ type: 'text', text: 'unsafe' }], source: { kind: 'user' }, reason: 'policy' } },
{ type: 'tool/call', seq: 5, time: 6, data: { turn: 1, step: 1, callId, name: 'bash', arguments: '{"cmd":"pwd"}' } },
{ type: 'tool/result', seq: 6, time: 7, data: { turn: 1, step: 1, callId, content: [{ type: 'text', text: 'failed' }], isError: true, error: { name: 'Oops', code: 'E_OOPS' } }, surfaceOp: 'append' },
{ type: 'tool/result', seq: 7, time: 8, data: { turn: 1, step: 1, callId, content: [], isError: false }, surfaceOp: 'append' },
{ type: 'todo/write', seq: 8, time: 9, data: { todos: [{ status: 'in_progress', content: 'ship search' }] } },
]
for (const event of events.slice(0, 4)) {
expect(extractSessionEventText(event)).toBe('visible\nthought\nread\n{"path":"a"}\nnested')
}
expect(extractSessionEventText(events[4]!)).toBe('unsafe\npolicy')
expect(extractSessionEventText(events[5]!)).toBe('bash\n{"cmd":"pwd"}')
expect(extractSessionEventText(events[6]!)).toBe('failed\nOops\nE_OOPS')
expect(extractSessionEventText(events[7]!)).toBe('')
expect(extractSessionEventText(events[8]!)).toBe('in_progress\nship search')
})
it('extracts meaningful turn outcomes and skips structural or unknown events', () => {
const reasons: Array<[SessionEvent<'turn/end'>['data']['reason'], string]> = [
[{ kind: 'error', step: 2, message: 'boom', code: 'E' }, 'error\nboom\nE'],
[{ kind: 'error', step: 2, message: 'boom' }, 'error\nboom'],
[{ kind: 'aborted', reason: 'cancelled' }, 'aborted\ncancelled'],
[{ kind: 'aborted' }, 'aborted'],
[{ kind: 'rejected', reason: 'denied' }, 'rejected\ndenied'],
[{ kind: 'disposed' }, 'disposed'],
[{ kind: 'max-tokens' }, 'max-tokens'],
[{ kind: 'interrupted' }, 'interrupted'],
[{ kind: 'completed' }, ''],
[{ kind: 'future-status' } as never, ''],
]
for (const [reason, text] of reasons) {
expect(extractSessionEventText({ type: 'turn/end', seq: 0, time: 1, data: { turn: 1, reason } })).toBe(text)
}
const structural: SessionEvent[] = [
{ type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
{ type: 'step/start', seq: 1, time: 1, data: { turn: 1, step: 1 } },
{ type: 'step/end', seq: 2, time: 1, data: { turn: 1, step: 1 } },
{ type: 'assistant/chunk', seq: 3, time: 1, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'raw' } } },
{ type: 'request/header', seq: 4, time: 1, data: { header: { config: { model: 'test' } }, reason: 'initial' } },
{ type: 'request/header-delta', seq: 5, time: 1, data: {} },
{ type: 'future/event', seq: 6, time: 1, data: { text: 'hidden' } } as never,
]
expect(structural.map(extractSessionEventText)).toEqual(['', '', '', '', '', '', ''])
})
})
describe('session-query document and filter helpers', () => {
const events: SessionEvent[] = [
{ type: 'user/message', seq: 0, time: 10, data: { content: [{ type: 'text', text: 'Hello\n(AI)+' }], source: { kind: 'user' } }, surfaceOp: 'append' },
{ type: 'assistant/chunk', seq: 1, time: 11, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'raw' } } },
{ type: 'assistant/message', seq: 2, time: 12, data: { turn: 1, step: 1, content: [{ type: 'text', text: 'replacement' }] }, surfaceOp: { op: 'replace', start: 0, end: 0 } },
{ type: 'turn/end', seq: 3, time: 13, data: { turn: 1, reason: { kind: 'interrupted' } } },
]
it('classifies every event and omits non-semantic documents', () => {
expect(buildSessionEventRecords(id, events).map(record => record.surface))
.toEqual(['shadowed', 'log-only', 'current', 'log-only'])
const documents = buildSessionEventSearchDocuments(id, events)
expect(documents.map(document => [document.seq, document.text, document.surface])).toEqual([
[0, 'Hello\n(AI)+', 'shadowed'],
[2, 'replacement', 'current'],
[3, 'interrupted', 'log-only'],
])
})
it('applies every session clause with OR values and validates closed values', () => {
const parent = SessionId('parent')
const records = [
{ header: header('a', { cwd: '/a', parentSession: parent }), live: true, persisted: false, marker: 1 },
{ header: header('b', { createdAt: 20 }), live: false, persisted: true, marker: 2 },
]
expect(filterSessionResults(records, [
{ kind: 'id', values: [SessionId('a'), SessionId('x')] },
{ kind: 'cwd', values: ['/a', null] },
{ kind: 'created-at', from: 5, to: 15 },
{ kind: 'parent', values: [parent, null] },
{ kind: 'availability', values: ['live'] },
])).toEqual([records[0]])
expect(filterSessionResults(records, [{ kind: 'cwd', values: [null] }])).toEqual([records[1]])
expect(filterSessionResults(records, [{ kind: 'parent', values: [null] }])).toEqual([records[1]])
expect(filterSessionResults(records, [{ kind: 'availability', values: ['persisted'] }])).toEqual([records[1]])
expect(() => filterSessionResults(records, [{ kind: 'availability', values: ['remote' as never] }]))
.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
})
it('applies event metadata and safe literal text clauses', () => {
const documents = buildSessionEventSearchDocuments(id, events).map((document, marker) => ({ ...document, marker }))
expect(filterSessionEventDocuments(documents, [
{ kind: 'seq', from: 0, to: 1 },
{ kind: 'time', from: 9, to: 11 },
{ kind: 'type', values: ['user/message', 'tool/result'] },
{ kind: 'surface', values: ['shadowed'] },
{ kind: 'text', text: 'hello (ai)+' },
])).toEqual([documents[0]])
expect(compileSessionTextFilter('CAFÉ').test('café')).toBe(true)
expect(filterSessionEventDocuments(documents)).toEqual(documents)
expect(filterSessionEventDocuments(documents, [{ kind: 'surface', values: [] }])).toEqual([])
expect(() => filterSessionEventDocuments(documents, [{ kind: 'surface', values: ['future' as never] }]))
.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
expect(() => compileSessionTextFilter(' \n ')).toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
})
it('rejects malformed range filters and malformed surfaces', () => {
const documents = buildSessionEventSearchDocuments(id, events)
for (const filter of [
{ kind: 'seq', from: Number.NaN },
{ kind: 'seq', to: Number.POSITIVE_INFINITY },
{ kind: 'time', from: 2, to: 1 },
] as const) {
expect(() => filterSessionEventDocuments(documents, [filter]))
.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
}
expect(() => filterSessionResults([], [{ kind: 'created-at', from: Number.NaN }]))
.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
expect(() => filterSessionResults([{ header: header('x'), live: true, persisted: false }], [
{ kind: 'created-at', from: Number.NaN },
])).toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
const malformed: SessionEvent[] = [{
type: 'assistant/message',
seq: 0,
time: 1,
data: { turn: 1, step: 1, content: [{ type: 'text', text: 'bad' }] },
surfaceOp: { op: 'replace', start: 9, end: 9 },
}]
expect(() => buildSessionEventRecords(id, malformed)).toThrow(expectCode('SESSION_QUERY_INVALID_SURFACE'))
})
it('exposes the scan path on the concrete exact-read service', async () => {
const ctx = new Context()
await ctx.plugin(SessionStore)
await ctx.plugin(SessionQueryService)
const session = ctx.sessions.create(id)
session.append('user/message', { content: [{ type: 'text', text: 'Alpha\n beta' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
session.append('user/message', { content: [{ type: 'text', text: 'other' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
await expect(ctx.sessionQuery.filterEvents(id, [{ kind: 'text', text: 'alpha beta' }]))
.resolves.toMatchObject([{ seq: 0, text: 'Alpha\n beta' }])
})
})
class TestSearchService extends SessionSearchService {
searchSessions(_request: SessionSearchRequest, _exec?: SessionSearchExecContext): Promise<SessionSearchPage<SessionSearchHit>> {
return Promise.resolve({ items: [] })
}
searchEvents(_request: SessionEventSearchRequest, _exec?: SessionSearchExecContext): Promise<SessionSearchPage<SessionEventSearchHit>> {
return Promise.resolve({ items: [] })
}
}
it('registers the abstract search seam under its independent ctx key', async () => {
const ctx = new Context()
const fiber = await ctx.plugin(TestSearchService)
await expect(ctx.sessionSearch.searchSessions({ query: 'AI' })).resolves.toEqual({ items: [] })
await expect(ctx.sessionSearch.searchEvents({ sessionId: id, query: 'AI' })).resolves.toEqual({ items: [] })
await fiber.dispose()
expect(ctx.sessionSearch).toBeUndefined()
})