sqlite.spec.ts 75 KB

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