sqlite.spec.ts 56 KB

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