1
0

sqlite.spec.ts 75 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659
  1. import { createUserMessage } from '@deepseek-ai/dsh-llm'
  2. import { afterEach, describe, expect, it, vi } from 'vitest'
  3. import { Context, type Fiber } from 'cordis'
  4. import { DatabaseSync } from 'node:sqlite'
  5. import { chmod, mkdtemp, rm, stat, writeFile } from 'node:fs/promises'
  6. import { tmpdir } from 'node:os'
  7. import { dirname, join } from 'node:path'
  8. import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
  9. import type { SessionEvent, SessionHeader, SessionId as SessionIdType } from '@deepseek-ai/dsh-session'
  10. import SessionPersistence, { SessionPersistenceRevision } from '@deepseek-ai/dsh-session-persistence'
  11. import type { SessionPersistenceSnapshot } from '@deepseek-ai/dsh-session-persistence'
  12. import SessionPersistenceSqlite from '@deepseek-ai/dsh-session-persistence-sqlite'
  13. import SessionQuerySqlite, {
  14. SESSION_QUERY_SQLITE_SCHEMA_VERSION,
  15. } from '@deepseek-ai/dsh-session-query-sqlite'
  16. import {
  17. SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY,
  18. SessionQueryError,
  19. SessionSearchCursor,
  20. type SessionAvailability,
  21. type SessionQueryErrorCode,
  22. type SessionSearchRequest,
  23. } from '@deepseek-ai/dsh-session-query'
  24. const temporaryDirectories: string[] = []
  25. afterEach(async () => {
  26. for (const directory of temporaryDirectories.splice(0)) {
  27. await rm(directory, { recursive: true, force: true })
  28. }
  29. })
  30. async function temporaryPath(name = 'search.db'): Promise<string> {
  31. const directory = await mkdtemp(join(tmpdir(), 'dsh-session-search-'))
  32. temporaryDirectories.push(directory)
  33. return join(directory, name)
  34. }
  35. function header(id: string, createdAt = 1, extra: Partial<SessionHeader> = {}): SessionHeader {
  36. return { version: SESSION_FORMAT_VERSION, id: SessionId(id), createdAt, ...extra }
  37. }
  38. function messageEvents(text: string, time = 1): SessionEvent[] {
  39. return [{
  40. type: 'user/message',
  41. seq: 0,
  42. time,
  43. data: createUserMessage({
  44. content: [{ type: 'text', text }], source: { kind: 'user' },
  45. }),
  46. surfaceOp: 'append',
  47. }]
  48. }
  49. function expectCode(code: SessionQueryErrorCode): Error {
  50. return expect.objectContaining({ code }) as Error
  51. }
  52. function replaceCursorOffset(
  53. cursor: ReturnType<typeof SessionSearchCursor>,
  54. offset: number,
  55. ): ReturnType<typeof SessionSearchCursor> {
  56. const payload = JSON.parse(
  57. Buffer.from(cursor, 'base64url').toString('utf8'),
  58. ) as Record<string, unknown>
  59. return SessionSearchCursor(Buffer.from(JSON.stringify({ ...payload, offset }), 'utf8').toString('base64url'))
  60. }
  61. class TestPersistence extends SessionPersistence {
  62. static entries = new Map<SessionIdType, { meta: SessionHeader; events: SessionEvent[] }>()
  63. static revisions = new Map<SessionIdType, number>()
  64. static nextRevision = 0
  65. static loads = new Map<SessionIdType, number>()
  66. static inspections = new Map<SessionIdType, number>()
  67. static inspectSignals: Array<AbortSignal | undefined> = []
  68. static snapshotSignals: Array<AbortSignal | undefined> = []
  69. static loadEffect: ((entry: { meta: SessionHeader; events: SessionEvent[] }) => void) | undefined
  70. static inspectEffect: ((
  71. entry: { meta: SessionHeader; events: SessionEvent[] },
  72. signal?: AbortSignal,
  73. ) => void | Promise<void>) | undefined
  74. static listGate: Promise<void> | undefined
  75. static listStarted: (() => void) | undefined
  76. static snapshotEffect: ((signal?: AbortSignal) => void | Promise<void>) | undefined
  77. static snapshotOverride: (() => SessionPersistenceSnapshot[]) | undefined
  78. static failure: unknown
  79. locate(_meta: SessionHeader): undefined {
  80. return undefined
  81. }
  82. static reset(entries: readonly { meta: SessionHeader; events: SessionEvent[] }[] = []): void {
  83. this.entries = new Map()
  84. this.revisions = new Map()
  85. this.loads = new Map()
  86. this.inspections = new Map()
  87. this.inspectSignals = []
  88. this.snapshotSignals = []
  89. this.loadEffect = undefined
  90. this.inspectEffect = undefined
  91. for (const entry of entries) this.set(entry)
  92. this.listGate = undefined
  93. this.listStarted = undefined
  94. this.snapshotEffect = undefined
  95. this.snapshotOverride = undefined
  96. this.failure = undefined
  97. }
  98. static set(entry: { meta: SessionHeader; events: SessionEvent[] }): void {
  99. this.entries.set(entry.meta.id, structuredClone(entry))
  100. this.revisions.set(entry.meta.id, ++this.nextRevision)
  101. }
  102. create(meta: SessionHeader): Promise<void> {
  103. TestPersistence.set({ meta, events: [] })
  104. return Promise.resolve()
  105. }
  106. append(id: SessionIdType, events: readonly SessionEvent[]): Promise<void> {
  107. const entry = TestPersistence.entries.get(id)
  108. if (entry === undefined) return Promise.reject(new Error('missing test session'))
  109. entry.events.push(...structuredClone(events))
  110. TestPersistence.revisions.set(id, ++TestPersistence.nextRevision)
  111. return Promise.resolve()
  112. }
  113. async load(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
  114. TestPersistence.loads.set(id, (TestPersistence.loads.get(id) ?? 0) + 1)
  115. if (TestPersistence.failure !== undefined) throw TestPersistence.failure
  116. const entry = TestPersistence.entries.get(id)
  117. if (entry === undefined) throw new Error('missing test session')
  118. if (TestPersistence.loadEffect !== undefined) {
  119. const effect = TestPersistence.loadEffect
  120. TestPersistence.loadEffect = undefined
  121. effect(entry)
  122. TestPersistence.revisions.set(id, ++TestPersistence.nextRevision)
  123. }
  124. return structuredClone(entry)
  125. }
  126. async inspect(id: SessionIdType, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
  127. TestPersistence.inspections.set(id, (TestPersistence.inspections.get(id) ?? 0) + 1)
  128. TestPersistence.inspectSignals.push(signal)
  129. if (TestPersistence.failure !== undefined) throw TestPersistence.failure
  130. const entry = TestPersistence.entries.get(id)
  131. if (entry === undefined) throw new Error('missing test session')
  132. await TestPersistence.inspectEffect?.(entry, signal)
  133. TestPersistence.inspectEffect = undefined
  134. return structuredClone(entry)
  135. }
  136. async readFrom(id: SessionIdType, fromSeq: number, signal?: AbortSignal): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
  137. const whole = await this.inspect(id, signal)
  138. return { meta: whole.meta, events: whole.events.filter(event => event.seq >= fromSeq) }
  139. }
  140. async list(): Promise<SessionHeader[]> {
  141. TestPersistence.listStarted?.()
  142. await TestPersistence.listGate
  143. if (TestPersistence.failure !== undefined) throw TestPersistence.failure
  144. return [...TestPersistence.entries.values()].map(entry => structuredClone(entry.meta))
  145. }
  146. async listSnapshots(signal?: AbortSignal): Promise<SessionPersistenceSnapshot[]> {
  147. TestPersistence.snapshotSignals.push(signal)
  148. TestPersistence.listStarted?.()
  149. await TestPersistence.listGate
  150. if (TestPersistence.failure !== undefined) throw TestPersistence.failure
  151. const snapshots = TestPersistence.snapshotOverride?.()
  152. ?? [...TestPersistence.entries.values()].map(entry => ({
  153. header: structuredClone(entry.meta),
  154. revision: SessionPersistenceRevision(`test:${TestPersistence.revisions.get(entry.meta.id)}`),
  155. }))
  156. await TestPersistence.snapshotEffect?.(signal)
  157. return snapshots
  158. }
  159. }
  160. async function liveContext(config: ConstructorParameters<typeof SessionQuerySqlite>[1] = { path: ':memory:' }): Promise<Context> {
  161. const ctx = new Context()
  162. await ctx.plugin(SessionStore)
  163. await ctx.plugin(SessionQuerySqlite, config)
  164. return ctx
  165. }
  166. describe('SQLite session search', () => {
  167. it('defaults and validates persisted inspection concurrency through its Cordis config', async () => {
  168. const defaultCtx = await liveContext()
  169. expect((defaultCtx.sessionQuery as SessionQuerySqlite).config.persistedInspectConcurrency)
  170. .toBe(SESSION_QUERY_DEFAULT_PERSISTED_INSPECT_CONCURRENCY)
  171. const configuredValue = 2
  172. const configured = new SessionQuerySqlite.Config({
  173. path: ':memory:',
  174. persistedInspectConcurrency: configuredValue,
  175. })
  176. expect(configured.persistedInspectConcurrency).toBe(configuredValue)
  177. const configuredCtx = await liveContext(configured)
  178. expect((configuredCtx.sessionQuery as SessionQuerySqlite).config.persistedInspectConcurrency)
  179. .toBe(configuredValue)
  180. for (const persistedInspectConcurrency of [0, Number.MAX_SAFE_INTEGER + 1]) {
  181. expect(() => new SessionQuerySqlite.Config({
  182. path: ':memory:',
  183. persistedInspectConcurrency,
  184. })).toThrow()
  185. }
  186. })
  187. it('searches two-character Unicode61 tokens in live-only sessions', async () => {
  188. const ctx = await liveContext({ path: ':memory:', snippetChars: 20 })
  189. const session = ctx.sessions.create(SessionId('live'), {
  190. meta: { cwd: '/work', createdAt: 10, seedLength: 1, delegationDepth: 2 },
  191. })
  192. session.append(
  193. 'user/message',
  194. createUserMessage({
  195. content: [{ type: 'text', text: 'An AI helper' }], source: { kind: 'user' },
  196. }),
  197. { surfaceOp: 'append' },
  198. )
  199. await expect(ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'AI' }))
  200. .resolves.toMatchObject({
  201. session: { ...session.header, seedLength: 1 },
  202. items: [{ sessionId: session.id, seq: 0, snippet: 'An AI helper' }],
  203. })
  204. await expect(ctx.sessionQuery.searchSessions({ query: 'AI' }))
  205. .resolves.toMatchObject({ items: [{ header: { ...session.header, seedLength: 1 }, live: true, persisted: false }] })
  206. })
  207. it('searches all surfaces by default and applies metadata before ranking', async () => {
  208. const ctx = await liveContext({ path: ':memory:', defaultLimit: 10, maxLimit: 20 })
  209. const parent = SessionId('parent')
  210. const events: SessionEvent[] = [
  211. { type: 'user/message', seq: 0, time: 10, data: createUserMessage({
  212. content: [{ type: 'text', text: 'needle original' }], source: { kind: 'user' },
  213. }), surfaceOp: 'append' },
  214. { type: 'assistant/chunk', seq: 1, time: 11, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'needle raw' } } },
  215. { type: 'user/message', seq: 2, time: 12, data: createUserMessage({
  216. content: [{ type: 'text', text: 'needle summary' }], source: { kind: 'plugin', plugin: 'test' },
  217. }), surfaceOp: { op: 'replace', start: 0, end: 0 }, sourceEventSeqs: [0] },
  218. { type: 'turn/end', seq: 3, time: 13, data: { turn: 1, reason: { kind: 'error', step: 1, message: 'needle failure' } } },
  219. ]
  220. ctx.sessions.create(SessionId('a'), { seed: events, meta: { cwd: '/a', parentSession: parent, createdAt: 20 } })
  221. ctx.sessions.create(SessionId('b'), { seed: messageEvents('needle peer', 12), meta: { createdAt: 20 } })
  222. const all = await ctx.sessionQuery.searchEvents({ sessionId: SessionId('a'), query: 'needle' })
  223. expect(new Set(all.items.map(item => item.surface))).toEqual(new Set(['current', 'shadowed', 'log-only']))
  224. await expect(ctx.sessionQuery.searchEvents({
  225. sessionId: SessionId('a'),
  226. query: 'needle',
  227. filters: [
  228. { kind: 'seq', from: 2, to: 2 },
  229. { kind: 'time', from: 12, to: 12 },
  230. { kind: 'type', values: ['user/message'] },
  231. { kind: 'surface', values: ['current'] },
  232. ],
  233. })).resolves.toMatchObject({ items: [{ seq: 2, surface: 'current' }] })
  234. const grouped = await ctx.sessionQuery.searchSessions({
  235. query: 'needle',
  236. sessionFilters: [
  237. { kind: 'id', values: [SessionId('a')] },
  238. { kind: 'cwd', values: ['/a'] },
  239. { kind: 'created-at', from: 20, to: 20 },
  240. { kind: 'parent', values: [parent] },
  241. { kind: 'availability', values: ['live'] },
  242. ],
  243. eventFilters: [{ kind: 'surface', values: ['shadowed'] }],
  244. })
  245. expect(grouped.items).toHaveLength(1)
  246. expect(grouped.items[0]).toMatchObject({
  247. header: { id: SessionId('a'), cwd: '/a', parentSession: parent },
  248. live: true,
  249. persisted: false,
  250. bestMatch: { seq: 0, surface: 'shadowed' },
  251. })
  252. })
  253. it('searches at the supported FTS5 outer-predicate boundary in both scopes', async () => {
  254. const ctx = await liveContext()
  255. const session = ctx.sessions.create(SessionId('predicate-boundary'), {
  256. seed: messageEvents('needle'),
  257. meta: { cwd: '/work' },
  258. })
  259. const sessionFilters = Array.from(
  260. { length: 14 },
  261. () => ({ kind: 'cwd' as const, values: ['/work', null] }),
  262. )
  263. const eventFilters = Array.from(
  264. { length: 13 },
  265. () => ({ kind: 'type' as const, values: ['user/message' as const] }),
  266. )
  267. await expect(ctx.sessionQuery.searchSessions({ query: 'needle', sessionFilters }))
  268. .resolves.toMatchObject({ items: [{ header: { id: session.id } }] })
  269. await expect(ctx.sessionQuery.searchEvents({
  270. sessionId: session.id,
  271. query: 'needle',
  272. filters: eventFilters,
  273. })).resolves.toMatchObject({ items: [{ sessionId: session.id, seq: 0 }] })
  274. })
  275. it('rejects unsupported FTS5 outer-predicate counts with typed errors', async () => {
  276. const ctx = await liveContext()
  277. const session = ctx.sessions.create(SessionId('predicate-limit'), { seed: messageEvents('needle') })
  278. const sessionFilters = Array.from(
  279. { length: 1_100 },
  280. () => ({ kind: 'id' as const, values: [session.id] }),
  281. )
  282. const eventFilters = Array.from(
  283. { length: 1_100 },
  284. () => ({ kind: 'type' as const, values: ['user/message' as const] }),
  285. )
  286. await expect(ctx.sessionQuery.searchSessions({ query: 'needle', sessionFilters }))
  287. .rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  288. await expect(ctx.sessionQuery.searchEvents({
  289. sessionId: session.id,
  290. query: 'needle',
  291. filters: eventFilters,
  292. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  293. await expect(ctx.sessionQuery.searchSessions({
  294. query: 'needle',
  295. sessionFilters: sessionFilters.slice(0, 7),
  296. eventFilters: eventFilters.slice(0, 8),
  297. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  298. await expect(ctx.sessionQuery.searchEvents({
  299. sessionId: session.id,
  300. query: 'needle',
  301. filters: eventFilters.slice(0, 14),
  302. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  303. })
  304. it('uses literal phrase tokens, stable ties, and bounded Unicode snippets', async () => {
  305. const ctx = await liveContext({ path: ':memory:', defaultLimit: 10, maxLimit: 10, snippetChars: 5 })
  306. ctx.sessions.create(SessionId('a'), { seed: messageEvents('😀😀 alpha beta BRAID 😀😀', 10), meta: { createdAt: 1 } })
  307. ctx.sessions.create(SessionId('b'), { seed: messageEvents('alpha beta', 10), meta: { createdAt: 1 } })
  308. ctx.sessions.create(SessionId('c'), { seed: messageEvents('alpha middle beta', 10), meta: { createdAt: 1 } })
  309. ctx.sessions.create(SessionId('d'), { seed: messageEvents('alpha beta', 10), meta: { createdAt: 1 } })
  310. ctx.sessions.create(SessionId('operator'), { seed: messageEvents('needle OR absent', 10), meta: { createdAt: 1 } })
  311. ctx.sessions.create(SessionId('only'), { seed: messageEvents('needle only', 10), meta: { createdAt: 1 } })
  312. ctx.sessions.create(SessionId('quote'), { seed: messageEvents('say "needle" exactly', 10), meta: { createdAt: 1 } })
  313. const phrase = await ctx.sessionQuery.searchSessions({ query: 'alpha beta' })
  314. expect(phrase.items.map(item => item.header.id)).toEqual([SessionId('b'), SessionId('d'), SessionId('a')])
  315. expect(phrase.items.every(item => Array.from(item.bestMatch.snippet).length <= 5)).toBe(true)
  316. await expect(ctx.sessionQuery.searchSessions({ query: 'AI' })).resolves.toEqual({ items: [] })
  317. await expect(ctx.sessionQuery.searchSessions({ query: 'needle OR absent' }))
  318. .resolves.toMatchObject({ items: [{ header: { id: SessionId('operator') } }] })
  319. await expect(ctx.sessionQuery.searchSessions({ query: 'say "needle"' }))
  320. .resolves.toMatchObject({ items: [{ header: { id: SessionId('quote') } }] })
  321. await expect(ctx.sessionQuery.searchSessions({ query: '*' })).resolves.toEqual({ items: [] })
  322. })
  323. it('ranks live and persisted matches on one source-comparable contract', async () => {
  324. const persisted = header('z-persisted')
  325. TestPersistence.reset([
  326. { meta: persisted, events: messageEvents('needle needle', 10) },
  327. ...Array.from({ length: 12 }, (_, index) => ({
  328. meta: header(`filler-${index}`),
  329. events: messageEvents('needle', 10),
  330. })),
  331. ])
  332. const ctx = await liveContext()
  333. const persistence = await ctx.plugin(TestPersistence)
  334. ctx.sessions.create(SessionId('a-live'), {
  335. seed: messageEvents('needle needle', 10),
  336. meta: { createdAt: persisted.createdAt },
  337. })
  338. const result = await ctx.sessionQuery.searchSessions({
  339. query: 'needle',
  340. sessionFilters: [{ kind: 'id', values: [SessionId('a-live'), persisted.id] }],
  341. })
  342. expect(result.items.map(item => item.header.id)).toEqual([SessionId('a-live'), persisted.id])
  343. await persistence.dispose()
  344. })
  345. it('positions snippets from FTS5 matches across diacritics and punctuation', async () => {
  346. const ctx = await liveContext({ path: ':memory:', snippetChars: 14 })
  347. const session = ctx.sessions.create(SessionId('snippet'), {
  348. seed: messageEvents('long long long—café,\nnext value', 10),
  349. })
  350. const page = await ctx.sessionQuery.searchEvents({ sessionId: session.id, query: 'CAFE' })
  351. expect(page.items).toHaveLength(1)
  352. expect(page.items[0]!.snippet).toContain('café')
  353. expect(page.items[0]!.snippet).toContain('—')
  354. expect(page.items[0]!.snippet).not.toContain('\n')
  355. expect(Array.from(page.items[0]!.snippet).length).toBeLessThanOrEqual(14)
  356. })
  357. it('binds cursors to requests and only invalidates within-session pages for target changes', async () => {
  358. const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 5 })
  359. const target = ctx.sessions.create(SessionId('target'), {
  360. seed: [
  361. ...messageEvents('needle one', 10),
  362. { ...messageEvents('needle two', 11)[0]!, seq: 1 },
  363. { ...messageEvents('needle three', 12)[0]!, seq: 2 },
  364. ],
  365. })
  366. ctx.sessions.create(SessionId('other'), { seed: messageEvents('needle other', 10) })
  367. const eventPage = await ctx.sessionQuery.searchEvents({ sessionId: target.id, query: 'needle', limit: 1 })
  368. const sessionPage = await ctx.sessionQuery.searchSessions({ query: 'needle', limit: 1 })
  369. expect(eventPage.nextCursor).toEqual(expect.any(String))
  370. expect(sessionPage.nextCursor).toEqual(expect.any(String))
  371. if (eventPage.nextCursor === undefined || sessionPage.nextCursor === undefined) throw new Error('expected cursors')
  372. const unsafeOffsetCursor = replaceCursorOffset(eventPage.nextCursor, 1e100)
  373. await expect(ctx.sessionQuery.searchEvents({
  374. sessionId: target.id,
  375. query: 'needle',
  376. limit: 1,
  377. cursor: unsafeOffsetCursor,
  378. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CURSOR'))
  379. const eventKeys = eventPage.items.map(item => `${item.sessionId}:${item.seq}`)
  380. let eventCursor: ReturnType<typeof SessionSearchCursor> | undefined = eventPage.nextCursor
  381. while (eventCursor !== undefined) {
  382. const next = await ctx.sessionQuery.searchEvents({
  383. sessionId: target.id,
  384. query: 'needle',
  385. limit: 1,
  386. cursor: eventCursor,
  387. })
  388. eventKeys.push(...next.items.map(item => `${item.sessionId}:${item.seq}`))
  389. eventCursor = next.nextCursor
  390. }
  391. expect(eventKeys).toHaveLength(3)
  392. expect(new Set(eventKeys).size).toBe(eventKeys.length)
  393. const sessionIds = sessionPage.items.map(item => item.header.id)
  394. let sessionCursor: ReturnType<typeof SessionSearchCursor> | undefined = sessionPage.nextCursor
  395. while (sessionCursor !== undefined) {
  396. const next = await ctx.sessionQuery.searchSessions({ query: 'needle', limit: 1, cursor: sessionCursor })
  397. sessionIds.push(...next.items.map(item => item.header.id))
  398. sessionCursor = next.nextCursor
  399. }
  400. expect(sessionIds).toHaveLength(2)
  401. expect(new Set(sessionIds).size).toBe(sessionIds.length)
  402. ctx.sessions.create(SessionId('unrelated'), { seed: messageEvents('needle unrelated', 20) })
  403. await expect(ctx.sessionQuery.searchEvents({
  404. sessionId: target.id,
  405. query: 'needle',
  406. limit: 1,
  407. cursor: eventPage.nextCursor,
  408. })).resolves.toMatchObject({ items: [{ sessionId: target.id }] })
  409. await expect(ctx.sessionQuery.searchSessions({ query: 'needle', limit: 1, cursor: sessionPage.nextCursor }))
  410. .rejects.toThrow(expectCode('SESSION_QUERY_STALE_CURSOR'))
  411. await expect(ctx.sessionQuery.searchEvents({
  412. sessionId: target.id,
  413. query: 'different',
  414. limit: 1,
  415. cursor: eventPage.nextCursor,
  416. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CURSOR'))
  417. target.append('user/message', createUserMessage({
  418. content: [{ type: 'text', text: 'needle four' }], source: { kind: 'user' },
  419. }), { surfaceOp: 'append' })
  420. await expect(ctx.sessionQuery.searchEvents({
  421. sessionId: target.id,
  422. query: 'needle',
  423. limit: 1,
  424. cursor: eventPage.nextCursor,
  425. })).rejects.toThrow(expectCode('SESSION_QUERY_STALE_CURSOR'))
  426. })
  427. it('invalidates session cursors after transient persistence topology changes', async () => {
  428. TestPersistence.reset()
  429. const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 5 })
  430. ctx.sessions.create(SessionId('first'), { seed: messageEvents('needle first') })
  431. ctx.sessions.create(SessionId('second'), { seed: messageEvents('needle second') })
  432. const page = await ctx.sessionQuery.searchSessions({ query: 'needle', limit: 1 })
  433. if (page.nextCursor === undefined) throw new Error('expected cursor')
  434. const persistence = await ctx.plugin(TestPersistence)
  435. await persistence.dispose()
  436. await expect(ctx.sessionQuery.searchSessions({
  437. query: 'needle',
  438. limit: 1,
  439. cursor: page.nextCursor,
  440. })).rejects.toThrow(expectCode('SESSION_QUERY_STALE_CURSOR'))
  441. })
  442. it('rejects invalid requests, filters, cursors, and direct config', async () => {
  443. const ctx = await liveContext({ path: ':memory:', defaultLimit: 2, maxLimit: 3 })
  444. const session = ctx.sessions.create(SessionId('valid'), { seed: messageEvents('needle') })
  445. for (const request of [
  446. { sessionId: session.id, query: '' },
  447. { sessionId: session.id, query: 'needle', limit: 0 },
  448. { sessionId: session.id, query: 'needle', limit: 4 },
  449. { sessionId: session.id, query: 'needle', filters: [{ kind: 'seq', from: 2, to: 1 }] },
  450. { sessionId: session.id, query: 'needle', filters: [{ kind: 'surface', values: ['future'] }] },
  451. { sessionId: session.id, query: 'bad\0query' },
  452. ] as const) {
  453. await expect(ctx.sessionQuery.searchEvents(request as never)).rejects.toBeInstanceOf(Error)
  454. }
  455. await expect(ctx.sessionQuery.searchSessions({
  456. query: 'needle',
  457. sessionFilters: [{ kind: 'availability', values: ['remote' as never] }],
  458. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  459. await expect(ctx.sessionQuery.searchSessions({
  460. query: 'needle',
  461. sessionFilters: [{ kind: 'future' } as never],
  462. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  463. await expect(ctx.sessionQuery.searchSessions({
  464. query: 'needle',
  465. eventFilters: [{ kind: 'future' } as never],
  466. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  467. await expect(ctx.sessionQuery.searchEvents({
  468. sessionId: session.id,
  469. query: 'needle',
  470. filters: [{ kind: 'future' } as never],
  471. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  472. await expect(ctx.sessionQuery.searchEvents({
  473. sessionId: session.id,
  474. query: 'needle',
  475. cursor: SessionSearchCursor('not-json'),
  476. }))
  477. .rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CURSOR'))
  478. await expect(ctx.sessionQuery.searchEvents({ sessionId: SessionId('absent'), query: 'needle' }))
  479. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  480. for (const config of [
  481. { path: '' },
  482. { path: ':memory:', defaultLimit: 0 },
  483. { path: ':memory:', maxLimit: 0 },
  484. { path: ':memory:', defaultLimit: 1e100 },
  485. { path: ':memory:', maxLimit: 1e100 },
  486. { path: ':memory:', snippetChars: 0 },
  487. { path: ':memory:', readWindowMax: -1 },
  488. { path: ':memory:', persistedInspectConcurrency: 0 },
  489. { path: ':memory:', persistedInspectConcurrency: Number.MAX_SAFE_INTEGER + 1 },
  490. { path: ':memory:', defaultLimit: 3, maxLimit: 2 },
  491. { path: ':memory:', journalMode: 'memory' },
  492. ]) {
  493. const direct = new Context()
  494. await direct.plugin(SessionStore)
  495. expect(() => new SessionQuerySqlite(direct, config as never))
  496. .toThrow(expectCode('SESSION_QUERY_INVALID_CONFIG'))
  497. expect(direct.sessionQuery).toBeUndefined()
  498. }
  499. })
  500. it('rejects aggregate filter bindings above SQLite\'s portable variable limit', async () => {
  501. const ctx = await liveContext()
  502. const session = ctx.sessions.create(SessionId('binding-limit'), { seed: messageEvents('needle') })
  503. // Each clause is below the ceiling; combined with its sibling and fixed
  504. // query bindings, the complete statement is not portable.
  505. const halfPortableLimit = 16_383
  506. const ids = Array.from(
  507. { length: halfPortableLimit },
  508. (_, index) => SessionId(`binding-${index}`),
  509. )
  510. const types = Array.from({ length: halfPortableLimit }, () => 'user/message' as const)
  511. const surfaces = Array.from({ length: halfPortableLimit }, () => 'current' as const)
  512. await expect(ctx.sessionQuery.searchSessions({
  513. query: 'needle',
  514. sessionFilters: [{ kind: 'id', values: ids }],
  515. eventFilters: [{ kind: 'type', values: types }],
  516. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  517. await expect(ctx.sessionQuery.searchEvents({
  518. sessionId: session.id,
  519. query: 'needle',
  520. filters: [
  521. { kind: 'type', values: types },
  522. { kind: 'surface', values: surfaces },
  523. ],
  524. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  525. })
  526. it('rejects one 125,000-value filter list with a typed error', async () => {
  527. const ctx = await liveContext()
  528. const ids = Array.from(
  529. { length: 125_000 },
  530. (_, index) => SessionId(`oversized-binding-${index}`),
  531. )
  532. await expect(ctx.sessionQuery.searchSessions({
  533. query: 'needle',
  534. sessionFilters: [{ kind: 'id', values: ids }],
  535. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  536. })
  537. })
  538. describe('SQLite reconciliation and source lifecycle', () => {
  539. it('owns queued request and filter values before waiting for the serializer', async () => {
  540. const durable = header('owned')
  541. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  542. const ctx = await liveContext()
  543. const persistence = await ctx.plugin(TestPersistence)
  544. let release!: () => void
  545. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  546. let markStarted!: () => void
  547. const started = new Promise<void>((resolve) => { markStarted = resolve })
  548. TestPersistence.listStarted = () => {
  549. TestPersistence.listStarted = undefined
  550. markStarted()
  551. }
  552. const blocking = ctx.sessionQuery.searchSessions({ query: 'needle' })
  553. await started
  554. const availability: SessionAvailability[] = ['persisted']
  555. const request: SessionSearchRequest = {
  556. query: 'needle',
  557. sessionFilters: [{ kind: 'availability', values: availability }],
  558. }
  559. const queued = ctx.sessionQuery.searchSessions(request)
  560. request.query = 'absent'
  561. availability[0] = 'live'
  562. release()
  563. await expect(blocking).resolves.toMatchObject({ items: [{ header: durable }] })
  564. await expect(queued).resolves.toMatchObject({ items: [{ header: durable }] })
  565. await persistence.dispose()
  566. })
  567. it('mounts persistence dynamically, shadows with TEMP live rows, reveals, and hides on unmount', async () => {
  568. const shared = header('shared', 10, { cwd: '/work' })
  569. const durable = header('durable', 5)
  570. TestPersistence.reset([
  571. { meta: shared, events: messageEvents('persisted needle') },
  572. { meta: durable, events: messageEvents('durable needle') },
  573. ])
  574. const ctx = await liveContext()
  575. await expect(ctx.sessionQuery.searchSessions({ query: 'durable' })).resolves.toEqual({ items: [] })
  576. const persistenceFiber = await ctx.plugin(TestPersistence)
  577. await expect(ctx.sessionQuery.searchSessions({ query: 'durable' }))
  578. .resolves.toMatchObject({ items: [{ header: durable, live: false, persisted: true }] })
  579. const live = ctx.sessions.prepare(shared.id, { meta: { createdAt: 10, cwd: '/work' } })
  580. live.append('user/message', createUserMessage({
  581. content: [{ type: 'text', text: 'live needle' }], source: { kind: 'user' },
  582. }), { surfaceOp: 'append' })
  583. const detach = ctx.sessions.enter(live)
  584. ctx.sessions.announce(live)
  585. await expect(ctx.sessionQuery.searchSessions({ query: 'persisted' })).resolves.toEqual({ items: [] })
  586. await expect(ctx.sessionQuery.searchSessions({ query: 'live' }))
  587. .resolves.toMatchObject({ items: [{ header: shared, live: true, persisted: true }] })
  588. detach()
  589. await expect(ctx.sessionQuery.searchSessions({ query: 'persisted' }))
  590. .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
  591. await persistenceFiber.dispose()
  592. await expect(ctx.sessionQuery.searchSessions({ query: 'durable' })).resolves.toEqual({ items: [] })
  593. await expect(ctx.sessionQuery.searchEvents({ sessionId: durable.id, query: 'needle' }))
  594. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  595. })
  596. it('does not load a persisted log while the same session is live', async () => {
  597. const shared = header('checkpointed-live', 10)
  598. TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
  599. const ctx = await liveContext()
  600. const live = ctx.sessions.prepare(shared.id, {
  601. seed: messageEvents('live needle'),
  602. meta: { createdAt: shared.createdAt },
  603. })
  604. const detach = ctx.sessions.enter(live)
  605. ctx.sessions.announce(live)
  606. const persistence = await ctx.plugin(TestPersistence)
  607. await expect(ctx.sessionQuery.searchSessions({
  608. query: 'live',
  609. sessionFilters: [{ kind: 'availability', values: ['persisted'] }],
  610. })).resolves.toMatchObject({
  611. items: [{ header: shared, live: true, persisted: true }],
  612. })
  613. expect(TestPersistence.loads.get(shared.id)).toBeUndefined()
  614. expect(TestPersistence.inspections.get(shared.id)).toBeUndefined()
  615. detach()
  616. await expect(ctx.sessionQuery.searchSessions({ query: 'persisted' }))
  617. .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
  618. expect(TestPersistence.loads.get(shared.id)).toBeUndefined()
  619. expect(TestPersistence.inspections.get(shared.id)).toBe(1)
  620. await persistence.dispose()
  621. })
  622. it('retries when a live owner attaches during persistence observation', async () => {
  623. TestPersistence.reset()
  624. const ctx = await liveContext()
  625. await ctx.plugin(TestPersistence)
  626. TestPersistence.snapshotEffect = () => {
  627. TestPersistence.snapshotEffect = undefined
  628. ctx.sessions.create(SessionId('attached'), { seed: messageEvents('attached needle') })
  629. }
  630. await expect(ctx.sessionQuery.searchSessions({ query: 'attached' }))
  631. .resolves.toMatchObject({ items: [{ header: { id: SessionId('attached') } }] })
  632. })
  633. it('cannot crash-repair a log when live ownership begins during persisted inspection', async () => {
  634. const shared = header('attach-during-inspect', 10)
  635. const persistedEvents = messageEvents('persisted needle')
  636. TestPersistence.reset([{ meta: shared, events: persistedEvents }])
  637. const ctx = await liveContext()
  638. await ctx.plugin(TestPersistence)
  639. TestPersistence.loadEffect = (entry) => {
  640. entry.events = messageEvents('incorrect repair')
  641. }
  642. TestPersistence.inspectEffect = () => {
  643. ctx.sessions.create(shared.id, {
  644. seed: messageEvents('live needle'),
  645. meta: { createdAt: shared.createdAt },
  646. })
  647. }
  648. await expect(ctx.sessionQuery.searchSessions({ query: 'live' }))
  649. .resolves.toMatchObject({ items: [{ header: shared, live: true, persisted: true }] })
  650. expect(TestPersistence.loads.get(shared.id)).toBeUndefined()
  651. expect(TestPersistence.entries.get(shared.id)?.events).toEqual(persistedEvents)
  652. })
  653. it('retries when one live owner replaces another during persistence observation', async () => {
  654. TestPersistence.reset()
  655. const ctx = await liveContext()
  656. const first = ctx.sessions.prepare(SessionId('first'), { seed: messageEvents('first needle') })
  657. const detachFirst = ctx.sessions.enter(first)
  658. ctx.sessions.announce(first)
  659. await ctx.plugin(TestPersistence)
  660. TestPersistence.snapshotEffect = () => {
  661. TestPersistence.snapshotEffect = undefined
  662. detachFirst()
  663. ctx.sessions.create(SessionId('second'), { seed: messageEvents('second needle') })
  664. }
  665. await expect(ctx.sessionQuery.searchSessions({ query: 'second' }))
  666. .resolves.toMatchObject({ items: [{ header: { id: SessionId('second') } }] })
  667. })
  668. it('uses the reconciled persistence binding through the query boundary', async () => {
  669. const durable = header('post-reconcile-unmount')
  670. TestPersistence.reset([{ meta: durable, events: [
  671. ...messageEvents('durable needle', 1),
  672. { ...messageEvents('durable needle again', 2)[0]!, seq: 1 },
  673. ] }])
  674. const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 2 })
  675. const persistence = await ctx.plugin(TestPersistence)
  676. const internals = ctx.sessionQuery as unknown as {
  677. _reconcile(signal: AbortSignal | undefined): Promise<{
  678. identity: symbol
  679. service?: SessionPersistence
  680. }>
  681. }
  682. const reconcile = internals._reconcile.bind(internals)
  683. const boundary = vi.spyOn(internals, '_reconcile').mockImplementation(async (signal) => {
  684. const binding = await reconcile(signal)
  685. await persistence.dispose()
  686. return binding
  687. })
  688. const page = await ctx.sessionQuery.searchEvents({
  689. sessionId: durable.id,
  690. query: 'needle',
  691. limit: 1,
  692. })
  693. expect(page.items).toMatchObject([{ sessionId: durable.id }])
  694. expect(page.nextCursor).toEqual(expect.any(String))
  695. boundary.mockRestore()
  696. await expect(ctx.sessionQuery.searchEvents({ sessionId: durable.id, query: 'needle' }))
  697. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  698. })
  699. it('discards a stale list rejection when persistence unmounts during observation', async () => {
  700. const durable = header('racing')
  701. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  702. const ctx = await liveContext()
  703. const persistenceFiber = await ctx.plugin(TestPersistence)
  704. let release!: () => void
  705. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  706. let markStarted!: () => void
  707. const started = new Promise<void>((resolve) => { markStarted = resolve })
  708. TestPersistence.listStarted = () => {
  709. TestPersistence.listStarted = undefined
  710. markStarted()
  711. }
  712. const search = ctx.sessionQuery.searchSessions({ query: 'needle' })
  713. await started
  714. await persistenceFiber.dispose()
  715. TestPersistence.failure = new Error('stale backend rejection')
  716. release()
  717. await expect(search).resolves.toEqual({ items: [] })
  718. })
  719. it('retries against a replacement after the prior binding rejects', async () => {
  720. const durable = header('replacement')
  721. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  722. const ctx = await liveContext()
  723. const prior = await ctx.plugin(TestPersistence)
  724. let rejectPrior!: (reason: unknown) => void
  725. TestPersistence.listGate = new Promise<void>((_resolve, reject) => { rejectPrior = reject })
  726. let markStarted!: () => void
  727. const started = new Promise<void>((resolve) => { markStarted = resolve })
  728. TestPersistence.listStarted = () => {
  729. TestPersistence.listStarted = undefined
  730. markStarted()
  731. }
  732. const search = ctx.sessionQuery.searchSessions({ query: 'needle' })
  733. await started
  734. await prior.dispose()
  735. TestPersistence.listGate = undefined
  736. const replacement = await ctx.plugin(TestPersistence)
  737. rejectPrior(new Error('stale prior binding'))
  738. await expect(search).resolves.toMatchObject({ items: [{ header: durable }] })
  739. await replacement.dispose()
  740. })
  741. it('reloads a replacement source even when its opaque revisions collide', async () => {
  742. const durable = header('colliding-replacement')
  743. TestPersistence.reset([{ meta: durable, events: messageEvents('old content') }])
  744. const revision = TestPersistence.revisions.get(durable.id)!
  745. const ctx = await liveContext()
  746. const prior = await ctx.plugin(TestPersistence)
  747. await expect(ctx.sessionQuery.searchSessions({ query: 'old' }))
  748. .resolves.toMatchObject({ items: [{ header: durable }] })
  749. await prior.dispose()
  750. TestPersistence.set({ meta: durable, events: messageEvents('new needle') })
  751. TestPersistence.revisions.set(durable.id, revision)
  752. const replacement = await ctx.plugin(TestPersistence)
  753. const page = await ctx.sessionQuery.searchSessions({ query: 'new needle' })
  754. expect(TestPersistence.inspections.get(durable.id)).toBe(2)
  755. expect(page).toMatchObject({ items: [{ header: durable }] })
  756. await expect(ctx.sessionQuery.searchSessions({ query: 'old' })).resolves.toEqual({ items: [] })
  757. expect(TestPersistence.inspections.get(durable.id)).toBe(2)
  758. await replacement.dispose()
  759. })
  760. it('retries when a successful observation belongs to a source unmounted during listing', async () => {
  761. const durable = header('successful-unmount')
  762. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  763. const ctx = await liveContext()
  764. const persistence = await ctx.plugin(TestPersistence)
  765. let lists = 0
  766. TestPersistence.snapshotEffect = async () => {
  767. lists += 1
  768. if (lists === 2) await persistence.dispose()
  769. }
  770. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' })).resolves.toEqual({ items: [] })
  771. expect(lists).toBe(2)
  772. })
  773. it('retries when the snapshot population changes during observation', async () => {
  774. const first = header('first')
  775. const added = header('added-during-list')
  776. TestPersistence.reset([{ meta: first, events: messageEvents('first needle') }])
  777. const ctx = await liveContext()
  778. await ctx.plugin(TestPersistence)
  779. TestPersistence.snapshotEffect = () => {
  780. TestPersistence.snapshotEffect = undefined
  781. TestPersistence.set({ meta: added, events: messageEvents('added needle') })
  782. }
  783. const page = await ctx.sessionQuery.searchSessions({ query: 'needle' })
  784. expect(page.items.map(item => item.header.id).sort()).toEqual([added.id, first.id].sort())
  785. expect(TestPersistence.inspections.get(first.id)).toBe(2)
  786. expect(TestPersistence.inspections.get(added.id)).toBe(1)
  787. })
  788. it('fails after one retry when persistence snapshots keep changing', async () => {
  789. const durable = header('continuous-mutation')
  790. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  791. const ctx = await liveContext()
  792. await ctx.plugin(TestPersistence)
  793. let lists = 0
  794. TestPersistence.snapshotEffect = () => {
  795. lists += 1
  796. TestPersistence.set({ meta: durable, events: messageEvents(`durable needle ${lists}`) })
  797. }
  798. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }))
  799. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  800. expect(lists).toBe(4)
  801. })
  802. it('retries if the persistence binding changes while live sessions are observed', async () => {
  803. const durable = header('live-boundary-retry')
  804. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  805. const ctx = await liveContext()
  806. await ctx.plugin(TestPersistence)
  807. const internals = ctx.sessionQuery as unknown as {
  808. _persistenceBinding: { identity: symbol; service?: SessionPersistence }
  809. }
  810. const originalList = ctx.sessions.list.bind(ctx.sessions)
  811. let bumped = false
  812. const list = vi.spyOn(ctx.sessions, 'list').mockImplementation(() => {
  813. if (!bumped) {
  814. bumped = true
  815. internals._persistenceBinding = {
  816. ...internals._persistenceBinding,
  817. identity: Symbol(),
  818. }
  819. }
  820. return originalList()
  821. })
  822. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }))
  823. .resolves.toMatchObject({ items: [{ header: durable }] })
  824. expect(TestPersistence.inspections.get(durable.id)).toBe(2)
  825. list.mockRestore()
  826. })
  827. it('rejects malformed snapshots and preserves typed persistence failures', async () => {
  828. const durable = header('invalid-snapshot')
  829. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  830. const ctx = await liveContext()
  831. await ctx.plugin(TestPersistence)
  832. TestPersistence.snapshotOverride = () => 'not-an-array' as never
  833. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }))
  834. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  835. TestPersistence.snapshotOverride = () => [{ header: durable, revision: 1 as never }]
  836. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }))
  837. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  838. TestPersistence.snapshotOverride = () => [
  839. { header: durable, revision: SessionPersistenceRevision('duplicate:1') },
  840. { header: durable, revision: SessionPersistenceRevision('duplicate:2') },
  841. ]
  842. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }))
  843. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  844. TestPersistence.snapshotOverride = undefined
  845. const typed = new SessionQueryError('typed persistence failure', 'SESSION_QUERY_PERSISTENCE_FAILED')
  846. TestPersistence.failure = typed
  847. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' })).rejects.toBe(typed)
  848. })
  849. it('rejects immutable header conflicts between live and persisted sources', async () => {
  850. const shared = header('conflict', 10, { delegationDepth: 1 })
  851. TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
  852. const ctx = await liveContext()
  853. await ctx.plugin(TestPersistence)
  854. ctx.sessions.create(shared.id, {
  855. seed: messageEvents('live needle'),
  856. meta: { createdAt: 10, delegationDepth: 2 },
  857. })
  858. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }))
  859. .rejects.toThrow(expectCode('SESSION_QUERY_SOURCE_CONFLICT'))
  860. })
  861. it('preserves unchanged persisted generations while reconciling new, changed, and deleted rows', async () => {
  862. const path = await temporaryPath()
  863. const unchanged = header('unchanged')
  864. const changed = header('changed')
  865. const deleted = header('deleted')
  866. TestPersistence.reset([
  867. { meta: unchanged, events: messageEvents('unchanged needle') },
  868. { meta: changed, events: messageEvents('old needle') },
  869. { meta: deleted, events: messageEvents('deleted needle') },
  870. ])
  871. const first = new Context()
  872. await first.plugin(SessionStore)
  873. const firstPersistence = await first.plugin(TestPersistence)
  874. const firstSearch = await first.plugin(SessionQuerySqlite, { path })
  875. await first.sessionQuery.searchSessions({ query: 'needle' })
  876. expect(Object.fromEntries(TestPersistence.inspections)).toEqual({ unchanged: 1, changed: 1, deleted: 1 })
  877. await first.sessionQuery.searchSessions({ query: 'needle' })
  878. expect(Object.fromEntries(TestPersistence.inspections)).toEqual({ unchanged: 1, changed: 1, deleted: 1 })
  879. await firstSearch.dispose()
  880. await firstPersistence.dispose()
  881. const beforeDb = new DatabaseSync(path)
  882. const beforeRows = beforeDb.prepare('SELECT id, generation FROM persisted_sessions ORDER BY id').all() as Array<{ id: string; generation: number }>
  883. beforeDb.close()
  884. const before = new Map(beforeRows.map(row => [row.id, row.generation]))
  885. const added = header('added')
  886. TestPersistence.entries.delete(deleted.id)
  887. TestPersistence.set({ meta: changed, events: messageEvents('changed needle') })
  888. TestPersistence.set({ meta: added, events: messageEvents('added needle') })
  889. const second = new Context()
  890. await second.plugin(SessionStore)
  891. const secondPersistence = await second.plugin(TestPersistence)
  892. const secondSearch = await second.plugin(SessionQuerySqlite, { path })
  893. const result = await second.sessionQuery.searchSessions({ query: 'needle' })
  894. expect(result.items.map(item => item.header.id).sort()).toEqual([added.id, changed.id, unchanged.id].sort())
  895. expect(Object.fromEntries(TestPersistence.inspections)).toEqual({
  896. unchanged: 1,
  897. changed: 2,
  898. deleted: 1,
  899. added: 1,
  900. })
  901. await secondSearch.dispose()
  902. await secondPersistence.dispose()
  903. const afterDb = new DatabaseSync(path)
  904. const afterRows = afterDb.prepare('SELECT id, generation FROM persisted_sessions ORDER BY id').all() as Array<{ id: string; generation: number }>
  905. afterDb.close()
  906. const after = new Map(afterRows.map(row => [row.id, row.generation]))
  907. expect(after.get(unchanged.id)).toBe(before.get(unchanged.id))
  908. expect(after.get(changed.id)).toBeGreaterThan(before.get(changed.id)!)
  909. expect(after.has(deleted.id)).toBe(false)
  910. expect(after.has(added.id)).toBe(true)
  911. })
  912. it('drops connection-local live overlays on reopen and retains persistent bases', async () => {
  913. const path = await temporaryPath()
  914. const shared = header('shared', 10)
  915. TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
  916. const first = new Context()
  917. await first.plugin(SessionStore)
  918. const persistence = await first.plugin(TestPersistence)
  919. const live = first.sessions.create(shared.id, { seed: messageEvents('live needle'), meta: { createdAt: 10 } })
  920. const search = await first.plugin(SessionQuerySqlite, { path })
  921. await expect(first.sessionQuery.searchEvents({ sessionId: live.id, query: 'live' })).resolves.toMatchObject({ items: [{}] })
  922. await search.dispose()
  923. await persistence.dispose()
  924. const second = new Context()
  925. await second.plugin(SessionStore)
  926. const persistenceAgain = await second.plugin(TestPersistence)
  927. const searchAgain = await second.plugin(SessionQuerySqlite, { path })
  928. await expect(second.sessionQuery.searchSessions({ query: 'live' })).resolves.toEqual({ items: [] })
  929. await expect(second.sessionQuery.searchSessions({ query: 'persisted' }))
  930. .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
  931. expect(TestPersistence.inspections.get(shared.id)).toBe(1)
  932. await searchAgain.dispose()
  933. await persistenceAgain.dispose()
  934. })
  935. it('refreshes after an external mutating load repair without loading from the query path', async () => {
  936. const durable = header('repair')
  937. TestPersistence.reset([{ meta: durable, events: messageEvents('before repair') }])
  938. const ctx = await liveContext()
  939. const persistence = await ctx.plugin(TestPersistence)
  940. await expect(ctx.sessionQuery.searchSessions({ query: 'before' }))
  941. .resolves.toMatchObject({ items: [{ header: durable }] })
  942. TestPersistence.loadEffect = (entry) => {
  943. entry.events = messageEvents('repaired needle')
  944. }
  945. await ctx.sessionPersistence.load(durable.id)
  946. await expect(ctx.sessionQuery.searchSessions({ query: 'repaired' }))
  947. .resolves.toMatchObject({ items: [{ header: durable }] })
  948. expect(TestPersistence.inspections.get(durable.id)).toBe(2)
  949. await ctx.sessionQuery.searchSessions({ query: 'repaired' })
  950. expect(TestPersistence.inspections.get(durable.id)).toBe(2)
  951. expect(TestPersistence.loads.get(durable.id)).toBe(1)
  952. await persistence.dispose()
  953. })
  954. it('recovers on the next search after source and SQLite transaction failures', async () => {
  955. TestPersistence.reset([{ meta: header('durable'), events: messageEvents('durable needle') }])
  956. const ctx = await liveContext()
  957. await ctx.plugin(TestPersistence)
  958. TestPersistence.failure = 'offline'
  959. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }))
  960. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  961. const signal = new AbortController().signal
  962. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal }))
  963. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  964. TestPersistence.failure = new Error('still offline')
  965. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal }))
  966. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  967. TestPersistence.failure = undefined
  968. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' })).resolves.toMatchObject({ items: [{}] })
  969. const live = ctx.sessions.create(SessionId('live'), { seed: messageEvents('base') })
  970. await ctx.sessionQuery.searchEvents({ sessionId: live.id, query: 'base' })
  971. const db = (ctx.sessionQuery as unknown as { _db: DatabaseSync })._db
  972. db.exec('PRAGMA query_only = ON')
  973. live.append('user/message', createUserMessage({
  974. content: [{ type: 'text', text: 'retry needle' }], source: { kind: 'user' },
  975. }), { surfaceOp: 'append' })
  976. await expect(ctx.sessionQuery.searchEvents({ sessionId: live.id, query: 'needle' }))
  977. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  978. db.exec('PRAGMA query_only = OFF')
  979. await expect(ctx.sessionQuery.searchEvents({ sessionId: live.id, query: 'needle' }))
  980. .resolves.toMatchObject({ items: [{ seq: 1 }] })
  981. })
  982. })
  983. describe('SQLite schema, cancellation, and real persistence integration', () => {
  984. it('creates a new database and WAL sidecars owner-only without changing its parent mode', async () => {
  985. if (process.platform === 'win32') return
  986. const path = await temporaryPath()
  987. const directory = dirname(path)
  988. await chmod(directory, 0o755)
  989. const ctx = await liveContext({ path })
  990. await ctx.sessionQuery.searchSessions({ query: 'needle' })
  991. expect((await stat(directory)).mode & 0o777).toBe(0o755)
  992. expect((await stat(path)).mode & 0o777).toBe(0o600)
  993. expect((await stat(`${path}-wal`)).mode & 0o777).toBe(0o600)
  994. expect((await stat(`${path}-shm`)).mode & 0o777).toBe(0o600)
  995. await (ctx.sessionQuery as SessionQuerySqlite).close()
  996. })
  997. it('creates a persistent rollback journal owner-only', async () => {
  998. if (process.platform === 'win32') return
  999. const path = await temporaryPath()
  1000. const ctx = await liveContext({ path, journalMode: 'persist' })
  1001. await ctx.sessionQuery.searchSessions({ query: 'needle' })
  1002. expect((await stat(path)).mode & 0o777).toBe(0o600)
  1003. expect((await stat(`${path}-journal`)).mode & 0o777).toBe(0o600)
  1004. await (ctx.sessionQuery as SessionQuerySqlite).close()
  1005. })
  1006. it('preserves the mode of an existing database file', async () => {
  1007. if (process.platform === 'win32') return
  1008. const path = await temporaryPath()
  1009. await writeFile(path, '', { mode: 0o644 })
  1010. await chmod(path, 0o644)
  1011. const ctx = await liveContext({ path, journalMode: 'delete' })
  1012. await ctx.sessionQuery.searchSessions({ query: 'needle' })
  1013. expect((await stat(path)).mode & 0o777).toBe(0o644)
  1014. await (ctx.sessionQuery as SessionQuerySqlite).close()
  1015. })
  1016. it('surfaces filesystem failures while pre-creating the database', async () => {
  1017. const path = `${await temporaryPath()}\0`
  1018. const ctx = new Context()
  1019. await ctx.plugin(SessionStore)
  1020. await expect(ctx.plugin(SessionQuerySqlite, { path })).rejects.toMatchObject({
  1021. code: 'SESSION_QUERY_INDEX_FAILED',
  1022. cause: { code: 'ERR_INVALID_ARG_VALUE' },
  1023. })
  1024. expect(ctx.sessionQuery).toBeUndefined()
  1025. })
  1026. it('resets a recognized incompatible schema but refuses unknown or foreign tables', async () => {
  1027. const stalePath = await temporaryPath('stale.db')
  1028. const staleOwner = await liveContext({ path: stalePath })
  1029. await (staleOwner.sessionQuery as SessionQuerySqlite).close()
  1030. const stale = new DatabaseSync(stalePath)
  1031. stale.exec('PRAGMA user_version = 999')
  1032. stale.close()
  1033. const staleCtx = await liveContext({ path: stalePath })
  1034. staleCtx.sessions.create(SessionId('live'), { seed: messageEvents('needle') })
  1035. await staleCtx.sessionQuery.searchSessions({ query: 'needle' })
  1036. await (staleCtx.sessionQuery as SessionQuerySqlite).close()
  1037. const rebuilt = new DatabaseSync(stalePath)
  1038. expect((rebuilt.prepare('PRAGMA user_version').get() as { user_version: number }).user_version)
  1039. .toBe(SESSION_QUERY_SQLITE_SCHEMA_VERSION)
  1040. rebuilt.close()
  1041. const augmentedPath = await temporaryPath('augmented.db')
  1042. const augmentedOwner = await liveContext({ path: augmentedPath })
  1043. await (augmentedOwner.sessionQuery as SessionQuerySqlite).close()
  1044. const augmented = new DatabaseSync(augmentedPath)
  1045. augmented.exec('CREATE TABLE unrelated(value TEXT)')
  1046. augmented.exec("INSERT INTO unrelated VALUES ('safe')")
  1047. augmented.exec('PRAGMA user_version = 999')
  1048. augmented.close()
  1049. const augmentedCtx = new Context()
  1050. await augmentedCtx.plugin(SessionStore)
  1051. await expect(augmentedCtx.plugin(SessionQuerySqlite, { path: augmentedPath }))
  1052. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  1053. expect(augmentedCtx.sessionQuery).toBeUndefined()
  1054. const stillAugmented = new DatabaseSync(augmentedPath)
  1055. expect(stillAugmented.prepare('SELECT value FROM unrelated').get()).toEqual({ value: 'safe' })
  1056. expect(stillAugmented.prepare('PRAGMA user_version').get()).toEqual({ user_version: 999 })
  1057. stillAugmented.close()
  1058. const currentAugmentedPath = await temporaryPath('current-augmented.db')
  1059. const currentAugmentedOwner = await liveContext({ path: currentAugmentedPath })
  1060. await (currentAugmentedOwner.sessionQuery as SessionQuerySqlite).close()
  1061. const currentAugmented = new DatabaseSync(currentAugmentedPath)
  1062. currentAugmented.exec('CREATE TABLE unrelated(value TEXT)')
  1063. currentAugmented.exec("INSERT INTO unrelated VALUES ('safe')")
  1064. currentAugmented.close()
  1065. const currentAugmentedCtx = new Context()
  1066. await currentAugmentedCtx.plugin(SessionStore)
  1067. await expect(currentAugmentedCtx.plugin(SessionQuerySqlite, {
  1068. path: currentAugmentedPath,
  1069. journalMode: 'delete',
  1070. })).rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  1071. expect(currentAugmentedCtx.sessionQuery).toBeUndefined()
  1072. const stillCurrentAugmented = new DatabaseSync(currentAugmentedPath)
  1073. expect(stillCurrentAugmented.prepare('SELECT value FROM unrelated').get()).toEqual({ value: 'safe' })
  1074. expect(stillCurrentAugmented.prepare('PRAGMA user_version').get())
  1075. .toEqual({ user_version: SESSION_QUERY_SQLITE_SCHEMA_VERSION })
  1076. expect(stillCurrentAugmented.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'wal' })
  1077. stillCurrentAugmented.close()
  1078. const foreignPath = await temporaryPath('foreign.db')
  1079. const foreign = new DatabaseSync(foreignPath)
  1080. foreign.exec('PRAGMA journal_mode = WAL')
  1081. foreign.exec('CREATE TABLE canonical(value TEXT)')
  1082. foreign.exec("INSERT INTO canonical VALUES ('safe')")
  1083. foreign.close()
  1084. const foreignCtx = new Context()
  1085. await foreignCtx.plugin(SessionStore)
  1086. await expect(foreignCtx.plugin(SessionQuerySqlite, { path: foreignPath, journalMode: 'delete' }))
  1087. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  1088. expect(foreignCtx.sessionQuery).toBeUndefined()
  1089. const stillForeign = new DatabaseSync(foreignPath)
  1090. expect(stillForeign.prepare('SELECT value FROM canonical').get()).toEqual({ value: 'safe' })
  1091. expect(stillForeign.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'wal' })
  1092. stillForeign.close()
  1093. const wildcardPath = await temporaryPath('sqlite-wildcard.db')
  1094. const wildcard = new DatabaseSync(wildcardPath)
  1095. wildcard.exec('PRAGMA journal_mode = WAL')
  1096. wildcard.exec('CREATE TABLE sqliteX(value TEXT)')
  1097. wildcard.exec("INSERT INTO sqliteX VALUES ('safe')")
  1098. wildcard.close()
  1099. const wildcardCtx = new Context()
  1100. await wildcardCtx.plugin(SessionStore)
  1101. await expect(wildcardCtx.plugin(SessionQuerySqlite, {
  1102. path: wildcardPath,
  1103. journalMode: 'delete',
  1104. })).rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  1105. expect(wildcardCtx.sessionQuery).toBeUndefined()
  1106. const stillWildcard = new DatabaseSync(wildcardPath)
  1107. expect(stillWildcard.prepare('SELECT value FROM sqliteX').get()).toEqual({ value: 'safe' })
  1108. expect(stillWildcard.prepare('PRAGMA application_id').get()).toEqual({ application_id: 0 })
  1109. expect(stillWildcard.prepare('PRAGMA user_version').get()).toEqual({ user_version: 0 })
  1110. expect(stillWildcard.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'wal' })
  1111. stillWildcard.close()
  1112. const otherAppPath = await temporaryPath('other-app.db')
  1113. const otherApp = new DatabaseSync(otherAppPath)
  1114. otherApp.exec('PRAGMA application_id = 123')
  1115. otherApp.close()
  1116. const otherAppCtx = new Context()
  1117. await otherAppCtx.plugin(SessionStore)
  1118. await expect(otherAppCtx.plugin(SessionQuerySqlite, { path: otherAppPath }))
  1119. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  1120. expect(otherAppCtx.sessionQuery).toBeUndefined()
  1121. })
  1122. it('fails plugin initialization without an unhandled rejection or partial service', async () => {
  1123. const path = await temporaryPath('never-queried.db')
  1124. const foreign = new DatabaseSync(path)
  1125. foreign.exec('CREATE TABLE canonical(value TEXT)')
  1126. foreign.close()
  1127. const unhandled: unknown[] = []
  1128. const onUnhandled = (reason: unknown) => { unhandled.push(reason) }
  1129. process.on('unhandledRejection', onUnhandled)
  1130. try {
  1131. const ctx = new Context()
  1132. await ctx.plugin(SessionStore)
  1133. await expect(ctx.plugin(SessionQuerySqlite, { path }))
  1134. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  1135. await new Promise<void>((resolve) => { setImmediate(resolve) })
  1136. expect(unhandled).toEqual([])
  1137. expect(ctx.sessionQuery).toBeUndefined()
  1138. } finally {
  1139. process.off('unhandledRejection', onUnhandled)
  1140. }
  1141. })
  1142. it.each(['sessions', 'events'] as const)(
  1143. 'forwards one exact reconciliation signal through both snapshot lists and persisted inspection for %s search',
  1144. async (scope) => {
  1145. const durable = header(`signal-${scope}`)
  1146. TestPersistence.reset([{ meta: durable, events: messageEvents('signal needle') }])
  1147. const ctx = await liveContext()
  1148. await ctx.plugin(TestPersistence)
  1149. const controller = new AbortController()
  1150. const result = scope === 'sessions'
  1151. ? await ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: controller.signal })
  1152. : await ctx.sessionQuery.searchEvents(
  1153. { sessionId: durable.id, query: 'needle' },
  1154. { signal: controller.signal },
  1155. )
  1156. expect(result.items).toHaveLength(1)
  1157. expect(TestPersistence.snapshotSignals).toEqual([controller.signal, controller.signal])
  1158. expect(TestPersistence.inspectSignals).toEqual([controller.signal])
  1159. },
  1160. )
  1161. it.each(['sessions', 'events'] as const)(
  1162. 'starts no persistence observation for a pre-aborted %s search',
  1163. async (scope) => {
  1164. const durable = header(`pre-aborted-${scope}`)
  1165. TestPersistence.reset([{ meta: durable, events: messageEvents('needle') }])
  1166. const ctx = await liveContext()
  1167. await ctx.plugin(TestPersistence)
  1168. const controller = new AbortController()
  1169. controller.abort(new Error(`pre-aborted ${scope}`))
  1170. const pending = scope === 'sessions'
  1171. ? ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: controller.signal })
  1172. : ctx.sessionQuery.searchEvents(
  1173. { sessionId: durable.id, query: 'needle' },
  1174. { signal: controller.signal },
  1175. )
  1176. await expect(pending).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1177. expect(TestPersistence.snapshotSignals).toEqual([])
  1178. expect(TestPersistence.inspectSignals).toEqual([])
  1179. },
  1180. )
  1181. it('awaits cooperative snapshot-list cancellation cleanup without starting another observation step', async () => {
  1182. const durable = header('cooperative-list-abort')
  1183. TestPersistence.reset([{ meta: durable, events: messageEvents('needle') }])
  1184. const ctx = await liveContext()
  1185. await ctx.plugin(TestPersistence)
  1186. const started = Promise.withResolvers<AbortSignal>()
  1187. const abortObserved = Promise.withResolvers<undefined>()
  1188. const cleanup = Promise.withResolvers<undefined>()
  1189. TestPersistence.snapshotEffect = async (signal) => {
  1190. TestPersistence.snapshotEffect = undefined
  1191. if (signal === undefined) throw new Error('expected reconciliation signal')
  1192. started.resolve(signal)
  1193. await new Promise<void>((resolve) => {
  1194. signal.addEventListener('abort', () => { resolve() }, { once: true })
  1195. })
  1196. abortObserved.resolve(undefined)
  1197. await cleanup.promise
  1198. signal.throwIfAborted()
  1199. }
  1200. const controller = new AbortController()
  1201. const pending = ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: controller.signal })
  1202. expect(await started.promise).toBe(controller.signal)
  1203. let settled = false
  1204. void pending.then(
  1205. () => { settled = true },
  1206. () => { settled = true },
  1207. )
  1208. controller.abort(new Error('cooperative list cancellation'))
  1209. await abortObserved.promise
  1210. expect(settled).toBe(false)
  1211. expect(TestPersistence.snapshotSignals).toEqual([controller.signal])
  1212. expect(TestPersistence.inspectSignals).toEqual([])
  1213. cleanup.resolve(undefined)
  1214. await expect(pending).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1215. })
  1216. it('keeps a second search serialized while an abort-ignoring snapshot list finishes', async () => {
  1217. const durable = header('serialized-list-abort')
  1218. TestPersistence.reset([{ meta: durable, events: messageEvents('needle') }])
  1219. const ctx = await liveContext()
  1220. await ctx.plugin(TestPersistence)
  1221. const cleanup = Promise.withResolvers<undefined>()
  1222. const started = Promise.withResolvers<undefined>()
  1223. TestPersistence.listGate = cleanup.promise
  1224. TestPersistence.listStarted = () => {
  1225. TestPersistence.listStarted = undefined
  1226. started.resolve(undefined)
  1227. }
  1228. const controller = new AbortController()
  1229. const first = ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: controller.signal })
  1230. await started.promise
  1231. let firstSettled = false
  1232. let secondSettled = false
  1233. void first.then(
  1234. () => { firstSettled = true },
  1235. () => { firstSettled = true },
  1236. )
  1237. controller.abort(new Error('ignored list cancellation'))
  1238. const second = ctx.sessionQuery.searchEvents({ sessionId: durable.id, query: 'needle' })
  1239. void second.then(
  1240. () => { secondSettled = true },
  1241. () => { secondSettled = true },
  1242. )
  1243. await Promise.resolve()
  1244. expect(firstSettled).toBe(false)
  1245. expect(secondSettled).toBe(false)
  1246. expect(TestPersistence.snapshotSignals).toEqual([controller.signal])
  1247. expect(TestPersistence.inspectSignals).toEqual([])
  1248. cleanup.resolve(undefined)
  1249. await expect(first).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1250. await expect(second).resolves.toMatchObject({ items: [{ sessionId: durable.id }] })
  1251. })
  1252. it('awaits an abort-ignoring inspection and starts neither another inspection nor the after-list', async () => {
  1253. const first = header('ignored-inspect-first')
  1254. const second = header('ignored-inspect-second')
  1255. TestPersistence.reset([
  1256. { meta: first, events: messageEvents('first needle') },
  1257. { meta: second, events: messageEvents('second needle') },
  1258. ])
  1259. const ctx = await liveContext()
  1260. await ctx.plugin(TestPersistence)
  1261. const started = Promise.withResolvers<AbortSignal>()
  1262. const cleanup = Promise.withResolvers<undefined>()
  1263. TestPersistence.inspectEffect = async (_entry, signal) => {
  1264. TestPersistence.inspectEffect = undefined
  1265. if (signal === undefined) throw new Error('expected reconciliation signal')
  1266. started.resolve(signal)
  1267. await cleanup.promise
  1268. }
  1269. const controller = new AbortController()
  1270. const pending = ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: controller.signal })
  1271. expect(await started.promise).toBe(controller.signal)
  1272. let settled = false
  1273. void pending.then(
  1274. () => { settled = true },
  1275. () => { settled = true },
  1276. )
  1277. controller.abort(new Error('ignored inspect cancellation'))
  1278. await Promise.resolve()
  1279. expect(settled).toBe(false)
  1280. expect(TestPersistence.snapshotSignals).toEqual([controller.signal])
  1281. expect(TestPersistence.inspections.get(first.id)).toBe(1)
  1282. expect(TestPersistence.inspections.get(second.id)).toBeUndefined()
  1283. cleanup.resolve(undefined)
  1284. await expect(pending).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1285. expect(TestPersistence.snapshotSignals).toEqual([controller.signal])
  1286. expect(TestPersistence.inspections.get(second.id)).toBeUndefined()
  1287. })
  1288. it('cancels both queued and in-flight source waits without committing them', async () => {
  1289. TestPersistence.reset()
  1290. const ctx = await liveContext()
  1291. await ctx.plugin(TestPersistence)
  1292. const boundaryController = new AbortController()
  1293. const boundary = ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: boundaryController.signal })
  1294. queueMicrotask(() => { boundaryController.abort() })
  1295. await expect(boundary).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1296. const readyController = new AbortController()
  1297. readyController.abort()
  1298. const internals = ctx.sessionQuery as unknown as {
  1299. _ensureReady(signal: AbortSignal): Promise<void>
  1300. }
  1301. await expect(internals._ensureReady(readyController.signal))
  1302. .rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1303. let releaseBlocking!: () => void
  1304. TestPersistence.listGate = new Promise<void>((resolve) => { releaseBlocking = resolve })
  1305. let markBlockingStarted!: () => void
  1306. const blockingStarted = new Promise<void>((resolve) => { markBlockingStarted = resolve })
  1307. TestPersistence.listStarted = () => {
  1308. TestPersistence.listStarted = undefined
  1309. markBlockingStarted()
  1310. }
  1311. const blocking = ctx.sessionQuery.searchSessions({ query: 'needle' })
  1312. await blockingStarted
  1313. const queuedController = new AbortController()
  1314. const queued = ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: queuedController.signal })
  1315. queuedController.abort()
  1316. await expect(queued).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1317. releaseBlocking()
  1318. await expect(blocking).resolves.toEqual({ items: [] })
  1319. TestPersistence.set({
  1320. meta: header('uncommitted'),
  1321. events: messageEvents('durable needle'),
  1322. })
  1323. let releaseActive!: () => void
  1324. TestPersistence.listGate = new Promise<void>((resolve) => { releaseActive = resolve })
  1325. let markActiveStarted!: () => void
  1326. const activeStarted = new Promise<void>((resolve) => { markActiveStarted = resolve })
  1327. TestPersistence.listStarted = () => {
  1328. TestPersistence.listStarted = undefined
  1329. markActiveStarted()
  1330. }
  1331. const activeController = new AbortController()
  1332. const active = ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: activeController.signal })
  1333. await activeStarted
  1334. activeController.abort()
  1335. let activeSettled = false
  1336. void active.then(
  1337. () => { activeSettled = true },
  1338. () => { activeSettled = true },
  1339. )
  1340. await Promise.resolve()
  1341. expect(activeSettled).toBe(false)
  1342. releaseActive()
  1343. await expect(active).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1344. const db = (ctx.sessionQuery as unknown as { _db: DatabaseSync })._db
  1345. expect(db.prepare('SELECT COUNT(*) AS count FROM persisted_sessions').get()).toEqual({ count: 0 })
  1346. await expect(ctx.sessionQuery.searchSessions({ query: 'needle' }))
  1347. .resolves.toMatchObject({ items: [{ header: { id: SessionId('uncommitted') } }] })
  1348. })
  1349. it.each([
  1350. [new Error('ready error'), 'ready error'],
  1351. ['non-error ready failure', 'session-search dependency rejected with a non-Error value'],
  1352. ])('normalizes a rejected readiness wait before mapping it to an index error', async (failure, detail) => {
  1353. TestPersistence.reset()
  1354. const ctx = await liveContext()
  1355. const internals = ctx.sessionQuery as unknown as {
  1356. _ready: Promise<void>
  1357. _ensureReady(signal: AbortSignal): Promise<void>
  1358. }
  1359. internals._ready = Promise.resolve().then(() => {
  1360. throw failure
  1361. })
  1362. await expect(internals._ensureReady(new AbortController().signal))
  1363. .rejects.toThrow(`session-search SQLite index failed to open: ${detail}`)
  1364. })
  1365. it('checks cancellation after readiness before reconciliation accesses SQLite', async () => {
  1366. TestPersistence.reset()
  1367. const ctx = await liveContext()
  1368. const internals = ctx.sessionQuery as unknown as {
  1369. _db: DatabaseSync
  1370. _ready: Promise<void>
  1371. _ensureReady(signal: AbortSignal | undefined): Promise<void>
  1372. }
  1373. const readiness = Promise.withResolvers<undefined>()
  1374. internals._ready = readiness.promise
  1375. const readyWaitStarted = Promise.withResolvers<undefined>()
  1376. const ensureReady = internals._ensureReady.bind(internals)
  1377. vi.spyOn(internals, '_ensureReady').mockImplementation(async (signal) => {
  1378. const pending = ensureReady(signal)
  1379. readyWaitStarted.resolve(undefined)
  1380. return pending
  1381. })
  1382. const prepare = vi.spyOn(internals._db, 'prepare')
  1383. const reason = new Error('cancelled after readiness')
  1384. const controller = new AbortController()
  1385. const pending = ctx.sessionQuery.searchSessions({ query: 'needle' }, { signal: controller.signal })
  1386. await readyWaitStarted.promise
  1387. const queueBoundaryAbort = readiness.promise.then(() => {
  1388. queueMicrotask(() => { controller.abort(reason) })
  1389. })
  1390. readiness.resolve(undefined)
  1391. await queueBoundaryAbort
  1392. await expect(pending).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  1393. expect(prepare).not.toHaveBeenCalled()
  1394. })
  1395. it('rejects queued and future work when close waits for an accepted operation', async () => {
  1396. TestPersistence.reset()
  1397. let release!: () => void
  1398. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  1399. let markStarted!: () => void
  1400. const started = new Promise<void>((resolve) => { markStarted = resolve })
  1401. TestPersistence.listStarted = () => {
  1402. TestPersistence.listStarted = undefined
  1403. markStarted()
  1404. }
  1405. const ctx = await liveContext()
  1406. await ctx.plugin(TestPersistence)
  1407. const search = ctx.sessionQuery as SessionQuerySqlite
  1408. const accepted = search.searchSessions({ query: 'needle' })
  1409. await started
  1410. const queued = search.searchSessions({ query: 'needle' })
  1411. const closing = search.close()
  1412. const repeatedClose = search.close()
  1413. expect(repeatedClose).toBe(closing)
  1414. release()
  1415. await expect(accepted).resolves.toEqual({ items: [] })
  1416. await expect(queued).rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  1417. await Promise.all([closing, repeatedClose])
  1418. await expect(search.searchSessions({ query: 'needle' }))
  1419. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  1420. expect(search.close()).toBe(closing)
  1421. })
  1422. it('awaits optional-persistence child-fiber quiescence on disposal', async () => {
  1423. TestPersistence.reset()
  1424. const ctx = new Context()
  1425. await ctx.plugin(SessionStore)
  1426. const search = await ctx.plugin(SessionQuerySqlite, { path: ':memory:' })
  1427. const persistence = await ctx.plugin(TestPersistence)
  1428. const optional = (ctx.sessionQuery as unknown as {
  1429. _optionalPersistenceFiber: Fiber
  1430. })._optionalPersistenceFiber
  1431. let release!: () => void
  1432. const cleanup = new Promise<void>((resolve) => { release = resolve })
  1433. optional.ctx.effect(() => () => cleanup)
  1434. let settled = false
  1435. const disposing = search.dispose().then(() => { settled = true })
  1436. await Promise.resolve()
  1437. expect(settled).toBe(false)
  1438. release()
  1439. await disposing
  1440. await persistence.dispose()
  1441. })
  1442. it('combines the real SQLite persistence backend with the real search service keylessly', async () => {
  1443. const persistencePath = await temporaryPath('canonical.db')
  1444. const searchPath = await temporaryPath('derived.db')
  1445. const ctx = new Context()
  1446. await ctx.plugin(SessionStore)
  1447. const persistence = await ctx.plugin(SessionPersistenceSqlite, { path: persistencePath })
  1448. const search = await ctx.plugin(SessionQuerySqlite, { path: searchPath })
  1449. const meta = header('real', 10, { cwd: '/work' })
  1450. await ctx.sessionPersistence.create(meta)
  1451. await ctx.sessionPersistence.append(meta.id, messageEvents('real SQLite needle'))
  1452. await expect(ctx.sessionQuery.searchSessions({ query: 'SQLite needle' }))
  1453. .resolves.toMatchObject({ items: [{ header: meta, persisted: true, live: false }] })
  1454. await expect(ctx.sessionQuery.searchEvents({ sessionId: meta.id, query: 'SQLite needle' }))
  1455. .resolves.toMatchObject({ session: meta, items: [{ sessionId: meta.id, seq: 0 }] })
  1456. await expect(ctx.sessionQuery.searchEvents({ sessionId: SessionId('absent'), query: 'needle' }))
  1457. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  1458. await search.dispose()
  1459. await expect(ctx.sessionPersistence.load(meta.id)).resolves.toMatchObject({ meta, events: [{ seq: 0 }] })
  1460. await persistence.dispose()
  1461. })
  1462. it('reconciles colliding local revisions when a derived index reopens against another SQLite store', async () => {
  1463. const persistencePathA = await temporaryPath('canonical-a.db')
  1464. const persistencePathB = await temporaryPath('canonical-b.db')
  1465. const searchPath = await temporaryPath('derived-collision.db')
  1466. const shared = header('same-id', 10)
  1467. const first = new Context()
  1468. await first.plugin(SessionStore)
  1469. const persistenceA = await first.plugin(SessionPersistenceSqlite, { path: persistencePathA })
  1470. await first.sessionPersistence.create(shared)
  1471. await first.sessionPersistence.append(shared.id, messageEvents('alpha source'))
  1472. const inspectA = vi.spyOn(first.sessionPersistence, 'inspect')
  1473. const searchA = await first.plugin(SessionQuerySqlite, { path: searchPath })
  1474. await expect(first.sessionQuery.searchSessions({ query: 'alpha' }))
  1475. .resolves.toMatchObject({ items: [{ header: shared }] })
  1476. expect(inspectA).toHaveBeenCalledTimes(1)
  1477. await searchA.dispose()
  1478. await persistenceA.dispose()
  1479. const reopened = new Context()
  1480. await reopened.plugin(SessionStore)
  1481. const persistenceAAgain = await reopened.plugin(SessionPersistenceSqlite, { path: persistencePathA })
  1482. const reopenedInspect = vi.spyOn(reopened.sessionPersistence, 'inspect')
  1483. const searchAAgain = await reopened.plugin(SessionQuerySqlite, { path: searchPath })
  1484. await expect(reopened.sessionQuery.searchSessions({ query: 'alpha' }))
  1485. .resolves.toMatchObject({ items: [{ header: shared }] })
  1486. expect(reopenedInspect).not.toHaveBeenCalled()
  1487. await searchAAgain.dispose()
  1488. await persistenceAAgain.dispose()
  1489. const second = new Context()
  1490. await second.plugin(SessionStore)
  1491. const persistenceB = await second.plugin(SessionPersistenceSqlite, { path: persistencePathB })
  1492. await second.sessionPersistence.create(shared)
  1493. await second.sessionPersistence.append(shared.id, messageEvents('bravo source'))
  1494. const inspectB = vi.spyOn(second.sessionPersistence, 'inspect')
  1495. const searchB = await second.plugin(SessionQuerySqlite, { path: searchPath })
  1496. await expect(second.sessionQuery.searchSessions({ query: 'bravo' }))
  1497. .resolves.toMatchObject({ items: [{ header: shared }] })
  1498. await expect(second.sessionQuery.searchSessions({ query: 'alpha' })).resolves.toEqual({ items: [] })
  1499. expect(inspectB).toHaveBeenCalledTimes(1)
  1500. await searchB.dispose()
  1501. await persistenceB.dispose()
  1502. })
  1503. })