sqlite.spec.ts 51 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102
  1. import { afterEach, describe, expect, it, vi } from 'vitest'
  2. import { Context, type Fiber } from 'cordis'
  3. import { DatabaseSync } from 'node:sqlite'
  4. import { mkdtemp, rm } from 'node:fs/promises'
  5. import { tmpdir } from 'node:os'
  6. import { join } from 'node:path'
  7. import SessionStore, { SESSION_FORMAT_VERSION, SessionId } from '@deepseek-ai/dsh-session'
  8. import type { SessionEvent, SessionHeader, SessionId as SessionIdType } from '@deepseek-ai/dsh-session'
  9. import SessionPersistence, { SessionPersistenceRevision } from '@deepseek-ai/dsh-session-persistence'
  10. import type { SessionPersistenceSnapshot } from '@deepseek-ai/dsh-session-persistence'
  11. import SessionPersistenceSqlite from '@deepseek-ai/dsh-session-persistence-sqlite'
  12. import SessionSearchSqlite, {
  13. SESSION_QUERY_SQLITE_APPLICATION_ID,
  14. SESSION_QUERY_SQLITE_SCHEMA_VERSION,
  15. } from '@deepseek-ai/dsh-session-query-sqlite'
  16. import {
  17. SessionQueryError,
  18. SessionSearchCursor,
  19. type SessionAvailability,
  20. type SessionQueryErrorCode,
  21. type SessionSearchRequest,
  22. } from '@deepseek-ai/dsh-session-query'
  23. const temporaryDirectories: string[] = []
  24. afterEach(async () => {
  25. for (const directory of temporaryDirectories.splice(0)) {
  26. await rm(directory, { recursive: true, force: true })
  27. }
  28. })
  29. async function temporaryPath(name = 'search.db'): Promise<string> {
  30. const directory = await mkdtemp(join(tmpdir(), 'dsh-session-search-'))
  31. temporaryDirectories.push(directory)
  32. return join(directory, name)
  33. }
  34. function header(id: string, createdAt = 1, extra: Partial<SessionHeader> = {}): SessionHeader {
  35. return { version: SESSION_FORMAT_VERSION, id: SessionId(id), createdAt, ...extra }
  36. }
  37. function messageEvents(text: string, time = 1): SessionEvent[] {
  38. return [{
  39. type: 'user/message',
  40. seq: 0,
  41. time,
  42. data: { content: [{ type: 'text', text }], source: { kind: 'user' } },
  43. surfaceOp: 'append',
  44. }]
  45. }
  46. function expectCode(code: SessionQueryErrorCode): Error {
  47. return expect.objectContaining({ code }) as Error
  48. }
  49. function replaceCursorOffset(
  50. cursor: ReturnType<typeof SessionSearchCursor>,
  51. offset: number,
  52. ): ReturnType<typeof SessionSearchCursor> {
  53. const payload = JSON.parse(
  54. Buffer.from(cursor, 'base64url').toString('utf8'),
  55. ) as Record<string, unknown>
  56. return SessionSearchCursor(Buffer.from(JSON.stringify({ ...payload, offset }), 'utf8').toString('base64url'))
  57. }
  58. class TestPersistence extends SessionPersistence {
  59. static entries = new Map<SessionIdType, { meta: SessionHeader; events: SessionEvent[] }>()
  60. static revisions = new Map<SessionIdType, number>()
  61. static nextRevision = 0
  62. static loads = new Map<SessionIdType, number>()
  63. static loadEffect: ((entry: { meta: SessionHeader; events: SessionEvent[] }) => void) | undefined
  64. static listGate: Promise<void> | undefined
  65. static listStarted: (() => void) | undefined
  66. static snapshotEffect: (() => void | Promise<void>) | undefined
  67. static snapshotOverride: (() => SessionPersistenceSnapshot[]) | undefined
  68. static failure: unknown
  69. static reset(entries: readonly { meta: SessionHeader; events: SessionEvent[] }[] = []): void {
  70. this.entries = new Map()
  71. this.revisions = new Map()
  72. this.loads = new Map()
  73. this.loadEffect = undefined
  74. for (const entry of entries) this.set(entry)
  75. this.listGate = undefined
  76. this.listStarted = undefined
  77. this.snapshotEffect = undefined
  78. this.snapshotOverride = undefined
  79. this.failure = undefined
  80. }
  81. static set(entry: { meta: SessionHeader; events: SessionEvent[] }): void {
  82. this.entries.set(entry.meta.id, structuredClone(entry))
  83. this.revisions.set(entry.meta.id, ++this.nextRevision)
  84. }
  85. create(meta: SessionHeader): Promise<void> {
  86. TestPersistence.set({ meta, events: [] })
  87. return Promise.resolve()
  88. }
  89. append(id: SessionIdType, events: readonly SessionEvent[]): Promise<void> {
  90. const entry = TestPersistence.entries.get(id)
  91. if (entry === undefined) return Promise.reject(new Error('missing test session'))
  92. entry.events.push(...structuredClone(events))
  93. TestPersistence.revisions.set(id, ++TestPersistence.nextRevision)
  94. return Promise.resolve()
  95. }
  96. async load(id: SessionIdType): Promise<{ meta: SessionHeader; events: SessionEvent[] }> {
  97. TestPersistence.loads.set(id, (TestPersistence.loads.get(id) ?? 0) + 1)
  98. if (TestPersistence.failure !== undefined) throw TestPersistence.failure
  99. const entry = TestPersistence.entries.get(id)
  100. if (entry === undefined) throw new Error('missing test session')
  101. if (TestPersistence.loadEffect !== undefined) {
  102. const effect = TestPersistence.loadEffect
  103. TestPersistence.loadEffect = undefined
  104. effect(entry)
  105. TestPersistence.revisions.set(id, ++TestPersistence.nextRevision)
  106. }
  107. return structuredClone(entry)
  108. }
  109. async list(): Promise<SessionHeader[]> {
  110. TestPersistence.listStarted?.()
  111. await TestPersistence.listGate
  112. if (TestPersistence.failure !== undefined) throw TestPersistence.failure
  113. return [...TestPersistence.entries.values()].map(entry => structuredClone(entry.meta))
  114. }
  115. async listSnapshots(): Promise<SessionPersistenceSnapshot[]> {
  116. TestPersistence.listStarted?.()
  117. await TestPersistence.listGate
  118. if (TestPersistence.failure !== undefined) throw TestPersistence.failure
  119. const snapshots = TestPersistence.snapshotOverride?.()
  120. ?? [...TestPersistence.entries.values()].map(entry => ({
  121. header: structuredClone(entry.meta),
  122. revision: SessionPersistenceRevision(`test:${TestPersistence.revisions.get(entry.meta.id)}`),
  123. }))
  124. await TestPersistence.snapshotEffect?.()
  125. return snapshots
  126. }
  127. }
  128. async function liveContext(config: ConstructorParameters<typeof SessionSearchSqlite>[1] = { path: ':memory:' }): Promise<Context> {
  129. const ctx = new Context()
  130. await ctx.plugin(SessionStore)
  131. await ctx.plugin(SessionSearchSqlite, config)
  132. return ctx
  133. }
  134. describe('SQLite session search', () => {
  135. it('searches two-character Unicode61 tokens in live-only sessions', async () => {
  136. const ctx = await liveContext({ path: ':memory:', snippetChars: 20 })
  137. const session = ctx.sessions.create(SessionId('live'), { meta: { cwd: '/work', createdAt: 10, seedLength: 1 } })
  138. session.append(
  139. 'user/message',
  140. { content: [{ type: 'text', text: 'An AI helper' }], source: { kind: 'user' } },
  141. { surfaceOp: 'append' },
  142. )
  143. await expect(ctx.sessionSearch.searchEvents({ sessionId: session.id, query: 'AI' }))
  144. .resolves.toMatchObject({ items: [{ sessionId: session.id, seq: 0, snippet: 'An AI helper' }] })
  145. await expect(ctx.sessionSearch.searchSessions({ query: 'AI' }))
  146. .resolves.toMatchObject({ items: [{ header: { ...session.header, seedLength: 1 }, live: true, persisted: false }] })
  147. })
  148. it('searches all surfaces by default and applies metadata before ranking', async () => {
  149. const ctx = await liveContext({ path: ':memory:', defaultLimit: 10, maxLimit: 20 })
  150. const parent = SessionId('parent')
  151. const events: SessionEvent[] = [
  152. { type: 'user/message', seq: 0, time: 10, data: { content: [{ type: 'text', text: 'needle original' }], source: { kind: 'user' } }, surfaceOp: 'append' },
  153. { type: 'assistant/chunk', seq: 1, time: 11, data: { turn: 1, step: 1, chunk: { type: 'text-delta', index: 0, text: 'needle raw' } } },
  154. { 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 } },
  155. { type: 'turn/end', seq: 3, time: 13, data: { turn: 1, reason: { kind: 'error', step: 1, message: 'needle failure' } } },
  156. ]
  157. ctx.sessions.create(SessionId('a'), { seed: events, meta: { cwd: '/a', parentSession: parent, createdAt: 20 } })
  158. ctx.sessions.create(SessionId('b'), { seed: messageEvents('needle peer', 12), meta: { createdAt: 20 } })
  159. const all = await ctx.sessionSearch.searchEvents({ sessionId: SessionId('a'), query: 'needle' })
  160. expect(new Set(all.items.map(item => item.surface))).toEqual(new Set(['current', 'shadowed', 'log-only']))
  161. await expect(ctx.sessionSearch.searchEvents({
  162. sessionId: SessionId('a'),
  163. query: 'needle',
  164. filters: [
  165. { kind: 'seq', from: 2, to: 2 },
  166. { kind: 'time', from: 12, to: 12 },
  167. { kind: 'type', values: ['user/message'] },
  168. { kind: 'surface', values: ['current'] },
  169. ],
  170. })).resolves.toMatchObject({ items: [{ seq: 2, surface: 'current' }] })
  171. const grouped = await ctx.sessionSearch.searchSessions({
  172. query: 'needle',
  173. sessionFilters: [
  174. { kind: 'id', values: [SessionId('a')] },
  175. { kind: 'cwd', values: ['/a'] },
  176. { kind: 'created-at', from: 20, to: 20 },
  177. { kind: 'parent', values: [parent] },
  178. { kind: 'availability', values: ['live'] },
  179. ],
  180. eventFilters: [{ kind: 'surface', values: ['shadowed'] }],
  181. })
  182. expect(grouped.items).toHaveLength(1)
  183. expect(grouped.items[0]).toMatchObject({
  184. header: { id: SessionId('a'), cwd: '/a', parentSession: parent },
  185. live: true,
  186. persisted: false,
  187. bestMatch: { seq: 0, surface: 'shadowed' },
  188. })
  189. })
  190. it('uses literal phrase tokens, stable ties, and bounded Unicode snippets', async () => {
  191. const ctx = await liveContext({ path: ':memory:', defaultLimit: 10, maxLimit: 10, snippetChars: 5 })
  192. ctx.sessions.create(SessionId('a'), { seed: messageEvents('😀😀 alpha beta BRAID 😀😀', 10), meta: { createdAt: 1 } })
  193. ctx.sessions.create(SessionId('b'), { seed: messageEvents('alpha beta', 10), meta: { createdAt: 1 } })
  194. ctx.sessions.create(SessionId('c'), { seed: messageEvents('alpha middle beta', 10), meta: { createdAt: 1 } })
  195. ctx.sessions.create(SessionId('d'), { seed: messageEvents('alpha beta', 10), meta: { createdAt: 1 } })
  196. ctx.sessions.create(SessionId('operator'), { seed: messageEvents('needle OR absent', 10), meta: { createdAt: 1 } })
  197. ctx.sessions.create(SessionId('only'), { seed: messageEvents('needle only', 10), meta: { createdAt: 1 } })
  198. ctx.sessions.create(SessionId('quote'), { seed: messageEvents('say "needle" exactly', 10), meta: { createdAt: 1 } })
  199. const phrase = await ctx.sessionSearch.searchSessions({ query: 'alpha beta' })
  200. expect(phrase.items.map(item => item.header.id)).toEqual([SessionId('b'), SessionId('d'), SessionId('a')])
  201. expect(phrase.items.every(item => Array.from(item.bestMatch.snippet).length <= 5)).toBe(true)
  202. await expect(ctx.sessionSearch.searchSessions({ query: 'AI' })).resolves.toEqual({ items: [] })
  203. await expect(ctx.sessionSearch.searchSessions({ query: 'needle OR absent' }))
  204. .resolves.toMatchObject({ items: [{ header: { id: SessionId('operator') } }] })
  205. await expect(ctx.sessionSearch.searchSessions({ query: 'say "needle"' }))
  206. .resolves.toMatchObject({ items: [{ header: { id: SessionId('quote') } }] })
  207. await expect(ctx.sessionSearch.searchSessions({ query: '*' })).resolves.toEqual({ items: [] })
  208. })
  209. it('ranks live and persisted matches on one source-comparable contract', async () => {
  210. const persisted = header('z-persisted')
  211. TestPersistence.reset([
  212. { meta: persisted, events: messageEvents('needle needle', 10) },
  213. ...Array.from({ length: 12 }, (_, index) => ({
  214. meta: header(`filler-${index}`),
  215. events: messageEvents('needle', 10),
  216. })),
  217. ])
  218. const ctx = await liveContext()
  219. const persistence = await ctx.plugin(TestPersistence)
  220. ctx.sessions.create(SessionId('a-live'), {
  221. seed: messageEvents('needle needle', 10),
  222. meta: { createdAt: persisted.createdAt },
  223. })
  224. const result = await ctx.sessionSearch.searchSessions({
  225. query: 'needle',
  226. sessionFilters: [{ kind: 'id', values: [SessionId('a-live'), persisted.id] }],
  227. })
  228. expect(result.items.map(item => item.header.id)).toEqual([SessionId('a-live'), persisted.id])
  229. await persistence.dispose()
  230. })
  231. it('positions snippets from FTS5 matches across diacritics and punctuation', async () => {
  232. const ctx = await liveContext({ path: ':memory:', snippetChars: 14 })
  233. const session = ctx.sessions.create(SessionId('snippet'), {
  234. seed: messageEvents('long long long—café,\nnext value', 10),
  235. })
  236. const page = await ctx.sessionSearch.searchEvents({ sessionId: session.id, query: 'CAFE' })
  237. expect(page.items).toHaveLength(1)
  238. expect(page.items[0]!.snippet).toContain('café')
  239. expect(page.items[0]!.snippet).toContain('—')
  240. expect(page.items[0]!.snippet).not.toContain('\n')
  241. expect(Array.from(page.items[0]!.snippet).length).toBeLessThanOrEqual(14)
  242. })
  243. it('binds cursors to requests and only invalidates within-session pages for target changes', async () => {
  244. const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 5 })
  245. const target = ctx.sessions.create(SessionId('target'), {
  246. seed: [
  247. ...messageEvents('needle one', 10),
  248. { ...messageEvents('needle two', 11)[0]!, seq: 1 },
  249. { ...messageEvents('needle three', 12)[0]!, seq: 2 },
  250. ],
  251. })
  252. ctx.sessions.create(SessionId('other'), { seed: messageEvents('needle other', 10) })
  253. const eventPage = await ctx.sessionSearch.searchEvents({ sessionId: target.id, query: 'needle', limit: 1 })
  254. const sessionPage = await ctx.sessionSearch.searchSessions({ query: 'needle', limit: 1 })
  255. expect(eventPage.nextCursor).toEqual(expect.any(String))
  256. expect(sessionPage.nextCursor).toEqual(expect.any(String))
  257. if (eventPage.nextCursor === undefined || sessionPage.nextCursor === undefined) throw new Error('expected cursors')
  258. const unsafeOffsetCursor = replaceCursorOffset(eventPage.nextCursor, 1e100)
  259. await expect(ctx.sessionSearch.searchEvents({
  260. sessionId: target.id,
  261. query: 'needle',
  262. limit: 1,
  263. cursor: unsafeOffsetCursor,
  264. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CURSOR'))
  265. const eventKeys = eventPage.items.map(item => `${item.sessionId}:${item.seq}`)
  266. let eventCursor: ReturnType<typeof SessionSearchCursor> | undefined = eventPage.nextCursor
  267. while (eventCursor !== undefined) {
  268. const next = await ctx.sessionSearch.searchEvents({
  269. sessionId: target.id,
  270. query: 'needle',
  271. limit: 1,
  272. cursor: eventCursor,
  273. })
  274. eventKeys.push(...next.items.map(item => `${item.sessionId}:${item.seq}`))
  275. eventCursor = next.nextCursor
  276. }
  277. expect(eventKeys).toHaveLength(3)
  278. expect(new Set(eventKeys).size).toBe(eventKeys.length)
  279. const sessionIds = sessionPage.items.map(item => item.header.id)
  280. let sessionCursor: ReturnType<typeof SessionSearchCursor> | undefined = sessionPage.nextCursor
  281. while (sessionCursor !== undefined) {
  282. const next = await ctx.sessionSearch.searchSessions({ query: 'needle', limit: 1, cursor: sessionCursor })
  283. sessionIds.push(...next.items.map(item => item.header.id))
  284. sessionCursor = next.nextCursor
  285. }
  286. expect(sessionIds).toHaveLength(2)
  287. expect(new Set(sessionIds).size).toBe(sessionIds.length)
  288. ctx.sessions.create(SessionId('unrelated'), { seed: messageEvents('needle unrelated', 20) })
  289. await expect(ctx.sessionSearch.searchEvents({
  290. sessionId: target.id,
  291. query: 'needle',
  292. limit: 1,
  293. cursor: eventPage.nextCursor,
  294. })).resolves.toMatchObject({ items: [{ sessionId: target.id }] })
  295. await expect(ctx.sessionSearch.searchSessions({ query: 'needle', limit: 1, cursor: sessionPage.nextCursor }))
  296. .rejects.toThrow(expectCode('SESSION_QUERY_STALE_CURSOR'))
  297. await expect(ctx.sessionSearch.searchEvents({
  298. sessionId: target.id,
  299. query: 'different',
  300. limit: 1,
  301. cursor: eventPage.nextCursor,
  302. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CURSOR'))
  303. target.append('user/message', { content: [{ type: 'text', text: 'needle four' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
  304. await expect(ctx.sessionSearch.searchEvents({
  305. sessionId: target.id,
  306. query: 'needle',
  307. limit: 1,
  308. cursor: eventPage.nextCursor,
  309. })).rejects.toThrow(expectCode('SESSION_QUERY_STALE_CURSOR'))
  310. })
  311. it('invalidates session cursors after transient persistence topology changes', async () => {
  312. TestPersistence.reset()
  313. const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 5 })
  314. ctx.sessions.create(SessionId('first'), { seed: messageEvents('needle first') })
  315. ctx.sessions.create(SessionId('second'), { seed: messageEvents('needle second') })
  316. const page = await ctx.sessionSearch.searchSessions({ query: 'needle', limit: 1 })
  317. if (page.nextCursor === undefined) throw new Error('expected cursor')
  318. const persistence = await ctx.plugin(TestPersistence)
  319. await persistence.dispose()
  320. await expect(ctx.sessionSearch.searchSessions({
  321. query: 'needle',
  322. limit: 1,
  323. cursor: page.nextCursor,
  324. })).rejects.toThrow(expectCode('SESSION_QUERY_STALE_CURSOR'))
  325. })
  326. it('rejects invalid requests, filters, cursors, and direct config', async () => {
  327. const ctx = await liveContext({ path: ':memory:', defaultLimit: 2, maxLimit: 3 })
  328. const session = ctx.sessions.create(SessionId('valid'), { seed: messageEvents('needle') })
  329. for (const request of [
  330. { sessionId: session.id, query: '' },
  331. { sessionId: session.id, query: 'needle', limit: 0 },
  332. { sessionId: session.id, query: 'needle', limit: 4 },
  333. { sessionId: session.id, query: 'needle', filters: [{ kind: 'seq', from: 2, to: 1 }] },
  334. { sessionId: session.id, query: 'needle', filters: [{ kind: 'surface', values: ['future'] }] },
  335. { sessionId: session.id, query: 'bad\0query' },
  336. ] as const) {
  337. await expect(ctx.sessionSearch.searchEvents(request as never)).rejects.toBeInstanceOf(Error)
  338. }
  339. await expect(ctx.sessionSearch.searchSessions({
  340. query: 'needle',
  341. sessionFilters: [{ kind: 'availability', values: ['remote' as never] }],
  342. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  343. await expect(ctx.sessionSearch.searchSessions({
  344. query: 'needle',
  345. sessionFilters: [{ kind: 'future' } as never],
  346. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  347. await expect(ctx.sessionSearch.searchSessions({
  348. query: 'needle',
  349. eventFilters: [{ kind: 'future' } as never],
  350. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  351. await expect(ctx.sessionSearch.searchEvents({
  352. sessionId: session.id,
  353. query: 'needle',
  354. filters: [{ kind: 'future' } as never],
  355. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  356. await expect(ctx.sessionSearch.searchEvents({
  357. sessionId: session.id,
  358. query: 'needle',
  359. cursor: SessionSearchCursor('not-json'),
  360. }))
  361. .rejects.toThrow(expectCode('SESSION_QUERY_INVALID_CURSOR'))
  362. await expect(ctx.sessionSearch.searchEvents({ sessionId: SessionId('absent'), query: 'needle' }))
  363. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  364. for (const config of [
  365. { path: '' },
  366. { path: ':memory:', defaultLimit: 0 },
  367. { path: ':memory:', maxLimit: 0 },
  368. { path: ':memory:', defaultLimit: 1e100 },
  369. { path: ':memory:', maxLimit: 1e100 },
  370. { path: ':memory:', snippetChars: 0 },
  371. { path: ':memory:', defaultLimit: 3, maxLimit: 2 },
  372. { path: ':memory:', journalMode: 'memory' },
  373. ]) {
  374. const direct = new Context()
  375. await direct.plugin(SessionStore)
  376. expect(() => new SessionSearchSqlite(direct, config as never))
  377. .toThrow(expectCode('SESSION_QUERY_INVALID_CONFIG'))
  378. }
  379. })
  380. it('rejects aggregate filter bindings above SQLite\'s portable variable limit', async () => {
  381. const ctx = await liveContext()
  382. const session = ctx.sessions.create(SessionId('binding-limit'), { seed: messageEvents('needle') })
  383. // Each clause is below the ceiling; combined with its sibling and fixed
  384. // query bindings, the complete statement is not portable.
  385. const halfPortableLimit = 16_383
  386. const ids = Array.from(
  387. { length: halfPortableLimit },
  388. (_, index) => SessionId(`binding-${index}`),
  389. )
  390. const types = Array.from({ length: halfPortableLimit }, () => 'user/message' as const)
  391. const surfaces = Array.from({ length: halfPortableLimit }, () => 'current' as const)
  392. await expect(ctx.sessionSearch.searchSessions({
  393. query: 'needle',
  394. sessionFilters: [{ kind: 'id', values: ids }],
  395. eventFilters: [{ kind: 'type', values: types }],
  396. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  397. await expect(ctx.sessionSearch.searchEvents({
  398. sessionId: session.id,
  399. query: 'needle',
  400. filters: [
  401. { kind: 'type', values: types },
  402. { kind: 'surface', values: surfaces },
  403. ],
  404. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  405. })
  406. it('rejects one 125,000-value filter list with a typed error', async () => {
  407. const ctx = await liveContext()
  408. const ids = Array.from(
  409. { length: 125_000 },
  410. (_, index) => SessionId(`oversized-binding-${index}`),
  411. )
  412. await expect(ctx.sessionSearch.searchSessions({
  413. query: 'needle',
  414. sessionFilters: [{ kind: 'id', values: ids }],
  415. })).rejects.toThrow(expectCode('SESSION_QUERY_INVALID_FILTER'))
  416. })
  417. })
  418. describe('SQLite reconciliation and source lifecycle', () => {
  419. it('owns queued request and filter values before waiting for the serializer', async () => {
  420. const durable = header('owned')
  421. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  422. const ctx = await liveContext()
  423. const persistence = await ctx.plugin(TestPersistence)
  424. let release!: () => void
  425. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  426. let markStarted!: () => void
  427. const started = new Promise<void>((resolve) => { markStarted = resolve })
  428. TestPersistence.listStarted = () => {
  429. TestPersistence.listStarted = undefined
  430. markStarted()
  431. }
  432. const blocking = ctx.sessionSearch.searchSessions({ query: 'needle' })
  433. await started
  434. const availability: SessionAvailability[] = ['persisted']
  435. const request: SessionSearchRequest = {
  436. query: 'needle',
  437. sessionFilters: [{ kind: 'availability', values: availability }],
  438. }
  439. const queued = ctx.sessionSearch.searchSessions(request)
  440. request.query = 'absent'
  441. availability[0] = 'live'
  442. release()
  443. await expect(blocking).resolves.toMatchObject({ items: [{ header: durable }] })
  444. await expect(queued).resolves.toMatchObject({ items: [{ header: durable }] })
  445. await persistence.dispose()
  446. })
  447. it('mounts persistence dynamically, shadows with TEMP live rows, reveals, and hides on unmount', async () => {
  448. const shared = header('shared', 10, { cwd: '/work' })
  449. const durable = header('durable', 5)
  450. TestPersistence.reset([
  451. { meta: shared, events: messageEvents('persisted needle') },
  452. { meta: durable, events: messageEvents('durable needle') },
  453. ])
  454. const ctx = await liveContext()
  455. await expect(ctx.sessionSearch.searchSessions({ query: 'durable' })).resolves.toEqual({ items: [] })
  456. const persistenceFiber = await ctx.plugin(TestPersistence)
  457. await expect(ctx.sessionSearch.searchSessions({ query: 'durable' }))
  458. .resolves.toMatchObject({ items: [{ header: durable, live: false, persisted: true }] })
  459. const live = ctx.sessions.prepare(shared.id, { meta: { createdAt: 10, cwd: '/work' } })
  460. live.append('user/message', { content: [{ type: 'text', text: 'live needle' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
  461. const detach = ctx.sessions.enter(live)
  462. ctx.sessions.announce(live)
  463. await expect(ctx.sessionSearch.searchSessions({ query: 'persisted' })).resolves.toEqual({ items: [] })
  464. await expect(ctx.sessionSearch.searchSessions({ query: 'live' }))
  465. .resolves.toMatchObject({ items: [{ header: shared, live: true, persisted: true }] })
  466. detach()
  467. await expect(ctx.sessionSearch.searchSessions({ query: 'persisted' }))
  468. .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
  469. await persistenceFiber.dispose()
  470. await expect(ctx.sessionSearch.searchSessions({ query: 'durable' })).resolves.toEqual({ items: [] })
  471. await expect(ctx.sessionSearch.searchEvents({ sessionId: durable.id, query: 'needle' }))
  472. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  473. })
  474. it('uses the reconciled persistence binding through the query boundary', async () => {
  475. const durable = header('post-reconcile-unmount')
  476. TestPersistence.reset([{ meta: durable, events: [
  477. ...messageEvents('durable needle', 1),
  478. { ...messageEvents('durable needle again', 2)[0]!, seq: 1 },
  479. ] }])
  480. const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 2 })
  481. const persistence = await ctx.plugin(TestPersistence)
  482. const internals = ctx.sessionSearch as unknown as {
  483. _reconcile(signal: AbortSignal | undefined): Promise<{
  484. identity: symbol
  485. service?: SessionPersistence
  486. }>
  487. }
  488. const reconcile = internals._reconcile.bind(internals)
  489. const boundary = vi.spyOn(internals, '_reconcile').mockImplementation(async (signal) => {
  490. const binding = await reconcile(signal)
  491. await persistence.dispose()
  492. return binding
  493. })
  494. const page = await ctx.sessionSearch.searchEvents({
  495. sessionId: durable.id,
  496. query: 'needle',
  497. limit: 1,
  498. })
  499. expect(page.items).toMatchObject([{ sessionId: durable.id }])
  500. expect(page.nextCursor).toEqual(expect.any(String))
  501. boundary.mockRestore()
  502. await expect(ctx.sessionSearch.searchEvents({ sessionId: durable.id, query: 'needle' }))
  503. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  504. })
  505. it('discards a stale list rejection when persistence unmounts during observation', async () => {
  506. const durable = header('racing')
  507. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  508. const ctx = await liveContext()
  509. const persistenceFiber = await ctx.plugin(TestPersistence)
  510. let release!: () => void
  511. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  512. let markStarted!: () => void
  513. const started = new Promise<void>((resolve) => { markStarted = resolve })
  514. TestPersistence.listStarted = () => {
  515. TestPersistence.listStarted = undefined
  516. markStarted()
  517. }
  518. const search = ctx.sessionSearch.searchSessions({ query: 'needle' })
  519. await started
  520. await persistenceFiber.dispose()
  521. TestPersistence.failure = new Error('stale backend rejection')
  522. release()
  523. await expect(search).resolves.toEqual({ items: [] })
  524. })
  525. it('retries against a replacement after the prior binding rejects', async () => {
  526. const durable = header('replacement')
  527. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  528. const ctx = await liveContext()
  529. const prior = await ctx.plugin(TestPersistence)
  530. let rejectPrior!: (reason: unknown) => void
  531. TestPersistence.listGate = new Promise<void>((_resolve, reject) => { rejectPrior = reject })
  532. let markStarted!: () => void
  533. const started = new Promise<void>((resolve) => { markStarted = resolve })
  534. TestPersistence.listStarted = () => {
  535. TestPersistence.listStarted = undefined
  536. markStarted()
  537. }
  538. const search = ctx.sessionSearch.searchSessions({ query: 'needle' })
  539. await started
  540. await prior.dispose()
  541. TestPersistence.listGate = undefined
  542. const replacement = await ctx.plugin(TestPersistence)
  543. rejectPrior(new Error('stale prior binding'))
  544. await expect(search).resolves.toMatchObject({ items: [{ header: durable }] })
  545. await replacement.dispose()
  546. })
  547. it('reloads a replacement source even when its opaque revisions collide', async () => {
  548. const durable = header('colliding-replacement')
  549. TestPersistence.reset([{ meta: durable, events: messageEvents('old content') }])
  550. const revision = TestPersistence.revisions.get(durable.id)!
  551. const ctx = await liveContext()
  552. const prior = await ctx.plugin(TestPersistence)
  553. await expect(ctx.sessionSearch.searchSessions({ query: 'old' }))
  554. .resolves.toMatchObject({ items: [{ header: durable }] })
  555. await prior.dispose()
  556. TestPersistence.set({ meta: durable, events: messageEvents('new needle') })
  557. TestPersistence.revisions.set(durable.id, revision)
  558. const replacement = await ctx.plugin(TestPersistence)
  559. const page = await ctx.sessionSearch.searchSessions({ query: 'new needle' })
  560. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  561. expect(page).toMatchObject({ items: [{ header: durable }] })
  562. await expect(ctx.sessionSearch.searchSessions({ query: 'old' })).resolves.toEqual({ items: [] })
  563. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  564. await replacement.dispose()
  565. })
  566. it('retries when a successful observation belongs to a source unmounted during listing', async () => {
  567. const durable = header('successful-unmount')
  568. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  569. const ctx = await liveContext()
  570. const persistence = await ctx.plugin(TestPersistence)
  571. let lists = 0
  572. TestPersistence.snapshotEffect = async () => {
  573. lists += 1
  574. if (lists === 2) await persistence.dispose()
  575. }
  576. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' })).resolves.toEqual({ items: [] })
  577. expect(lists).toBe(2)
  578. })
  579. it('retries when the snapshot population changes during observation', async () => {
  580. const first = header('first')
  581. const added = header('added-during-list')
  582. TestPersistence.reset([{ meta: first, events: messageEvents('first needle') }])
  583. const ctx = await liveContext()
  584. await ctx.plugin(TestPersistence)
  585. TestPersistence.snapshotEffect = () => {
  586. TestPersistence.snapshotEffect = undefined
  587. TestPersistence.set({ meta: added, events: messageEvents('added needle') })
  588. }
  589. const page = await ctx.sessionSearch.searchSessions({ query: 'needle' })
  590. expect(page.items.map(item => item.header.id).sort()).toEqual([added.id, first.id].sort())
  591. expect(TestPersistence.loads.get(first.id)).toBe(2)
  592. expect(TestPersistence.loads.get(added.id)).toBe(1)
  593. })
  594. it('fails after one retry when persistence snapshots keep changing', async () => {
  595. const durable = header('continuous-mutation')
  596. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  597. const ctx = await liveContext()
  598. await ctx.plugin(TestPersistence)
  599. let lists = 0
  600. TestPersistence.snapshotEffect = () => {
  601. lists += 1
  602. TestPersistence.set({ meta: durable, events: messageEvents(`durable needle ${lists}`) })
  603. }
  604. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  605. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  606. expect(lists).toBe(4)
  607. })
  608. it('retries if the persistence binding changes while live sessions are observed', async () => {
  609. const durable = header('live-boundary-retry')
  610. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  611. const ctx = await liveContext()
  612. await ctx.plugin(TestPersistence)
  613. const internals = ctx.sessionSearch as unknown as {
  614. _persistenceBinding: { identity: symbol; service?: SessionPersistence }
  615. }
  616. const originalList = ctx.sessions.list.bind(ctx.sessions)
  617. let bumped = false
  618. const list = vi.spyOn(ctx.sessions, 'list').mockImplementation(() => {
  619. if (!bumped) {
  620. bumped = true
  621. internals._persistenceBinding = {
  622. ...internals._persistenceBinding,
  623. identity: Symbol(),
  624. }
  625. }
  626. return originalList()
  627. })
  628. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  629. .resolves.toMatchObject({ items: [{ header: durable }] })
  630. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  631. list.mockRestore()
  632. })
  633. it('rejects malformed snapshots and preserves typed persistence failures', async () => {
  634. const durable = header('invalid-snapshot')
  635. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  636. const ctx = await liveContext()
  637. await ctx.plugin(TestPersistence)
  638. TestPersistence.snapshotOverride = () => 'not-an-array' as never
  639. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  640. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  641. TestPersistence.snapshotOverride = () => [{ header: durable, revision: 1 as never }]
  642. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  643. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  644. TestPersistence.snapshotOverride = () => [
  645. { header: durable, revision: SessionPersistenceRevision('duplicate:1') },
  646. { header: durable, revision: SessionPersistenceRevision('duplicate:2') },
  647. ]
  648. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  649. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  650. TestPersistence.snapshotOverride = undefined
  651. const typed = new SessionQueryError('typed persistence failure', 'SESSION_QUERY_PERSISTENCE_FAILED')
  652. TestPersistence.failure = typed
  653. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' })).rejects.toBe(typed)
  654. })
  655. it('rejects immutable header conflicts between live and persisted sources', async () => {
  656. const shared = header('conflict', 10)
  657. TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
  658. const ctx = await liveContext()
  659. await ctx.plugin(TestPersistence)
  660. ctx.sessions.create(shared.id, { seed: messageEvents('live needle'), meta: { createdAt: 11 } })
  661. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  662. .rejects.toThrow(expectCode('SESSION_QUERY_SOURCE_CONFLICT'))
  663. })
  664. it('preserves unchanged persisted generations while reconciling new, changed, and deleted rows', async () => {
  665. const path = await temporaryPath()
  666. const unchanged = header('unchanged')
  667. const changed = header('changed')
  668. const deleted = header('deleted')
  669. TestPersistence.reset([
  670. { meta: unchanged, events: messageEvents('unchanged needle') },
  671. { meta: changed, events: messageEvents('old needle') },
  672. { meta: deleted, events: messageEvents('deleted needle') },
  673. ])
  674. const first = new Context()
  675. await first.plugin(SessionStore)
  676. const firstPersistence = await first.plugin(TestPersistence)
  677. const firstSearch = await first.plugin(SessionSearchSqlite, { path })
  678. await first.sessionSearch.searchSessions({ query: 'needle' })
  679. expect(Object.fromEntries(TestPersistence.loads)).toEqual({ unchanged: 1, changed: 1, deleted: 1 })
  680. await first.sessionSearch.searchSessions({ query: 'needle' })
  681. expect(Object.fromEntries(TestPersistence.loads)).toEqual({ unchanged: 1, changed: 1, deleted: 1 })
  682. await firstSearch.dispose()
  683. await firstPersistence.dispose()
  684. const beforeDb = new DatabaseSync(path)
  685. const beforeRows = beforeDb.prepare('SELECT id, generation FROM persisted_sessions ORDER BY id').all() as Array<{ id: string; generation: number }>
  686. beforeDb.close()
  687. const before = new Map(beforeRows.map(row => [row.id, row.generation]))
  688. const added = header('added')
  689. TestPersistence.entries.delete(deleted.id)
  690. TestPersistence.set({ meta: changed, events: messageEvents('changed needle') })
  691. TestPersistence.set({ meta: added, events: messageEvents('added needle') })
  692. const second = new Context()
  693. await second.plugin(SessionStore)
  694. const secondPersistence = await second.plugin(TestPersistence)
  695. const secondSearch = await second.plugin(SessionSearchSqlite, { path })
  696. const result = await second.sessionSearch.searchSessions({ query: 'needle' })
  697. expect(result.items.map(item => item.header.id).sort()).toEqual([added.id, changed.id, unchanged.id].sort())
  698. expect(Object.fromEntries(TestPersistence.loads)).toEqual({
  699. unchanged: 1,
  700. changed: 2,
  701. deleted: 1,
  702. added: 1,
  703. })
  704. await secondSearch.dispose()
  705. await secondPersistence.dispose()
  706. const afterDb = new DatabaseSync(path)
  707. const afterRows = afterDb.prepare('SELECT id, generation FROM persisted_sessions ORDER BY id').all() as Array<{ id: string; generation: number }>
  708. afterDb.close()
  709. const after = new Map(afterRows.map(row => [row.id, row.generation]))
  710. expect(after.get(unchanged.id)).toBe(before.get(unchanged.id))
  711. expect(after.get(changed.id)).toBeGreaterThan(before.get(changed.id)!)
  712. expect(after.has(deleted.id)).toBe(false)
  713. expect(after.has(added.id)).toBe(true)
  714. })
  715. it('drops connection-local live overlays on reopen and retains persistent bases', async () => {
  716. const path = await temporaryPath()
  717. const shared = header('shared', 10)
  718. TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
  719. const first = new Context()
  720. await first.plugin(SessionStore)
  721. const persistence = await first.plugin(TestPersistence)
  722. const live = first.sessions.create(shared.id, { seed: messageEvents('live needle'), meta: { createdAt: 10 } })
  723. const search = await first.plugin(SessionSearchSqlite, { path })
  724. await expect(first.sessionSearch.searchEvents({ sessionId: live.id, query: 'live' })).resolves.toMatchObject({ items: [{}] })
  725. await search.dispose()
  726. await persistence.dispose()
  727. const second = new Context()
  728. await second.plugin(SessionStore)
  729. const persistenceAgain = await second.plugin(TestPersistence)
  730. const searchAgain = await second.plugin(SessionSearchSqlite, { path })
  731. await expect(second.sessionSearch.searchSessions({ query: 'live' })).resolves.toEqual({ items: [] })
  732. await expect(second.sessionSearch.searchSessions({ query: 'persisted' }))
  733. .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
  734. expect(TestPersistence.loads.get(shared.id)).toBe(1)
  735. await searchAgain.dispose()
  736. await persistenceAgain.dispose()
  737. })
  738. it('refreshes the stored revision after a mutating load repair', async () => {
  739. const durable = header('repair')
  740. TestPersistence.reset([{ meta: durable, events: messageEvents('before repair') }])
  741. TestPersistence.loadEffect = (entry) => {
  742. entry.events = messageEvents('repaired needle')
  743. }
  744. const ctx = await liveContext()
  745. await ctx.plugin(TestPersistence)
  746. await expect(ctx.sessionSearch.searchSessions({ query: 'repaired' }))
  747. .resolves.toMatchObject({ items: [{ header: durable }] })
  748. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  749. await ctx.sessionSearch.searchSessions({ query: 'repaired' })
  750. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  751. })
  752. it('recovers on the next search after source and SQLite transaction failures', async () => {
  753. TestPersistence.reset([{ meta: header('durable'), events: messageEvents('durable needle') }])
  754. const ctx = await liveContext()
  755. await ctx.plugin(TestPersistence)
  756. TestPersistence.failure = 'offline'
  757. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  758. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  759. const signal = new AbortController().signal
  760. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal }))
  761. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  762. TestPersistence.failure = new Error('still offline')
  763. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal }))
  764. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  765. TestPersistence.failure = undefined
  766. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' })).resolves.toMatchObject({ items: [{}] })
  767. const live = ctx.sessions.create(SessionId('live'), { seed: messageEvents('base') })
  768. await ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'base' })
  769. const db = (ctx.sessionSearch as unknown as { _db: DatabaseSync })._db
  770. db.exec('PRAGMA query_only = ON')
  771. live.append('user/message', { content: [{ type: 'text', text: 'retry needle' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
  772. await expect(ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'needle' }))
  773. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  774. db.exec('PRAGMA query_only = OFF')
  775. await expect(ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'needle' }))
  776. .resolves.toMatchObject({ items: [{ seq: 1 }] })
  777. })
  778. })
  779. describe('SQLite schema, cancellation, and real persistence integration', () => {
  780. it('resets a recognized incompatible derived schema but refuses a foreign database', async () => {
  781. const stalePath = await temporaryPath('stale.db')
  782. const stale = new DatabaseSync(stalePath)
  783. stale.exec(`PRAGMA application_id = ${SESSION_QUERY_SQLITE_APPLICATION_ID}`)
  784. stale.exec('PRAGMA user_version = 999')
  785. stale.exec('CREATE TABLE stale(value TEXT)')
  786. stale.close()
  787. const staleCtx = await liveContext({ path: stalePath })
  788. staleCtx.sessions.create(SessionId('live'), { seed: messageEvents('needle') })
  789. await staleCtx.sessionSearch.searchSessions({ query: 'needle' })
  790. await (staleCtx.sessionSearch as SessionSearchSqlite).close()
  791. const rebuilt = new DatabaseSync(stalePath)
  792. expect((rebuilt.prepare('PRAGMA user_version').get() as { user_version: number }).user_version)
  793. .toBe(SESSION_QUERY_SQLITE_SCHEMA_VERSION)
  794. expect(rebuilt.prepare("SELECT name FROM sqlite_master WHERE name = 'stale'").get()).toBeUndefined()
  795. rebuilt.close()
  796. const foreignPath = await temporaryPath('foreign.db')
  797. const foreign = new DatabaseSync(foreignPath)
  798. foreign.exec('PRAGMA journal_mode = WAL')
  799. foreign.exec('CREATE TABLE canonical(value TEXT)')
  800. foreign.exec("INSERT INTO canonical VALUES ('safe')")
  801. foreign.close()
  802. const foreignCtx = await liveContext({ path: foreignPath, journalMode: 'delete' })
  803. await expect(foreignCtx.sessionSearch.searchSessions({ query: 'needle' }))
  804. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  805. const stillForeign = new DatabaseSync(foreignPath)
  806. expect(stillForeign.prepare('SELECT value FROM canonical').get()).toEqual({ value: 'safe' })
  807. expect(stillForeign.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'wal' })
  808. stillForeign.close()
  809. await (foreignCtx.sessionSearch as SessionSearchSqlite).close()
  810. const otherAppPath = await temporaryPath('other-app.db')
  811. const otherApp = new DatabaseSync(otherAppPath)
  812. otherApp.exec('PRAGMA application_id = 123')
  813. otherApp.close()
  814. const otherAppCtx = await liveContext({ path: otherAppPath })
  815. await expect(otherAppCtx.sessionSearch.searchSessions({ query: 'needle' }))
  816. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  817. await (otherAppCtx.sessionSearch as SessionSearchSqlite).close()
  818. })
  819. it('observes asynchronous open rejection even when no query is made', async () => {
  820. const path = await temporaryPath('never-queried.db')
  821. const foreign = new DatabaseSync(path)
  822. foreign.exec('CREATE TABLE canonical(value TEXT)')
  823. foreign.close()
  824. const unhandled: unknown[] = []
  825. const onUnhandled = (reason: unknown) => { unhandled.push(reason) }
  826. process.on('unhandledRejection', onUnhandled)
  827. try {
  828. const ctx = await liveContext({ path })
  829. await new Promise<void>((resolve) => { setImmediate(resolve) })
  830. expect(unhandled).toEqual([])
  831. await (ctx.sessionSearch as SessionSearchSqlite).close()
  832. } finally {
  833. process.off('unhandledRejection', onUnhandled)
  834. }
  835. })
  836. it('cancels both queued and in-flight source waits without committing them', async () => {
  837. TestPersistence.reset()
  838. const ctx = await liveContext()
  839. await ctx.plugin(TestPersistence)
  840. const boundaryController = new AbortController()
  841. const boundary = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: boundaryController.signal })
  842. queueMicrotask(() => { boundaryController.abort() })
  843. await expect(boundary).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  844. const readyController = new AbortController()
  845. readyController.abort()
  846. const internals = ctx.sessionSearch as unknown as {
  847. _ensureReady(signal: AbortSignal): Promise<void>
  848. }
  849. await expect(internals._ensureReady(readyController.signal))
  850. .rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  851. let releaseBlocking!: () => void
  852. TestPersistence.listGate = new Promise<void>((resolve) => { releaseBlocking = resolve })
  853. let markBlockingStarted!: () => void
  854. const blockingStarted = new Promise<void>((resolve) => { markBlockingStarted = resolve })
  855. TestPersistence.listStarted = () => {
  856. TestPersistence.listStarted = undefined
  857. markBlockingStarted()
  858. }
  859. const blocking = ctx.sessionSearch.searchSessions({ query: 'needle' })
  860. await blockingStarted
  861. const queuedController = new AbortController()
  862. const queued = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: queuedController.signal })
  863. queuedController.abort()
  864. await expect(queued).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  865. releaseBlocking()
  866. await expect(blocking).resolves.toEqual({ items: [] })
  867. TestPersistence.set({
  868. meta: header('uncommitted'),
  869. events: messageEvents('durable needle'),
  870. })
  871. let releaseActive!: () => void
  872. TestPersistence.listGate = new Promise<void>((resolve) => { releaseActive = resolve })
  873. let markActiveStarted!: () => void
  874. const activeStarted = new Promise<void>((resolve) => { markActiveStarted = resolve })
  875. TestPersistence.listStarted = () => {
  876. TestPersistence.listStarted = undefined
  877. markActiveStarted()
  878. }
  879. const activeController = new AbortController()
  880. const active = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: activeController.signal })
  881. await activeStarted
  882. activeController.abort()
  883. await expect(active).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  884. releaseActive()
  885. const db = (ctx.sessionSearch as unknown as { _db: DatabaseSync })._db
  886. expect(db.prepare('SELECT COUNT(*) AS count FROM persisted_sessions').get()).toEqual({ count: 0 })
  887. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  888. .resolves.toMatchObject({ items: [{ header: { id: SessionId('uncommitted') } }] })
  889. })
  890. it('rejects queued and future work when close waits for an accepted operation', async () => {
  891. TestPersistence.reset()
  892. let release!: () => void
  893. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  894. let markStarted!: () => void
  895. const started = new Promise<void>((resolve) => { markStarted = resolve })
  896. TestPersistence.listStarted = () => {
  897. TestPersistence.listStarted = undefined
  898. markStarted()
  899. }
  900. const ctx = await liveContext()
  901. await ctx.plugin(TestPersistence)
  902. const search = ctx.sessionSearch as SessionSearchSqlite
  903. const accepted = search.searchSessions({ query: 'needle' })
  904. await started
  905. const queued = search.searchSessions({ query: 'needle' })
  906. const closing = search.close()
  907. const repeatedClose = search.close()
  908. expect(repeatedClose).toBe(closing)
  909. release()
  910. await expect(accepted).resolves.toEqual({ items: [] })
  911. await expect(queued).rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  912. await Promise.all([closing, repeatedClose])
  913. await expect(search.searchSessions({ query: 'needle' }))
  914. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  915. expect(search.close()).toBe(closing)
  916. })
  917. it('awaits optional-persistence child-fiber quiescence on disposal', async () => {
  918. TestPersistence.reset()
  919. const ctx = new Context()
  920. await ctx.plugin(SessionStore)
  921. const search = await ctx.plugin(SessionSearchSqlite, { path: ':memory:' })
  922. const persistence = await ctx.plugin(TestPersistence)
  923. const optional = (ctx.sessionSearch as unknown as {
  924. _optionalPersistenceFiber: Fiber
  925. })._optionalPersistenceFiber
  926. let release!: () => void
  927. const cleanup = new Promise<void>((resolve) => { release = resolve })
  928. optional.ctx.effect(() => () => cleanup)
  929. let settled = false
  930. const disposing = search.dispose().then(() => { settled = true })
  931. await Promise.resolve()
  932. expect(settled).toBe(false)
  933. release()
  934. await disposing
  935. await persistence.dispose()
  936. })
  937. it('combines the real SQLite persistence backend with the real search service keylessly', async () => {
  938. const persistencePath = await temporaryPath('canonical.db')
  939. const searchPath = await temporaryPath('derived.db')
  940. const ctx = new Context()
  941. await ctx.plugin(SessionStore)
  942. const persistence = await ctx.plugin(SessionPersistenceSqlite, { path: persistencePath })
  943. const search = await ctx.plugin(SessionSearchSqlite, { path: searchPath })
  944. const meta = header('real', 10, { cwd: '/work' })
  945. await ctx.sessionPersistence.create(meta)
  946. await ctx.sessionPersistence.append(meta.id, messageEvents('real SQLite needle'))
  947. await expect(ctx.sessionSearch.searchSessions({ query: 'SQLite needle' }))
  948. .resolves.toMatchObject({ items: [{ header: meta, persisted: true, live: false }] })
  949. await expect(ctx.sessionSearch.searchEvents({ sessionId: meta.id, query: 'SQLite needle' }))
  950. .resolves.toMatchObject({ items: [{ sessionId: meta.id, seq: 0 }] })
  951. await expect(ctx.sessionSearch.searchEvents({ sessionId: SessionId('absent'), query: 'needle' }))
  952. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  953. await search.dispose()
  954. await expect(ctx.sessionPersistence.load(meta.id)).resolves.toMatchObject({ meta, events: [{ seq: 0 }] })
  955. await persistence.dispose()
  956. })
  957. it('reconciles colliding local revisions when a derived index reopens against another SQLite store', async () => {
  958. const persistencePathA = await temporaryPath('canonical-a.db')
  959. const persistencePathB = await temporaryPath('canonical-b.db')
  960. const searchPath = await temporaryPath('derived-collision.db')
  961. const shared = header('same-id', 10)
  962. const first = new Context()
  963. await first.plugin(SessionStore)
  964. const persistenceA = await first.plugin(SessionPersistenceSqlite, { path: persistencePathA })
  965. await first.sessionPersistence.create(shared)
  966. await first.sessionPersistence.append(shared.id, messageEvents('alpha source'))
  967. const loadA = vi.spyOn(first.sessionPersistence, 'load')
  968. const searchA = await first.plugin(SessionSearchSqlite, { path: searchPath })
  969. await expect(first.sessionSearch.searchSessions({ query: 'alpha' }))
  970. .resolves.toMatchObject({ items: [{ header: shared }] })
  971. expect(loadA).toHaveBeenCalledTimes(1)
  972. await searchA.dispose()
  973. await persistenceA.dispose()
  974. const reopened = new Context()
  975. await reopened.plugin(SessionStore)
  976. const persistenceAAgain = await reopened.plugin(SessionPersistenceSqlite, { path: persistencePathA })
  977. const reopenedLoad = vi.spyOn(reopened.sessionPersistence, 'load')
  978. const searchAAgain = await reopened.plugin(SessionSearchSqlite, { path: searchPath })
  979. await expect(reopened.sessionSearch.searchSessions({ query: 'alpha' }))
  980. .resolves.toMatchObject({ items: [{ header: shared }] })
  981. expect(reopenedLoad).not.toHaveBeenCalled()
  982. await searchAAgain.dispose()
  983. await persistenceAAgain.dispose()
  984. const second = new Context()
  985. await second.plugin(SessionStore)
  986. const persistenceB = await second.plugin(SessionPersistenceSqlite, { path: persistencePathB })
  987. await second.sessionPersistence.create(shared)
  988. await second.sessionPersistence.append(shared.id, messageEvents('bravo source'))
  989. const loadB = vi.spyOn(second.sessionPersistence, 'load')
  990. const searchB = await second.plugin(SessionSearchSqlite, { path: searchPath })
  991. await expect(second.sessionSearch.searchSessions({ query: 'bravo' }))
  992. .resolves.toMatchObject({ items: [{ header: shared }] })
  993. await expect(second.sessionSearch.searchSessions({ query: 'alpha' })).resolves.toEqual({ items: [] })
  994. expect(loadB).toHaveBeenCalledTimes(1)
  995. await searchB.dispose()
  996. await persistenceB.dispose()
  997. })
  998. })