sqlite.spec.ts 24 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473
  1. import { afterEach, describe, expect, it } from 'vitest'
  2. import { Context } from 'cordis'
  3. import { mkdtemp, rm } from 'node:fs/promises'
  4. import { tmpdir } from 'node:os'
  5. import { join } from 'node:path'
  6. import SessionStore, { SessionId } from '@deepseek-ai/dsh-session'
  7. import type { Session, SessionEvent, SurfaceEvent, SurfaceEventType } from '@deepseek-ai/dsh-session'
  8. import SessionPersistenceSqlite, { SCHEMA_VERSION } from '@deepseek-ai/dsh-session-persistence-sqlite'
  9. import { openDatabase, rowToEvent, scanRows, type EventRow } from '../src/schema.ts'
  10. import { runPersistenceContract, meta, oneTurnLog, appendLog } from '../../session-persistence/tests/contract.ts'
  11. import { runCoordinatorContract, type CoordinatorFixture } from '../../session-persistence/tests/coordinator-contract.ts'
  12. const dirs: string[] = []
  13. afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) })
  14. async function freshDbPath(): Promise<string> {
  15. const dir = await mkdtemp(join(tmpdir(), 'dsh-sqlite-'))
  16. dirs.push(dir)
  17. return join(dir, 'sessions.db')
  18. }
  19. /** A context with the session store + SQLite backend, plus a teardown. */
  20. async function backend(path = ':memory:'): Promise<{ ctx: Context; dispose: () => Promise<void> }> {
  21. const ctx = new Context()
  22. await ctx.plugin(SessionStore)
  23. const fiber = await ctx.plugin(SessionPersistenceSqlite, { path })
  24. return { ctx, dispose: () => fiber.dispose() }
  25. }
  26. // The payoff: the SAME backend-agnostic contract the JSONL backend runs, now
  27. // proving the SQLite backend satisfies identical semantics.
  28. runPersistenceContract('sqlite', async () => {
  29. const ctx = new Context()
  30. await ctx.plugin(SessionStore)
  31. const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' })
  32. return {
  33. persistence: ctx.sessionPersistence,
  34. dispose: async () => { await fiber.dispose() },
  35. }
  36. })
  37. // Run the shared coordinator orchestration suite against the real SQLite backend.
  38. // A FILE-backed db (not :memory:) is the shared storage scope so two mounted
  39. // instances see the same rows (HMR/reload). `corruptTail` INSERTs a row past the
  40. // committed seq whose `data` is invalid JSON — a never-committed torn tail that
  41. // drives the coordinator's commitRepair-with-tornMarker branch over real db rows.
  42. runCoordinatorContract('sqlite', async (): Promise<CoordinatorFixture> => {
  43. const dir = await mkdtemp(join(tmpdir(), 'dsh-sqlite-coord-'))
  44. const path = join(dir, 'sessions.db')
  45. return {
  46. mount: async ctx => ctx.plugin(SessionPersistenceSqlite, { path }),
  47. corruptTail: async (id) => {
  48. // A row past the committed region whose `data` does not parse: scanRows
  49. // bounds the preserved prefix at it and returns its seq as tornFrom, which
  50. // the backend surfaces to the coordinator as the tornMarker to delete from.
  51. const db = openDatabase(path)
  52. const next = (db.prepare('SELECT COALESCE(MAX(seq), -1) + 1 AS n FROM events WHERE session_id = ?')
  53. .get(id) as { n: number }).n
  54. db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, ?, ?, ?, ?)')
  55. .run(id, next, 'assistant/chunk', 99, '{not valid json')
  56. db.close()
  57. },
  58. cleanup: async () => { await rm(dir, { recursive: true, force: true }) },
  59. }
  60. })
  61. describe('scanRows', () => {
  62. // scanRows works off EventRows (data is a JSON string column); build them from
  63. // SessionEvents so the unit tests read in terms of the event vocabulary. Surface
  64. // fields are serialized to their nullable columns so a round trip is faithful.
  65. const rows = (events: SessionEvent[]): EventRow[] =>
  66. events.map((e) => {
  67. const se = e as SessionEvent<SurfaceEventType>
  68. return {
  69. seq: e.seq, type: e.type, time: e.time, data: JSON.stringify(e.data),
  70. source_event_seqs: se.sourceEventSeqs !== undefined ? JSON.stringify(se.sourceEventSeqs) : null,
  71. surface_op: se.surfaceOp !== undefined ? JSON.stringify(se.surfaceOp) : null,
  72. }
  73. })
  74. it('preserves the full log when it ends exactly on a turn/end (no torn tail)', () => {
  75. const { preserved, tornFrom } = scanRows(rows(oneTurnLog()))
  76. expect(preserved).toEqual(oneTurnLog())
  77. expect(tornFrom).toBeUndefined()
  78. })
  79. it('PRESERVES the real events of an interrupted turn after the last turn/end', () => {
  80. // turn 1 committed (0..5) + a crashed turn 2 (turn/start 6, step/start 7, no
  81. // close): all 8 rows are intact, so the whole prefix is preserved and there
  82. // is no torn fragment to delete. (load() then synthesizes the closers.)
  83. const withOpenTurn: SessionEvent[] = [
  84. ...oneTurnLog(),
  85. { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
  86. { type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
  87. ]
  88. const { preserved, tornFrom } = scanRows(rows(withOpenTurn))
  89. expect(preserved.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
  90. expect(tornFrom).toBeUndefined()
  91. })
  92. it('preserves the contiguous prefix and flags a torn tail at a seq gap', () => {
  93. // A gap after seq 0 (no committed turn/end): seq 0 is the preserved
  94. // interrupted-turn event; the gap bounds it and marks the torn fragment.
  95. const gapped: SessionEvent[] = [
  96. { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
  97. { type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // seq 1 missing
  98. ]
  99. const { preserved, tornFrom } = scanRows(rows(gapped))
  100. expect(preserved.map(e => e.seq)).toEqual([0])
  101. expect(tornFrom).toBe(1)
  102. })
  103. it('an empty log preserves nothing and has no torn tail', () => {
  104. expect(scanRows([])).toEqual({ preserved: [] })
  105. })
  106. it('throws on a seq gap inside the committed region (before the last turn/end)', () => {
  107. const gapped: SessionEvent[] = [
  108. { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
  109. { type: 'step/start', seq: 2, time: 2, data: { turn: 1, step: 1 } }, // seq 1 missing
  110. { type: 'turn/end', seq: 3, time: 3, data: { turn: 1, reason: { kind: 'completed' } } },
  111. ]
  112. expect(() => scanRows(rows(gapped))).toThrow(/seq gap in committed region/)
  113. })
  114. it('throws on an unparsable row inside the committed region', () => {
  115. const withCorruptCommitted: EventRow[] = [
  116. { seq: 0, type: 'turn/start', time: 1, data: '{not json', source_event_seqs: null, surface_op: null }, // corrupt, sits before a turn/end
  117. { seq: 1, type: 'turn/end', time: 2, data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }), source_event_seqs: null, surface_op: null },
  118. ]
  119. expect(() => scanRows(withCorruptCommitted)).toThrow(/unparsable committed event/)
  120. })
  121. it('tolerates an unparsable torn-tail row after the last turn/end', () => {
  122. const withCorruptTail: EventRow[] = [
  123. ...rows(oneTurnLog()),
  124. { seq: 6, type: 'turn/start', time: 7, data: '{not json', source_event_seqs: null, surface_op: null }, // torn fragment, no committed turn/end after
  125. ]
  126. const { preserved, tornFrom } = scanRows(withCorruptTail)
  127. expect(preserved).toEqual(oneTurnLog())
  128. expect(tornFrom).toBe(6)
  129. })
  130. })
  131. describe('SessionPersistenceSqlite: durability and crash semantics', () => {
  132. it('an interrupted turn (rows after the last turn/end) is PRESERVED and closed during load', async () => {
  133. const path = await freshDbPath()
  134. const m = meta('crash')
  135. // Run 1: persist a complete turn, then a half-written second turn (no turn/end).
  136. const ctx1 = new Context()
  137. await ctx1.plugin(SessionStore)
  138. const fiber1 = await ctx1.plugin(SessionPersistenceSqlite, { path })
  139. await ctx1.sessionPersistence.create(m)
  140. await ctx1.sessionPersistence.append(m.id, oneTurnLog())
  141. await ctx1.sessionPersistence.append(m.id, [
  142. { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
  143. { type: 'step/start', seq: 7, time: 8, data: { turn: 2, step: 1 } },
  144. ])
  145. await fiber1.dispose()
  146. // Run 2: load PRESERVES the interrupted turn's real events (a turn can be huge
  147. // — never truncated) and closes the orphaned turn with synthetic boundary
  148. // events: step/end (the step was open) then turn/end {interrupted}.
  149. const ctx2 = new Context()
  150. await ctx2.plugin(SessionStore)
  151. const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path })
  152. const loaded = await ctx2.sessionPersistence.load(m.id)
  153. expect(loaded.events.map(e => e.type)).toEqual([
  154. 'turn/start', 'user/message', 'step/start', 'assistant/message', 'step/end', 'turn/end', // turn 1
  155. 'turn/start', 'step/start', 'step/end', 'turn/end', // turn 2: real events + synthetic closers
  156. ])
  157. expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9])
  158. const last = loaded.events.at(-1)!
  159. expect(last.type === 'turn/end' && last.data.reason).toEqual({ kind: 'interrupted' })
  160. // load durably closed the turn, so the next append continues at the balanced
  161. // length (seq 10) and a reload round-trips identically.
  162. await ctx2.sessionPersistence.append(m.id, [
  163. { type: 'turn/start', seq: 10, time: 9, data: { turn: 3, trigger: { kind: 'message', source: { kind: 'user' } } } },
  164. { type: 'turn/end', seq: 11, time: 10, data: { turn: 3, reason: { kind: 'completed' } } },
  165. ])
  166. const reloaded = await ctx2.sessionPersistence.load(m.id)
  167. expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 10, 11])
  168. await fiber2.dispose()
  169. })
  170. it('load() durably closes the interrupted turn: the synthetic closers are on disk after load', async () => {
  171. const path = await freshDbPath()
  172. const m = meta('load-closes')
  173. const b1 = await backend(path)
  174. await b1.ctx.sessionPersistence.create(m)
  175. await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5
  176. await b1.dispose()
  177. // Hand-write an interrupted turn (turn/start seq 6, no turn/end).
  178. const db = openDatabase(path)
  179. db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, 6, ?, 7, ?)')
  180. .run(m.id, 'turn/start', JSON.stringify({ turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } }))
  181. db.close()
  182. const b2 = await backend(path)
  183. const loaded = await b2.ctx.sessionPersistence.load(m.id)
  184. // turn 2's real turn/start (seq 6) is preserved + a synthetic turn/end (seq 7).
  185. expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
  186. expect(loaded.events.at(-1)!.type).toBe('turn/end')
  187. // load() is mutating: the synthetic turn/end MUST be on disk so the stored log
  188. // is balanced and the cursor is truthful (contract: load closes, not defers).
  189. const probe = openDatabase(path)
  190. const stored = probe.prepare('SELECT seq, type FROM events WHERE session_id = ? ORDER BY seq').all(m.id) as { seq: number; type: string }[]
  191. probe.close()
  192. expect(stored.map(r => r.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
  193. expect(stored.at(-1)!.type).toBe('turn/end')
  194. await b2.dispose()
  195. })
  196. it('all-tail load: a session whose only turn never closed is preserved and closed on load', async () => {
  197. const path = await freshDbPath()
  198. const m = meta('all-tail')
  199. const b1 = await backend(path)
  200. await b1.ctx.sessionPersistence.create(m)
  201. // A first turn that NEVER completed: turn/start + user/message, no turn/end.
  202. await b1.ctx.sessionPersistence.append(m.id, [
  203. { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
  204. { type: 'user/message', seq: 1, time: 2, data: { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } } },
  205. ])
  206. await b1.dispose()
  207. // A fresh backend loads it: the interrupted (only) turn's real events are
  208. // preserved and closed with a synthetic turn/end {interrupted} — NOT
  209. // truncated. The session was materialized, so list() reports it present.
  210. const b2 = await backend(path)
  211. const loaded = await b2.ctx.sessionPersistence.load(m.id)
  212. expect(loaded.events.map(e => e.type)).toEqual(['turn/start', 'user/message', 'turn/end'])
  213. expect(loaded.events.at(-1)!.type === 'turn/end' && loaded.events.at(-1)!.data).toMatchObject({ reason: { kind: 'interrupted' } })
  214. expect((await b2.ctx.sessionPersistence.list()).map(x => x.id)).toContain(m.id)
  215. await b2.dispose()
  216. })
  217. it('rejects opening a database whose schema version is not the current build (newer OR older)', async () => {
  218. const path = await freshDbPath()
  219. openDatabase(path).close() // stamp user_version = SCHEMA_VERSION
  220. // Bump user_version past what this build supports.
  221. const dbNewer = openDatabase(path)
  222. dbNewer.exec(`PRAGMA user_version = ${SCHEMA_VERSION + 1}`)
  223. dbNewer.close()
  224. expect(() => openDatabase(path)).toThrow(/incompatible with this build/)
  225. // A stale OLDER version (e.g. a pre-summary-drop v1 DB) is also rejected —
  226. // we do not migrate (unreleased software, no backward-compat).
  227. const olderPath = await freshDbPath()
  228. openDatabase(olderPath).close()
  229. const dbOlder = openDatabase(olderPath)
  230. dbOlder.exec('PRAGMA user_version = 1')
  231. dbOlder.close()
  232. expect(() => openDatabase(olderPath)).toThrow(/incompatible with this build/)
  233. })
  234. it('rejects a sibling v3 database (the merge-collided version) rather than opening it against missing columns', async () => {
  235. // Two unmerged branches each shipped a DISTINCT layout under user_version 3
  236. // (one added only `seed_length`, the other only the surface columns). The
  237. // merged build is v4; an on-disk v3 is ambiguous and is missing at least one
  238. // of this build's columns, so it MUST be rejected, not opened. Stamp a v3
  239. // database and confirm the version check refuses it.
  240. const path = await freshDbPath()
  241. openDatabase(path).close() // creates + stamps user_version = SCHEMA_VERSION (4)
  242. const db = openDatabase(path)
  243. db.exec('PRAGMA user_version = 3')
  244. db.close()
  245. expect(() => openDatabase(path)).toThrow(/schema version 3, incompatible with this build/)
  246. })
  247. it('a corrupt-JSON row in the uncommitted tail is discarded on load, not unloadable', async () => {
  248. const path = await freshDbPath()
  249. const m = meta('corrupt-tail')
  250. const b1 = await backend(path)
  251. await b1.ctx.sessionPersistence.create(m)
  252. await b1.ctx.sessionPersistence.append(m.id, oneTurnLog()) // committed: seqs 0..5
  253. await b1.dispose()
  254. // Hand-insert a torn tail row (seq 6, no closing turn/end) whose `data` is
  255. // invalid JSON. The contract: only a parse error in the COMMITTED region is
  256. // unloadable; a torn tail must be discarded. scanRows finds the last
  257. // turn/end on the seq+type columns (never parsing tail `data`), so the
  258. // unparsable row after it bounds the preserved prefix and is deleted by load.
  259. const db = openDatabase(path)
  260. db.prepare('INSERT INTO events (session_id, seq, type, time, data) VALUES (?, 6, ?, 7, ?)')
  261. .run(m.id, 'turn/start', '{not valid json')
  262. db.close()
  263. const b2 = await backend(path)
  264. const loaded = await b2.ctx.sessionPersistence.load(m.id)
  265. expect(loaded.events).toEqual(oneTurnLog()) // torn tail discarded, committed intact (turn 1 already balanced → no closers)
  266. // load physically deleted the corrupt tail row, so a fresh append continues.
  267. await b2.ctx.sessionPersistence.append(m.id, [
  268. { type: 'turn/start', seq: 6, time: 8, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
  269. { type: 'turn/end', seq: 7, time: 9, data: { turn: 2, reason: { kind: 'completed' } } },
  270. ])
  271. const reloaded = await b2.ctx.sessionPersistence.load(m.id)
  272. expect(reloaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
  273. await b2.dispose()
  274. })
  275. it('append rolls back the whole batch on a mid-batch seq collision (transaction)', async () => {
  276. const ctx = new Context()
  277. await ctx.plugin(SessionStore)
  278. const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' })
  279. const m = meta('rollback')
  280. await ctx.sessionPersistence.create(m)
  281. await ctx.sessionPersistence.append(m.id, oneTurnLog()) // seqs 0..5
  282. // A batch that re-states an already-stored seq must be rejected and leave
  283. // the stored log unchanged (the UNIQUE (session_id, seq) constraint fires
  284. // inside the transaction → ROLLBACK).
  285. await expect(ctx.sessionPersistence.append(m.id, oneTurnLog())).rejects.toThrow()
  286. const loaded = await ctx.sessionPersistence.load(m.id)
  287. expect(loaded.events).toEqual(oneTurnLog()) // unchanged
  288. await fiber.dispose()
  289. })
  290. it('persists across separate backend instances over the same file', async () => {
  291. const path = await freshDbPath()
  292. const m = meta('persist', '/proj')
  293. const ctx1 = new Context()
  294. await ctx1.plugin(SessionStore)
  295. const fiber1 = await ctx1.plugin(SessionPersistenceSqlite, { path })
  296. await ctx1.sessionPersistence.create(m)
  297. await ctx1.sessionPersistence.append(m.id, oneTurnLog())
  298. await fiber1.dispose()
  299. const ctx2 = new Context()
  300. await ctx2.plugin(SessionStore)
  301. const fiber2 = await ctx2.plugin(SessionPersistenceSqlite, { path })
  302. expect((await ctx2.sessionPersistence.list()).map(x => x.id)).toContain(m.id)
  303. const loaded = await ctx2.sessionPersistence.load(m.id)
  304. expect(loaded.meta).toMatchObject({ id: m.id, cwd: '/proj' })
  305. expect(loaded.events).toEqual(oneTurnLog())
  306. await fiber2.dispose()
  307. })
  308. it('exposes the schema version constant', () => {
  309. expect(SCHEMA_VERSION).toBe(4)
  310. })
  311. })
  312. describe('SessionPersistenceSqlite: edge cases', () => {
  313. it('append rolls back and rethrows when an event INSERT fails inside the transaction', async () => {
  314. const path = await freshDbPath()
  315. const m = meta('rollback-insert')
  316. const b1 = await backend(path)
  317. await b1.ctx.sessionPersistence.create(m)
  318. await b1.ctx.sessionPersistence.append(m.id, oneTurnLog())
  319. // A SECOND backend over the same file loads the session first, so it adopts
  320. // cursor 6 (the committed length) into its OWN in-memory state.
  321. const b2 = await backend(path)
  322. await b2.ctx.sessionPersistence.load(m.id) // cursor 6 in b2
  323. const turn2: SessionEvent[] = [
  324. { type: 'turn/start', seq: 6, time: 7, data: { turn: 2, trigger: { kind: 'message', source: { kind: 'user' } } } },
  325. { type: 'turn/end', seq: 7, time: 8, data: { turn: 2, reason: { kind: 'completed' } } },
  326. ]
  327. // b1 commits seq 6..7 first.
  328. await b1.ctx.sessionPersistence.append(m.id, turn2)
  329. // b2 still thinks its cursor is 6, so this batch passes the contiguity check
  330. // but its INSERT of seq 6 hits the UNIQUE (session_id, seq) constraint
  331. // mid-transaction → ROLLBACK + rethrow.
  332. await expect(b2.ctx.sessionPersistence.append(m.id, turn2)).rejects.toThrow(/UNIQUE/)
  333. // b1's turn is intact; b2's rolled-back attempt left nothing extra.
  334. const loaded = await b1.ctx.sessionPersistence.load(m.id)
  335. expect(loaded.events.map(e => e.seq)).toEqual([0, 1, 2, 3, 4, 5, 6, 7])
  336. await b1.dispose()
  337. await b2.dispose()
  338. })
  339. it('HMR: a DIFFERENT session colliding with a materialized on-disk id is rejected', async () => {
  340. const path = await freshDbPath()
  341. // Instance 1 materializes a session and disposes.
  342. const b1 = await backend(path)
  343. const s1 = b1.ctx.sessions.create(SessionId('hmr-collide'))
  344. appendLog(s1, oneTurnLog())
  345. await b1.ctx.parallel('session/flush', s1)
  346. await b1.dispose()
  347. // A fresh context with an UNRELATED live session reusing the id meets a
  348. // materialized row that is NOT a prefix of its events → reject.
  349. const ctx = new Context()
  350. await ctx.plugin(SessionStore)
  351. let session!: Session
  352. await ctx.plugin(Object.assign((inner: Context) => {
  353. session = inner.sessions.create(SessionId('hmr-collide'))
  354. }, { inject: ['sessions'] }))
  355. session.append('turn/start', { turn: 9, trigger: { kind: 'message', source: { kind: 'user' } } })
  356. await ctx.plugin(SessionPersistenceSqlite, { path })
  357. await expect(ctx.parallel('session/flush', session)).rejects.toThrow(/id collision/)
  358. await ctx.fiber.dispose()
  359. })
  360. })
  361. describe('surface field round-trip', () => {
  362. it('rowToEvent parses surface fields from EventRow columns', () => {
  363. const row: EventRow = {
  364. seq: 0, type: 'assistant/message', time: 1,
  365. data: JSON.stringify({ turn: 1, step: 1, content: [] }),
  366. source_event_seqs: JSON.stringify([3, 5]),
  367. surface_op: JSON.stringify('append'),
  368. }
  369. const event = rowToEvent(row)
  370. expect((event as SurfaceEvent).sourceEventSeqs).toEqual([3, 5])
  371. expect((event as SurfaceEvent).surfaceOp).toBe('append')
  372. })
  373. it('rowToEvent handles replace surfaceOp object', () => {
  374. const row: EventRow = {
  375. seq: 0, type: 'assistant/message', time: 1,
  376. data: JSON.stringify({ turn: 1, step: 1, content: [] }),
  377. source_event_seqs: JSON.stringify([0, 1]),
  378. surface_op: JSON.stringify({ op: 'replace', start: 0, end: 1 }),
  379. }
  380. const event = rowToEvent(row)
  381. expect((event as SurfaceEvent).sourceEventSeqs).toEqual([0, 1])
  382. expect((event as SurfaceEvent).surfaceOp).toEqual({ op: 'replace', start: 0, end: 1 })
  383. })
  384. it('scanRows with surface columns reconstructs events with surface fields', () => {
  385. const rows: EventRow[] = [
  386. { seq: 0, type: 'user/message', time: 1,
  387. data: JSON.stringify({ content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }),
  388. source_event_seqs: null, surface_op: '{"op":"replace","start":0,"end":0}' },
  389. { seq: 1, type: 'turn/end', time: 2,
  390. data: JSON.stringify({ turn: 1, reason: { kind: 'completed' } }),
  391. source_event_seqs: null, surface_op: null },
  392. ]
  393. const { preserved } = scanRows(rows)
  394. expect(preserved).toHaveLength(2)
  395. expect((preserved[0]! as SurfaceEvent).surfaceOp).toEqual({ op: 'replace', start: 0, end: 0 })
  396. expect((preserved[0]! as SurfaceEvent).sourceEventSeqs).toBeUndefined()
  397. expect((preserved[1] as SessionEvent<SurfaceEventType>).surfaceOp).toBeUndefined()
  398. })
  399. it('append and load round-trips surface fields through SQLite', async () => {
  400. const ctx = new Context()
  401. await ctx.plugin(SessionStore)
  402. const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' })
  403. const session = ctx.sessions.create(SessionId('roundtrip-surface'))
  404. session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
  405. session.append('user/message', { content: [{ type: 'text', text: 'hi' }], source: { kind: 'user' } }, { surfaceOp: 'append' })
  406. session.append('assistant/message', { turn: 1, step: 1, content: [] }, { surfaceOp: 'append', sourceEventSeqs: [0] })
  407. session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
  408. await ctx.parallel('session/flush', session)
  409. const loaded = await ctx.sessionPersistence.load(SessionId('roundtrip-surface'))
  410. expect(loaded.events).toHaveLength(4)
  411. const um = loaded.events[1]!
  412. expect((um as SurfaceEvent).surfaceOp).toBe('append')
  413. expect((um as SurfaceEvent).sourceEventSeqs).toBeUndefined()
  414. const am = loaded.events[2]!
  415. expect((am as SurfaceEvent).surfaceOp).toBe('append')
  416. expect((am as SurfaceEvent).sourceEventSeqs).toEqual([0])
  417. await fiber.dispose()
  418. })
  419. it('persists events with surfaceOp but no sourceEventSeqs (covers null branch in surfaceBindings)', async () => {
  420. const ctx = new Context()
  421. await ctx.plugin(SessionStore)
  422. const fiber = await ctx.plugin(SessionPersistenceSqlite, { path: ':memory:' })
  423. const session = ctx.sessions.create(SessionId('surface-noseq'))
  424. session.append('turn/start', { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } })
  425. session.append('steering/message', { turn: 1, content: [], source: { kind: 'user' } }, { surfaceOp: 'append' })
  426. session.append('turn/end', { turn: 1, reason: { kind: 'completed' } })
  427. await ctx.parallel('session/flush', session)
  428. const loaded = await ctx.sessionPersistence.load(SessionId('surface-noseq'))
  429. expect((loaded.events[1]! as SurfaceEvent).surfaceOp).toBe('append')
  430. expect((loaded.events[1]! as SurfaceEvent).sourceEventSeqs).toBeUndefined()
  431. await fiber.dispose()
  432. })
  433. })