sqlite.spec.ts 49 KB

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