sqlite.spec.ts 51 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089
  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. })
  407. describe('SQLite reconciliation and source lifecycle', () => {
  408. it('owns queued request and filter values before waiting for the serializer', async () => {
  409. const durable = header('owned')
  410. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  411. const ctx = await liveContext()
  412. const persistence = await ctx.plugin(TestPersistence)
  413. let release!: () => void
  414. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  415. let markStarted!: () => void
  416. const started = new Promise<void>((resolve) => { markStarted = resolve })
  417. TestPersistence.listStarted = () => {
  418. TestPersistence.listStarted = undefined
  419. markStarted()
  420. }
  421. const blocking = ctx.sessionSearch.searchSessions({ query: 'needle' })
  422. await started
  423. const availability: SessionAvailability[] = ['persisted']
  424. const request: SessionSearchRequest = {
  425. query: 'needle',
  426. sessionFilters: [{ kind: 'availability', values: availability }],
  427. }
  428. const queued = ctx.sessionSearch.searchSessions(request)
  429. request.query = 'absent'
  430. availability[0] = 'live'
  431. release()
  432. await expect(blocking).resolves.toMatchObject({ items: [{ header: durable }] })
  433. await expect(queued).resolves.toMatchObject({ items: [{ header: durable }] })
  434. await persistence.dispose()
  435. })
  436. it('mounts persistence dynamically, shadows with TEMP live rows, reveals, and hides on unmount', async () => {
  437. const shared = header('shared', 10, { cwd: '/work' })
  438. const durable = header('durable', 5)
  439. TestPersistence.reset([
  440. { meta: shared, events: messageEvents('persisted needle') },
  441. { meta: durable, events: messageEvents('durable needle') },
  442. ])
  443. const ctx = await liveContext()
  444. await expect(ctx.sessionSearch.searchSessions({ query: 'durable' })).resolves.toEqual({ items: [] })
  445. const persistenceFiber = await ctx.plugin(TestPersistence)
  446. await expect(ctx.sessionSearch.searchSessions({ query: 'durable' }))
  447. .resolves.toMatchObject({ items: [{ header: durable, live: false, persisted: true }] })
  448. const live = ctx.sessions.prepare(shared.id, { meta: { createdAt: 10, cwd: '/work' } })
  449. live.append('user/message', { content: [{ type: 'text', text: 'live needle' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
  450. const detach = ctx.sessions.enter(live)
  451. ctx.sessions.announce(live)
  452. await expect(ctx.sessionSearch.searchSessions({ query: 'persisted' })).resolves.toEqual({ items: [] })
  453. await expect(ctx.sessionSearch.searchSessions({ query: 'live' }))
  454. .resolves.toMatchObject({ items: [{ header: shared, live: true, persisted: true }] })
  455. detach()
  456. await expect(ctx.sessionSearch.searchSessions({ query: 'persisted' }))
  457. .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
  458. await persistenceFiber.dispose()
  459. await expect(ctx.sessionSearch.searchSessions({ query: 'durable' })).resolves.toEqual({ items: [] })
  460. await expect(ctx.sessionSearch.searchEvents({ sessionId: durable.id, query: 'needle' }))
  461. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  462. })
  463. it('uses the reconciled persistence binding through the query boundary', async () => {
  464. const durable = header('post-reconcile-unmount')
  465. TestPersistence.reset([{ meta: durable, events: [
  466. ...messageEvents('durable needle', 1),
  467. { ...messageEvents('durable needle again', 2)[0]!, seq: 1 },
  468. ] }])
  469. const ctx = await liveContext({ path: ':memory:', defaultLimit: 1, maxLimit: 2 })
  470. const persistence = await ctx.plugin(TestPersistence)
  471. const internals = ctx.sessionSearch as unknown as {
  472. _reconcile(signal: AbortSignal | undefined): Promise<{
  473. identity: symbol
  474. service?: SessionPersistence
  475. }>
  476. }
  477. const reconcile = internals._reconcile.bind(internals)
  478. const boundary = vi.spyOn(internals, '_reconcile').mockImplementation(async (signal) => {
  479. const binding = await reconcile(signal)
  480. await persistence.dispose()
  481. return binding
  482. })
  483. const page = await ctx.sessionSearch.searchEvents({
  484. sessionId: durable.id,
  485. query: 'needle',
  486. limit: 1,
  487. })
  488. expect(page.items).toMatchObject([{ sessionId: durable.id }])
  489. expect(page.nextCursor).toEqual(expect.any(String))
  490. boundary.mockRestore()
  491. await expect(ctx.sessionSearch.searchEvents({ sessionId: durable.id, query: 'needle' }))
  492. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  493. })
  494. it('discards a stale list rejection when persistence unmounts during observation', async () => {
  495. const durable = header('racing')
  496. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  497. const ctx = await liveContext()
  498. const persistenceFiber = await ctx.plugin(TestPersistence)
  499. let release!: () => void
  500. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  501. let markStarted!: () => void
  502. const started = new Promise<void>((resolve) => { markStarted = resolve })
  503. TestPersistence.listStarted = () => {
  504. TestPersistence.listStarted = undefined
  505. markStarted()
  506. }
  507. const search = ctx.sessionSearch.searchSessions({ query: 'needle' })
  508. await started
  509. await persistenceFiber.dispose()
  510. TestPersistence.failure = new Error('stale backend rejection')
  511. release()
  512. await expect(search).resolves.toEqual({ items: [] })
  513. })
  514. it('retries against a replacement after the prior binding rejects', async () => {
  515. const durable = header('replacement')
  516. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  517. const ctx = await liveContext()
  518. const prior = await ctx.plugin(TestPersistence)
  519. let rejectPrior!: (reason: unknown) => void
  520. TestPersistence.listGate = new Promise<void>((_resolve, reject) => { rejectPrior = reject })
  521. let markStarted!: () => void
  522. const started = new Promise<void>((resolve) => { markStarted = resolve })
  523. TestPersistence.listStarted = () => {
  524. TestPersistence.listStarted = undefined
  525. markStarted()
  526. }
  527. const search = ctx.sessionSearch.searchSessions({ query: 'needle' })
  528. await started
  529. await prior.dispose()
  530. TestPersistence.listGate = undefined
  531. const replacement = await ctx.plugin(TestPersistence)
  532. rejectPrior(new Error('stale prior binding'))
  533. await expect(search).resolves.toMatchObject({ items: [{ header: durable }] })
  534. await replacement.dispose()
  535. })
  536. it('reloads a replacement source even when its opaque revisions collide', async () => {
  537. const durable = header('colliding-replacement')
  538. TestPersistence.reset([{ meta: durable, events: messageEvents('old content') }])
  539. const revision = TestPersistence.revisions.get(durable.id)!
  540. const ctx = await liveContext()
  541. const prior = await ctx.plugin(TestPersistence)
  542. await expect(ctx.sessionSearch.searchSessions({ query: 'old' }))
  543. .resolves.toMatchObject({ items: [{ header: durable }] })
  544. await prior.dispose()
  545. TestPersistence.set({ meta: durable, events: messageEvents('new needle') })
  546. TestPersistence.revisions.set(durable.id, revision)
  547. const replacement = await ctx.plugin(TestPersistence)
  548. const page = await ctx.sessionSearch.searchSessions({ query: 'new needle' })
  549. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  550. expect(page).toMatchObject({ items: [{ header: durable }] })
  551. await expect(ctx.sessionSearch.searchSessions({ query: 'old' })).resolves.toEqual({ items: [] })
  552. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  553. await replacement.dispose()
  554. })
  555. it('retries when a successful observation belongs to a source unmounted during listing', async () => {
  556. const durable = header('successful-unmount')
  557. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  558. const ctx = await liveContext()
  559. const persistence = await ctx.plugin(TestPersistence)
  560. let lists = 0
  561. TestPersistence.snapshotEffect = async () => {
  562. lists += 1
  563. if (lists === 2) await persistence.dispose()
  564. }
  565. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' })).resolves.toEqual({ items: [] })
  566. expect(lists).toBe(2)
  567. })
  568. it('retries when the snapshot population changes during observation', async () => {
  569. const first = header('first')
  570. const added = header('added-during-list')
  571. TestPersistence.reset([{ meta: first, events: messageEvents('first needle') }])
  572. const ctx = await liveContext()
  573. await ctx.plugin(TestPersistence)
  574. TestPersistence.snapshotEffect = () => {
  575. TestPersistence.snapshotEffect = undefined
  576. TestPersistence.set({ meta: added, events: messageEvents('added needle') })
  577. }
  578. const page = await ctx.sessionSearch.searchSessions({ query: 'needle' })
  579. expect(page.items.map(item => item.header.id).sort()).toEqual([added.id, first.id].sort())
  580. expect(TestPersistence.loads.get(first.id)).toBe(2)
  581. expect(TestPersistence.loads.get(added.id)).toBe(1)
  582. })
  583. it('fails after one retry when persistence snapshots keep changing', async () => {
  584. const durable = header('continuous-mutation')
  585. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  586. const ctx = await liveContext()
  587. await ctx.plugin(TestPersistence)
  588. let lists = 0
  589. TestPersistence.snapshotEffect = () => {
  590. lists += 1
  591. TestPersistence.set({ meta: durable, events: messageEvents(`durable needle ${lists}`) })
  592. }
  593. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  594. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  595. expect(lists).toBe(4)
  596. })
  597. it('retries if the persistence binding changes while live sessions are observed', async () => {
  598. const durable = header('live-boundary-retry')
  599. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  600. const ctx = await liveContext()
  601. await ctx.plugin(TestPersistence)
  602. const internals = ctx.sessionSearch as unknown as {
  603. _persistenceBinding: { identity: symbol; service?: SessionPersistence }
  604. }
  605. const originalList = ctx.sessions.list.bind(ctx.sessions)
  606. let bumped = false
  607. const list = vi.spyOn(ctx.sessions, 'list').mockImplementation(() => {
  608. if (!bumped) {
  609. bumped = true
  610. internals._persistenceBinding = {
  611. ...internals._persistenceBinding,
  612. identity: Symbol(),
  613. }
  614. }
  615. return originalList()
  616. })
  617. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  618. .resolves.toMatchObject({ items: [{ header: durable }] })
  619. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  620. list.mockRestore()
  621. })
  622. it('rejects malformed snapshots and preserves typed persistence failures', async () => {
  623. const durable = header('invalid-snapshot')
  624. TestPersistence.reset([{ meta: durable, events: messageEvents('durable needle') }])
  625. const ctx = await liveContext()
  626. await ctx.plugin(TestPersistence)
  627. TestPersistence.snapshotOverride = () => 'not-an-array' as never
  628. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  629. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  630. TestPersistence.snapshotOverride = () => [{ header: durable, revision: 1 as never }]
  631. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  632. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  633. TestPersistence.snapshotOverride = () => [
  634. { header: durable, revision: SessionPersistenceRevision('duplicate:1') },
  635. { header: durable, revision: SessionPersistenceRevision('duplicate:2') },
  636. ]
  637. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  638. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  639. TestPersistence.snapshotOverride = undefined
  640. const typed = new SessionQueryError('typed persistence failure', 'SESSION_QUERY_PERSISTENCE_FAILED')
  641. TestPersistence.failure = typed
  642. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' })).rejects.toBe(typed)
  643. })
  644. it('rejects immutable header conflicts between live and persisted sources', async () => {
  645. const shared = header('conflict', 10)
  646. TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
  647. const ctx = await liveContext()
  648. await ctx.plugin(TestPersistence)
  649. ctx.sessions.create(shared.id, { seed: messageEvents('live needle'), meta: { createdAt: 11 } })
  650. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  651. .rejects.toThrow(expectCode('SESSION_QUERY_SOURCE_CONFLICT'))
  652. })
  653. it('preserves unchanged persisted generations while reconciling new, changed, and deleted rows', async () => {
  654. const path = await temporaryPath()
  655. const unchanged = header('unchanged')
  656. const changed = header('changed')
  657. const deleted = header('deleted')
  658. TestPersistence.reset([
  659. { meta: unchanged, events: messageEvents('unchanged needle') },
  660. { meta: changed, events: messageEvents('old needle') },
  661. { meta: deleted, events: messageEvents('deleted needle') },
  662. ])
  663. const first = new Context()
  664. await first.plugin(SessionStore)
  665. const firstPersistence = await first.plugin(TestPersistence)
  666. const firstSearch = await first.plugin(SessionSearchSqlite, { path })
  667. await first.sessionSearch.searchSessions({ query: 'needle' })
  668. expect(Object.fromEntries(TestPersistence.loads)).toEqual({ unchanged: 1, changed: 1, deleted: 1 })
  669. await first.sessionSearch.searchSessions({ query: 'needle' })
  670. expect(Object.fromEntries(TestPersistence.loads)).toEqual({ unchanged: 1, changed: 1, deleted: 1 })
  671. await firstSearch.dispose()
  672. await firstPersistence.dispose()
  673. const beforeDb = new DatabaseSync(path)
  674. const beforeRows = beforeDb.prepare('SELECT id, generation FROM persisted_sessions ORDER BY id').all() as Array<{ id: string; generation: number }>
  675. beforeDb.close()
  676. const before = new Map(beforeRows.map(row => [row.id, row.generation]))
  677. const added = header('added')
  678. TestPersistence.entries.delete(deleted.id)
  679. TestPersistence.set({ meta: changed, events: messageEvents('changed needle') })
  680. TestPersistence.set({ meta: added, events: messageEvents('added needle') })
  681. const second = new Context()
  682. await second.plugin(SessionStore)
  683. const secondPersistence = await second.plugin(TestPersistence)
  684. const secondSearch = await second.plugin(SessionSearchSqlite, { path })
  685. const result = await second.sessionSearch.searchSessions({ query: 'needle' })
  686. expect(result.items.map(item => item.header.id).sort()).toEqual([added.id, changed.id, unchanged.id].sort())
  687. expect(Object.fromEntries(TestPersistence.loads)).toEqual({
  688. unchanged: 1,
  689. changed: 2,
  690. deleted: 1,
  691. added: 1,
  692. })
  693. await secondSearch.dispose()
  694. await secondPersistence.dispose()
  695. const afterDb = new DatabaseSync(path)
  696. const afterRows = afterDb.prepare('SELECT id, generation FROM persisted_sessions ORDER BY id').all() as Array<{ id: string; generation: number }>
  697. afterDb.close()
  698. const after = new Map(afterRows.map(row => [row.id, row.generation]))
  699. expect(after.get(unchanged.id)).toBe(before.get(unchanged.id))
  700. expect(after.get(changed.id)).toBeGreaterThan(before.get(changed.id)!)
  701. expect(after.has(deleted.id)).toBe(false)
  702. expect(after.has(added.id)).toBe(true)
  703. })
  704. it('drops connection-local live overlays on reopen and retains persistent bases', async () => {
  705. const path = await temporaryPath()
  706. const shared = header('shared', 10)
  707. TestPersistence.reset([{ meta: shared, events: messageEvents('persisted needle') }])
  708. const first = new Context()
  709. await first.plugin(SessionStore)
  710. const persistence = await first.plugin(TestPersistence)
  711. const live = first.sessions.create(shared.id, { seed: messageEvents('live needle'), meta: { createdAt: 10 } })
  712. const search = await first.plugin(SessionSearchSqlite, { path })
  713. await expect(first.sessionSearch.searchEvents({ sessionId: live.id, query: 'live' })).resolves.toMatchObject({ items: [{}] })
  714. await search.dispose()
  715. await persistence.dispose()
  716. const second = new Context()
  717. await second.plugin(SessionStore)
  718. const persistenceAgain = await second.plugin(TestPersistence)
  719. const searchAgain = await second.plugin(SessionSearchSqlite, { path })
  720. await expect(second.sessionSearch.searchSessions({ query: 'live' })).resolves.toEqual({ items: [] })
  721. await expect(second.sessionSearch.searchSessions({ query: 'persisted' }))
  722. .resolves.toMatchObject({ items: [{ header: shared, live: false, persisted: true }] })
  723. expect(TestPersistence.loads.get(shared.id)).toBe(1)
  724. await searchAgain.dispose()
  725. await persistenceAgain.dispose()
  726. })
  727. it('refreshes the stored revision after a mutating load repair', async () => {
  728. const durable = header('repair')
  729. TestPersistence.reset([{ meta: durable, events: messageEvents('before repair') }])
  730. TestPersistence.loadEffect = (entry) => {
  731. entry.events = messageEvents('repaired needle')
  732. }
  733. const ctx = await liveContext()
  734. await ctx.plugin(TestPersistence)
  735. await expect(ctx.sessionSearch.searchSessions({ query: 'repaired' }))
  736. .resolves.toMatchObject({ items: [{ header: durable }] })
  737. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  738. await ctx.sessionSearch.searchSessions({ query: 'repaired' })
  739. expect(TestPersistence.loads.get(durable.id)).toBe(2)
  740. })
  741. it('recovers on the next search after source and SQLite transaction failures', async () => {
  742. TestPersistence.reset([{ meta: header('durable'), events: messageEvents('durable needle') }])
  743. const ctx = await liveContext()
  744. await ctx.plugin(TestPersistence)
  745. TestPersistence.failure = 'offline'
  746. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  747. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  748. const signal = new AbortController().signal
  749. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal }))
  750. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  751. TestPersistence.failure = new Error('still offline')
  752. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal }))
  753. .rejects.toThrow(expectCode('SESSION_QUERY_PERSISTENCE_FAILED'))
  754. TestPersistence.failure = undefined
  755. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' })).resolves.toMatchObject({ items: [{}] })
  756. const live = ctx.sessions.create(SessionId('live'), { seed: messageEvents('base') })
  757. await ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'base' })
  758. const db = (ctx.sessionSearch as unknown as { _db: DatabaseSync })._db
  759. db.exec('PRAGMA query_only = ON')
  760. live.append('user/message', { content: [{ type: 'text', text: 'retry needle' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
  761. await expect(ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'needle' }))
  762. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  763. db.exec('PRAGMA query_only = OFF')
  764. await expect(ctx.sessionSearch.searchEvents({ sessionId: live.id, query: 'needle' }))
  765. .resolves.toMatchObject({ items: [{ seq: 1 }] })
  766. })
  767. })
  768. describe('SQLite schema, cancellation, and real persistence integration', () => {
  769. it('resets a recognized incompatible derived schema but refuses a foreign database', async () => {
  770. const stalePath = await temporaryPath('stale.db')
  771. const stale = new DatabaseSync(stalePath)
  772. stale.exec(`PRAGMA application_id = ${SESSION_QUERY_SQLITE_APPLICATION_ID}`)
  773. stale.exec('PRAGMA user_version = 999')
  774. stale.exec('CREATE TABLE stale(value TEXT)')
  775. stale.close()
  776. const staleCtx = await liveContext({ path: stalePath })
  777. staleCtx.sessions.create(SessionId('live'), { seed: messageEvents('needle') })
  778. await staleCtx.sessionSearch.searchSessions({ query: 'needle' })
  779. await (staleCtx.sessionSearch as SessionSearchSqlite).close()
  780. const rebuilt = new DatabaseSync(stalePath)
  781. expect((rebuilt.prepare('PRAGMA user_version').get() as { user_version: number }).user_version)
  782. .toBe(SESSION_QUERY_SQLITE_SCHEMA_VERSION)
  783. expect(rebuilt.prepare("SELECT name FROM sqlite_master WHERE name = 'stale'").get()).toBeUndefined()
  784. rebuilt.close()
  785. const foreignPath = await temporaryPath('foreign.db')
  786. const foreign = new DatabaseSync(foreignPath)
  787. foreign.exec('PRAGMA journal_mode = WAL')
  788. foreign.exec('CREATE TABLE canonical(value TEXT)')
  789. foreign.exec("INSERT INTO canonical VALUES ('safe')")
  790. foreign.close()
  791. const foreignCtx = await liveContext({ path: foreignPath, journalMode: 'delete' })
  792. await expect(foreignCtx.sessionSearch.searchSessions({ query: 'needle' }))
  793. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  794. const stillForeign = new DatabaseSync(foreignPath)
  795. expect(stillForeign.prepare('SELECT value FROM canonical').get()).toEqual({ value: 'safe' })
  796. expect(stillForeign.prepare('PRAGMA journal_mode').get()).toEqual({ journal_mode: 'wal' })
  797. stillForeign.close()
  798. await (foreignCtx.sessionSearch as SessionSearchSqlite).close()
  799. const otherAppPath = await temporaryPath('other-app.db')
  800. const otherApp = new DatabaseSync(otherAppPath)
  801. otherApp.exec('PRAGMA application_id = 123')
  802. otherApp.close()
  803. const otherAppCtx = await liveContext({ path: otherAppPath })
  804. await expect(otherAppCtx.sessionSearch.searchSessions({ query: 'needle' }))
  805. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  806. await (otherAppCtx.sessionSearch as SessionSearchSqlite).close()
  807. })
  808. it('observes asynchronous open rejection even when no query is made', async () => {
  809. const path = await temporaryPath('never-queried.db')
  810. const foreign = new DatabaseSync(path)
  811. foreign.exec('CREATE TABLE canonical(value TEXT)')
  812. foreign.close()
  813. const unhandled: unknown[] = []
  814. const onUnhandled = (reason: unknown) => { unhandled.push(reason) }
  815. process.on('unhandledRejection', onUnhandled)
  816. try {
  817. const ctx = await liveContext({ path })
  818. await new Promise<void>((resolve) => { setImmediate(resolve) })
  819. expect(unhandled).toEqual([])
  820. await (ctx.sessionSearch as SessionSearchSqlite).close()
  821. } finally {
  822. process.off('unhandledRejection', onUnhandled)
  823. }
  824. })
  825. it('cancels both queued and in-flight source waits without committing them', async () => {
  826. TestPersistence.reset()
  827. const ctx = await liveContext()
  828. await ctx.plugin(TestPersistence)
  829. const boundaryController = new AbortController()
  830. const boundary = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: boundaryController.signal })
  831. queueMicrotask(() => { boundaryController.abort() })
  832. await expect(boundary).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  833. const readyController = new AbortController()
  834. readyController.abort()
  835. const internals = ctx.sessionSearch as unknown as {
  836. _ensureReady(signal: AbortSignal): Promise<void>
  837. }
  838. await expect(internals._ensureReady(readyController.signal))
  839. .rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  840. let releaseBlocking!: () => void
  841. TestPersistence.listGate = new Promise<void>((resolve) => { releaseBlocking = resolve })
  842. let markBlockingStarted!: () => void
  843. const blockingStarted = new Promise<void>((resolve) => { markBlockingStarted = resolve })
  844. TestPersistence.listStarted = () => {
  845. TestPersistence.listStarted = undefined
  846. markBlockingStarted()
  847. }
  848. const blocking = ctx.sessionSearch.searchSessions({ query: 'needle' })
  849. await blockingStarted
  850. const queuedController = new AbortController()
  851. const queued = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: queuedController.signal })
  852. queuedController.abort()
  853. await expect(queued).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  854. releaseBlocking()
  855. await expect(blocking).resolves.toEqual({ items: [] })
  856. TestPersistence.set({
  857. meta: header('uncommitted'),
  858. events: messageEvents('durable needle'),
  859. })
  860. let releaseActive!: () => void
  861. TestPersistence.listGate = new Promise<void>((resolve) => { releaseActive = resolve })
  862. let markActiveStarted!: () => void
  863. const activeStarted = new Promise<void>((resolve) => { markActiveStarted = resolve })
  864. TestPersistence.listStarted = () => {
  865. TestPersistence.listStarted = undefined
  866. markActiveStarted()
  867. }
  868. const activeController = new AbortController()
  869. const active = ctx.sessionSearch.searchSessions({ query: 'needle' }, { signal: activeController.signal })
  870. await activeStarted
  871. activeController.abort()
  872. await expect(active).rejects.toThrow(expectCode('SESSION_QUERY_ABORTED'))
  873. releaseActive()
  874. const db = (ctx.sessionSearch as unknown as { _db: DatabaseSync })._db
  875. expect(db.prepare('SELECT COUNT(*) AS count FROM persisted_sessions').get()).toEqual({ count: 0 })
  876. await expect(ctx.sessionSearch.searchSessions({ query: 'needle' }))
  877. .resolves.toMatchObject({ items: [{ header: { id: SessionId('uncommitted') } }] })
  878. })
  879. it('rejects queued and future work when close waits for an accepted operation', async () => {
  880. TestPersistence.reset()
  881. let release!: () => void
  882. TestPersistence.listGate = new Promise<void>((resolve) => { release = resolve })
  883. let markStarted!: () => void
  884. const started = new Promise<void>((resolve) => { markStarted = resolve })
  885. TestPersistence.listStarted = () => {
  886. TestPersistence.listStarted = undefined
  887. markStarted()
  888. }
  889. const ctx = await liveContext()
  890. await ctx.plugin(TestPersistence)
  891. const search = ctx.sessionSearch as SessionSearchSqlite
  892. const accepted = search.searchSessions({ query: 'needle' })
  893. await started
  894. const queued = search.searchSessions({ query: 'needle' })
  895. const closing = search.close()
  896. const repeatedClose = search.close()
  897. expect(repeatedClose).toBe(closing)
  898. release()
  899. await expect(accepted).resolves.toEqual({ items: [] })
  900. await expect(queued).rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  901. await Promise.all([closing, repeatedClose])
  902. await expect(search.searchSessions({ query: 'needle' }))
  903. .rejects.toThrow(expectCode('SESSION_QUERY_INDEX_FAILED'))
  904. expect(search.close()).toBe(closing)
  905. })
  906. it('awaits optional-persistence child-fiber quiescence on disposal', async () => {
  907. TestPersistence.reset()
  908. const ctx = new Context()
  909. await ctx.plugin(SessionStore)
  910. const search = await ctx.plugin(SessionSearchSqlite, { path: ':memory:' })
  911. const persistence = await ctx.plugin(TestPersistence)
  912. const optional = (ctx.sessionSearch as unknown as {
  913. _optionalPersistenceFiber: Fiber
  914. })._optionalPersistenceFiber
  915. let release!: () => void
  916. const cleanup = new Promise<void>((resolve) => { release = resolve })
  917. optional.ctx.effect(() => () => cleanup)
  918. let settled = false
  919. const disposing = search.dispose().then(() => { settled = true })
  920. await Promise.resolve()
  921. expect(settled).toBe(false)
  922. release()
  923. await disposing
  924. await persistence.dispose()
  925. })
  926. it('combines the real SQLite persistence backend with the real search service keylessly', async () => {
  927. const persistencePath = await temporaryPath('canonical.db')
  928. const searchPath = await temporaryPath('derived.db')
  929. const ctx = new Context()
  930. await ctx.plugin(SessionStore)
  931. const persistence = await ctx.plugin(SessionPersistenceSqlite, { path: persistencePath })
  932. const search = await ctx.plugin(SessionSearchSqlite, { path: searchPath })
  933. const meta = header('real', 10, { cwd: '/work' })
  934. await ctx.sessionPersistence.create(meta)
  935. await ctx.sessionPersistence.append(meta.id, messageEvents('real SQLite needle'))
  936. await expect(ctx.sessionSearch.searchSessions({ query: 'SQLite needle' }))
  937. .resolves.toMatchObject({ items: [{ header: meta, persisted: true, live: false }] })
  938. await expect(ctx.sessionSearch.searchEvents({ sessionId: meta.id, query: 'SQLite needle' }))
  939. .resolves.toMatchObject({ items: [{ sessionId: meta.id, seq: 0 }] })
  940. await expect(ctx.sessionSearch.searchEvents({ sessionId: SessionId('absent'), query: 'needle' }))
  941. .rejects.toThrow(expectCode('SESSION_QUERY_SESSION_NOT_FOUND'))
  942. await search.dispose()
  943. await expect(ctx.sessionPersistence.load(meta.id)).resolves.toMatchObject({ meta, events: [{ seq: 0 }] })
  944. await persistence.dispose()
  945. })
  946. it('reconciles colliding local revisions when a derived index reopens against another SQLite store', async () => {
  947. const persistencePathA = await temporaryPath('canonical-a.db')
  948. const persistencePathB = await temporaryPath('canonical-b.db')
  949. const searchPath = await temporaryPath('derived-collision.db')
  950. const shared = header('same-id', 10)
  951. const first = new Context()
  952. await first.plugin(SessionStore)
  953. const persistenceA = await first.plugin(SessionPersistenceSqlite, { path: persistencePathA })
  954. await first.sessionPersistence.create(shared)
  955. await first.sessionPersistence.append(shared.id, messageEvents('alpha source'))
  956. const loadA = vi.spyOn(first.sessionPersistence, 'load')
  957. const searchA = await first.plugin(SessionSearchSqlite, { path: searchPath })
  958. await expect(first.sessionSearch.searchSessions({ query: 'alpha' }))
  959. .resolves.toMatchObject({ items: [{ header: shared }] })
  960. expect(loadA).toHaveBeenCalledTimes(1)
  961. await searchA.dispose()
  962. await persistenceA.dispose()
  963. const reopened = new Context()
  964. await reopened.plugin(SessionStore)
  965. const persistenceAAgain = await reopened.plugin(SessionPersistenceSqlite, { path: persistencePathA })
  966. const reopenedLoad = vi.spyOn(reopened.sessionPersistence, 'load')
  967. const searchAAgain = await reopened.plugin(SessionSearchSqlite, { path: searchPath })
  968. await expect(reopened.sessionSearch.searchSessions({ query: 'alpha' }))
  969. .resolves.toMatchObject({ items: [{ header: shared }] })
  970. expect(reopenedLoad).not.toHaveBeenCalled()
  971. await searchAAgain.dispose()
  972. await persistenceAAgain.dispose()
  973. const second = new Context()
  974. await second.plugin(SessionStore)
  975. const persistenceB = await second.plugin(SessionPersistenceSqlite, { path: persistencePathB })
  976. await second.sessionPersistence.create(shared)
  977. await second.sessionPersistence.append(shared.id, messageEvents('bravo source'))
  978. const loadB = vi.spyOn(second.sessionPersistence, 'load')
  979. const searchB = await second.plugin(SessionSearchSqlite, { path: searchPath })
  980. await expect(second.sessionSearch.searchSessions({ query: 'bravo' }))
  981. .resolves.toMatchObject({ items: [{ header: shared }] })
  982. await expect(second.sessionSearch.searchSessions({ query: 'alpha' })).resolves.toEqual({ items: [] })
  983. expect(loadB).toHaveBeenCalledTimes(1)
  984. await searchB.dispose()
  985. await persistenceB.dispose()
  986. })
  987. })