resume.spec.ts 26 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570
  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 LlmService from '@deepseek-ai/dsh-llm'
  7. import SessionStore, { Session, SessionId } from '@deepseek-ai/dsh-session'
  8. import type { SessionEvent } from '@deepseek-ai/dsh-session'
  9. import SystemPrompt from '@deepseek-ai/dsh-system-prompt'
  10. import ToolRegistry from '@deepseek-ai/dsh-tools'
  11. import AgentRegistry, { AgentId } from '@deepseek-ai/dsh-agent'
  12. import SessionPersistenceJsonl from '@deepseek-ai/dsh-session-persistence-jsonl'
  13. import AgentLoop, { ReactLoopAgent } from '@deepseek-ai/dsh-agent-loop'
  14. import { MockAdapter, textResponse } from './mock-adapter.ts'
  15. const dirs: string[] = []
  16. afterEach(async () => { for (const d of dirs.splice(0)) await rm(d, { recursive: true, force: true }) })
  17. async function persistentHarness(adapter: MockAdapter): Promise<{ ctx: Context; root: string }> {
  18. const root = await mkdtemp(join(tmpdir(), 'dsh-resume-'))
  19. dirs.push(root)
  20. return { ctx: await mountPersistentHarness(root, adapter), root }
  21. }
  22. async function mountPersistentHarness(root: string, adapter: MockAdapter): Promise<Context> {
  23. const ctx = new Context()
  24. await ctx.plugin(LlmService)
  25. await ctx.plugin(SessionStore)
  26. await ctx.plugin(SystemPrompt)
  27. await ctx.plugin(ToolRegistry)
  28. await ctx.plugin(AgentRegistry)
  29. await ctx.plugin(AgentLoop, { agents: [] })
  30. await ctx.plugin(SessionPersistenceJsonl, { root })
  31. ctx.llm.registerAdapter(['mock'], adapter)
  32. return ctx
  33. }
  34. async function persistSession(sessionId: SessionId): Promise<string> {
  35. const { ctx, root } = await persistentHarness(new MockAdapter([textResponse('seed')]))
  36. // Persistence deliberately has no artifact for a truly empty session. A
  37. // balanced completed turn is the smallest resumable log and avoids running
  38. // the model merely to construct this lifecycle fixture.
  39. const seed: SessionEvent[] = [
  40. { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
  41. { type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } },
  42. ]
  43. const session = ctx.sessions.create(sessionId, { seed })
  44. await ctx.sessions.flush(session)
  45. await ctx.fiber.dispose()
  46. return root
  47. }
  48. function waitForIdle(ctx: Context, agent: ReactLoopAgent): Promise<void> {
  49. return new Promise((resolve) => {
  50. const dispose = ctx.on('agent/status', (subject, status) => {
  51. if (subject === agent && status === 'idle') { dispose(); resolve() }
  52. })
  53. })
  54. }
  55. /** Fail a lifecycle regression promptly instead of waiting for Vitest's suite timeout. */
  56. async function promptly<T>(task: Promise<T>): Promise<T> {
  57. const timeout = Promise.withResolvers<never>()
  58. const timer = setTimeout(() => { timeout.reject(new Error('lifecycle task did not settle promptly')) }, 1000)
  59. try {
  60. return await Promise.race([task, timeout.promise])
  61. } finally {
  62. clearTimeout(timer)
  63. }
  64. }
  65. /** Throw an arbitrary callback value to exercise the public unknown-error boundary. */
  66. function throwUnknown(value: unknown): never {
  67. throw value
  68. }
  69. describe('the session-persistence RFC: AgentLoop factory create/resume', () => {
  70. it('normalizes a non-Error resume publication failure for rollback and rethrows it', async () => {
  71. const sessionId = SessionId('unknown-resume-failure-s')
  72. const root = await persistSession(sessionId)
  73. const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
  74. const failure = { source: 'resume' }
  75. ctx.on('session/created', () => throwUnknown(failure))
  76. await expect(ctx.agents.resume({
  77. agentId: AgentId('unknown-resume-failure'),
  78. resumeSessionId: sessionId,
  79. })).rejects.toBe(failure)
  80. expect(ctx.agents.get(AgentId('unknown-resume-failure'))).toBeUndefined()
  81. expect(ctx.sessions.get(sessionId)).toBeUndefined()
  82. await ctx.fiber.dispose()
  83. })
  84. it('createAgent uses the caller-supplied sessionId (not ${id}-session)', async () => {
  85. const adapter = new MockAdapter([textResponse('hi')])
  86. const { ctx } = await persistentHarness(adapter)
  87. const { agent } = await ctx.agents.create({ agentId: AgentId('a1'), sessionId: SessionId('custom-session'), meta: { cwd: '/w' } })
  88. expect(agent.session.id).toBe('custom-session')
  89. expect(agent.session.header.cwd).toBe('/w')
  90. await ctx.fiber.dispose()
  91. })
  92. it('createAgent rejects a duplicate agent id BEFORE creating the session (no orphan)', async () => {
  93. const adapter = new MockAdapter([textResponse('hi')])
  94. const { ctx } = await persistentHarness(adapter)
  95. await ctx.agents.create({ agentId: AgentId('dup'), sessionId: SessionId('sess-a') })
  96. // A second create with the SAME agent id but a fresh session id must reject
  97. // up front — and must NOT leave an orphaned 'sess-b' session behind.
  98. await expect(ctx.agents.create({ agentId: AgentId('dup'), sessionId: SessionId('sess-b') })).rejects.toThrow(/already registered/)
  99. expect(ctx.sessions.get(SessionId('sess-b'))).toBeUndefined()
  100. await ctx.fiber.dispose()
  101. })
  102. it('createAgent works without meta (no cwd)', async () => {
  103. const adapter = new MockAdapter([textResponse('hi')])
  104. const { ctx } = await persistentHarness(adapter)
  105. const { agent } = await ctx.agents.create({ agentId: AgentId('a-nometa'), sessionId: SessionId('nometa-session') })
  106. expect(agent.session.id).toBe('nometa-session')
  107. expect(agent.session.header.cwd).toBeUndefined()
  108. await ctx.fiber.dispose()
  109. })
  110. it('resume of a session with no cwd carries an undefined cwd header', async () => {
  111. // Lifecycle 1: create a no-cwd session and run a turn.
  112. const adapter1 = new MockAdapter([textResponse('a')])
  113. const { ctx: ctx1, root } = await persistentHarness(adapter1)
  114. const a1 = (await ctx1.agents.create({ agentId: AgentId('m'), sessionId: SessionId('nocwd-sess') })).agent as ReactLoopAgent
  115. a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
  116. await waitForIdle(ctx1, a1)
  117. await ctx1.fiber.dispose()
  118. // Lifecycle 2: resume it; the header cwd stays undefined (no-cwd branch).
  119. const adapter2 = new MockAdapter([textResponse('b')])
  120. const ctx2 = new Context()
  121. await ctx2.plugin(LlmService)
  122. await ctx2.plugin(SessionStore)
  123. await ctx2.plugin(SystemPrompt)
  124. await ctx2.plugin(ToolRegistry)
  125. await ctx2.plugin(AgentRegistry)
  126. await ctx2.plugin(AgentLoop, { agents: [] })
  127. await ctx2.plugin(SessionPersistenceJsonl, { root })
  128. ctx2.llm.registerAdapter(['mock'], adapter2)
  129. const a2 = (await ctx2.agents.resume({ agentId: AgentId('m'), resumeSessionId: SessionId('nocwd-sess') })).agent as ReactLoopAgent
  130. expect(a2.session.header.cwd).toBeUndefined()
  131. await ctx2.fiber.dispose()
  132. })
  133. it('agent/session-start fires "startup" for createAgent and "resume" for resume()', async () => {
  134. // Lifecycle 1: a fresh createAgent emits session-start with source 'startup'.
  135. const adapter1 = new MockAdapter([textResponse('a')])
  136. const { ctx: ctx1, root } = await persistentHarness(adapter1)
  137. const sources1: string[] = []
  138. ctx1.on('agent/session-start', (_agent, source) => void sources1.push(source))
  139. const a1 = (await ctx1.agents.create({ agentId: AgentId('s'), sessionId: SessionId('start-sess') })).agent as ReactLoopAgent
  140. expect(sources1).toEqual(['startup'])
  141. a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
  142. await waitForIdle(ctx1, a1)
  143. await ctx1.fiber.dispose()
  144. // Lifecycle 2: resuming the persisted session emits session-start 'resume'.
  145. const adapter2 = new MockAdapter([textResponse('b')])
  146. const ctx2 = new Context()
  147. await ctx2.plugin(LlmService)
  148. await ctx2.plugin(SessionStore)
  149. await ctx2.plugin(SystemPrompt)
  150. await ctx2.plugin(ToolRegistry)
  151. await ctx2.plugin(AgentRegistry)
  152. await ctx2.plugin(AgentLoop, { agents: [] })
  153. await ctx2.plugin(SessionPersistenceJsonl, { root })
  154. ctx2.llm.registerAdapter(['mock'], adapter2)
  155. const sources2: string[] = []
  156. ctx2.on('agent/session-start', (_agent, source) => void sources2.push(source))
  157. await ctx2.agents.resume({ agentId: AgentId('s'), resumeSessionId: SessionId('start-sess') })
  158. expect(sources2).toEqual(['resume'])
  159. await ctx2.fiber.dispose()
  160. })
  161. it('resume awaits setup while unpublished, then publishes a fully composed world in order', async () => {
  162. const sessionId = SessionId('resume-setup-success')
  163. const root = await persistSession(sessionId)
  164. const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
  165. const gate = Promise.withResolvers<undefined>()
  166. const setupStarted = Promise.withResolvers<undefined>()
  167. const order: string[] = []
  168. ctx.on('session/created', (session) => {
  169. expect(ctx.sessions.get(session.id)).toBe(session)
  170. expect(ctx.agents.get(AgentId('resumed-atomic'))?.session).toBe(session)
  171. order.push('session/created')
  172. })
  173. ctx.on('agent/created', (agent) => {
  174. expect(agent.status).toBe('idle')
  175. order.push('agent/created')
  176. })
  177. ctx.on('agent/session-start', (agent) => {
  178. expect(() => { agent.cancel('now live') }).not.toThrow()
  179. order.push('agent/session-start')
  180. })
  181. const resuming = ctx.agents.resume({
  182. agentId: AgentId('resumed-atomic'),
  183. resumeSessionId: sessionId,
  184. agentOptions: { model: 'mock' },
  185. setup: async (agentCtx) => {
  186. expect(agentCtx.agent?.id).toBe(AgentId('resumed-atomic'))
  187. expect(agentCtx.agent?.session.events).toHaveLength(2)
  188. agentCtx.on('session/created', () => void order.push('setup-listener:session/created'))
  189. agentCtx.on('agent/created', () => void order.push('setup-listener:agent/created'))
  190. order.push('setup:start')
  191. setupStarted.resolve(undefined)
  192. await gate.promise
  193. order.push('setup:end')
  194. },
  195. })
  196. await setupStarted.promise
  197. expect(ctx.agents.get(AgentId('resumed-atomic'))).toBeUndefined()
  198. expect(ctx.sessions.get(sessionId)).toBeUndefined()
  199. expect(order).toEqual(['setup:start'])
  200. gate.resolve(undefined)
  201. const handle = await resuming
  202. expect(order).toEqual([
  203. 'setup:start',
  204. 'setup:end',
  205. 'session/created',
  206. 'setup-listener:session/created',
  207. 'agent/created',
  208. 'setup-listener:agent/created',
  209. 'agent/session-start',
  210. ])
  211. await handle.dispose()
  212. await ctx.fiber.dispose()
  213. })
  214. it('successful resume disposal retires its caller-owned transaction effects', async () => {
  215. const sessionId = SessionId('resume-retired-effects-s')
  216. const agentId = AgentId('resume-retired-effects')
  217. const root = await persistSession(sessionId)
  218. const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
  219. const handle = await ctx.agents.resume({
  220. agentId,
  221. resumeSessionId: sessionId,
  222. agentOptions: { model: 'mock' },
  223. })
  224. const transactionLabels = [
  225. `agentLoop.owner(${agentId})`,
  226. `agentLoop.lifecycle(${agentId})`,
  227. ]
  228. expect(ctx.fiber.getEffects().map(effect => effect.label)).toEqual(expect.arrayContaining(transactionLabels))
  229. await handle.dispose()
  230. expect(ctx.fiber.getEffects().filter(effect => transactionLabels.includes(effect.label))).toEqual([])
  231. await ctx.fiber.dispose()
  232. })
  233. it('resume setup rejection publishes nothing, unwinds, and releases both identities', async () => {
  234. const sessionId = SessionId('resume-setup-reject')
  235. const root = await persistSession(sessionId)
  236. const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
  237. const published: string[] = []
  238. ctx.on('session/created', () => void published.push('session/created'))
  239. ctx.on('agent/created', () => void published.push('agent/created'))
  240. ctx.on('agent/session-start', () => void published.push('agent/session-start'))
  241. await expect(ctx.agents.resume({
  242. agentId: AgentId('resume-reject'),
  243. resumeSessionId: sessionId,
  244. agentOptions: { model: 'mock' },
  245. setup: async () => {
  246. await Promise.resolve()
  247. throw new Error('resume setup failed')
  248. },
  249. })).rejects.toThrow('resume setup failed')
  250. expect(published).toEqual([])
  251. expect(ctx.agents.get(AgentId('resume-reject'))).toBeUndefined()
  252. expect(ctx.sessions.get(sessionId)).toBeUndefined()
  253. const retry = await ctx.agents.resume({
  254. agentId: AgentId('resume-reject'),
  255. resumeSessionId: sessionId,
  256. agentOptions: { model: 'mock' },
  257. })
  258. await retry.dispose()
  259. await ctx.fiber.dispose()
  260. })
  261. it('owner unload aborts resume setup and cannot publish after the callback settles', async () => {
  262. const sessionId = SessionId('resume-setup-owner-unload')
  263. const root = await persistSession(sessionId)
  264. const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
  265. const gate = Promise.withResolvers<undefined>()
  266. const setupStarted = Promise.withResolvers<undefined>()
  267. const published: string[] = []
  268. ctx.on('session/created', () => void published.push('session/created'))
  269. ctx.on('agent/created', () => void published.push('agent/created'))
  270. let resuming!: ReturnType<typeof ctx.agents.resume>
  271. const owner = await ctx.plugin(Object.assign((inner: Context) => {
  272. resuming = inner.agents.resume({
  273. agentId: AgentId('resume-owner-race'),
  274. resumeSessionId: sessionId,
  275. agentOptions: { model: 'mock' },
  276. setup: async () => {
  277. setupStarted.resolve(undefined)
  278. await gate.promise
  279. },
  280. })
  281. }, { inject: ['agents'] }))
  282. await setupStarted.promise
  283. await owner.dispose()
  284. await expect(resuming).rejects.toThrow(/owner disposed during setup/)
  285. expect(published).toEqual([])
  286. expect(ctx.agents.get(AgentId('resume-owner-race'))).toBeUndefined()
  287. expect(ctx.sessions.get(sessionId)).toBeUndefined()
  288. gate.resolve(undefined)
  289. await Promise.resolve()
  290. expect(published).toEqual([])
  291. await ctx.fiber.dispose()
  292. })
  293. it('owner unload aborts a never-settling persistence load, releases identities, and blocks late publication', async () => {
  294. const sessionId = SessionId('resume-load-owner-unload')
  295. const agentId = AgentId('resume-load-race')
  296. const root = await persistSession(sessionId)
  297. const ctx = await mountPersistentHarness(root, new MockAdapter([textResponse('next')]))
  298. const snapshot = await ctx.sessionPersistence.load(sessionId)
  299. const lateLoad = Promise.withResolvers<typeof snapshot>()
  300. const loadStarted = Promise.withResolvers<undefined>()
  301. let loads = 0
  302. ctx.sessionPersistence.load = (id) => {
  303. expect(id).toBe(sessionId)
  304. loads += 1
  305. if (loads === 1) {
  306. loadStarted.resolve(undefined)
  307. return lateLoad.promise
  308. }
  309. return Promise.resolve(structuredClone(snapshot))
  310. }
  311. const published: string[] = []
  312. ctx.on('session/created', () => void published.push('session/created'))
  313. ctx.on('agent/created', () => void published.push('agent/created'))
  314. ctx.on('agent/session-start', () => void published.push('agent/session-start'))
  315. let resuming!: ReturnType<typeof ctx.agents.resume>
  316. const owner = await ctx.plugin(Object.assign((inner: Context) => {
  317. resuming = inner.agents.resume({ agentId, resumeSessionId: sessionId, agentOptions: { model: 'mock' } })
  318. }, { inject: ['agents'] }))
  319. await loadStarted.promise
  320. const rejection = expect(promptly(resuming)).rejects.toThrow(/owner disposed during setup/)
  321. await promptly(owner.dispose())
  322. expect(published).toEqual([])
  323. expect(ctx.agents.get(agentId)).toBeUndefined()
  324. expect(ctx.sessions.get(sessionId)).toBeUndefined()
  325. // owner.dispose() awaited transaction settlement, so the same identities
  326. // can be reused before awaiting the public rejection.
  327. const retry = await promptly(ctx.agents.resume({ agentId, resumeSessionId: sessionId, agentOptions: { model: 'mock' } }))
  328. await rejection
  329. expect(loads).toBe(2)
  330. expect(published).toEqual(['session/created', 'agent/created', 'agent/session-start'])
  331. // Settlement of the abandoned backend promise cannot resume the old
  332. // transaction or emit a second publication after the retry owns the ids.
  333. lateLoad.resolve(structuredClone(snapshot))
  334. await Promise.resolve()
  335. await Promise.resolve()
  336. expect(ctx.agents.get(agentId)).toBe(retry.agent)
  337. expect(ctx.sessions.get(sessionId)).toBe(retry.agent.session)
  338. expect(published).toEqual(['session/created', 'agent/created', 'agent/session-start'])
  339. await retry.dispose()
  340. await ctx.fiber.dispose()
  341. })
  342. it('AgentLoop unload aborts persistence load and awaits wrapper settlement', async () => {
  343. const sessionId = SessionId('resume-load-factory-unload')
  344. const agentId = AgentId('resume-load-factory-race')
  345. const root = await persistSession(sessionId)
  346. const ctx = new Context()
  347. await ctx.plugin(LlmService)
  348. await ctx.plugin(SessionStore)
  349. await ctx.plugin(SystemPrompt)
  350. await ctx.plugin(ToolRegistry)
  351. await ctx.plugin(AgentRegistry)
  352. const loopFiber = await ctx.plugin(AgentLoop, { agents: [] })
  353. await ctx.plugin(SessionPersistenceJsonl, { root })
  354. ctx.llm.registerAdapter(['mock'], new MockAdapter([textResponse('next')]))
  355. const snapshot = await ctx.sessionPersistence.load(sessionId)
  356. const lateLoad = Promise.withResolvers<typeof snapshot>()
  357. const loadStarted = Promise.withResolvers<undefined>()
  358. ctx.sessionPersistence.load = (id) => {
  359. expect(id).toBe(sessionId)
  360. loadStarted.resolve(undefined)
  361. return lateLoad.promise
  362. }
  363. const published: string[] = []
  364. ctx.on('session/created', () => void published.push('session/created'))
  365. ctx.on('agent/created', () => void published.push('agent/created'))
  366. const resuming = ctx.agents.resume({ agentId, resumeSessionId: sessionId, agentOptions: { model: 'mock' } })
  367. await loadStarted.promise
  368. const rejection = expect(promptly(resuming)).rejects.toThrow(/agent loop is not active/)
  369. await promptly(loopFiber.dispose())
  370. await rejection
  371. expect(published).toEqual([])
  372. expect(ctx.agents.get(agentId)).toBeUndefined()
  373. expect(ctx.sessions.get(sessionId)).toBeUndefined()
  374. lateLoad.resolve(structuredClone(snapshot))
  375. await Promise.resolve()
  376. await Promise.resolve()
  377. expect(published).toEqual([])
  378. await ctx.fiber.dispose()
  379. })
  380. it('resume of a forked session preserves the parentSession lineage and seed boundary in the header', async () => {
  381. // Lifecycle 1: persist a FORKED session (carries parentSession + seedLength
  382. // in its header) by creating it with a complete-turn seed — the write path
  383. // materializes the fork (header + seed) on disk.
  384. const seed: SessionEvent[] = [
  385. { type: 'turn/start', seq: 0, time: 1, data: { turn: 1, trigger: { kind: 'message', source: { kind: 'user' } } } },
  386. { type: 'turn/end', seq: 1, time: 2, data: { turn: 1, reason: { kind: 'completed' } } },
  387. ]
  388. const adapter1 = new MockAdapter([textResponse('a')])
  389. const { ctx: ctx1, root } = await persistentHarness(adapter1)
  390. const forked = ctx1.sessions.create(SessionId('forked-sess'), {
  391. seed,
  392. meta: { cwd: '/w', parentSession: SessionId('parent-sess'), seedLength: seed.length },
  393. })
  394. await ctx1.parallel('session/flush', forked)
  395. await ctx1.fiber.dispose()
  396. // Lifecycle 2: resume it; the parentSession + seedLength header survives the
  397. // round-trip (exercises resume's parentSession- and seedLength-present
  398. // branches). seedLength must come from the PERSISTED header, not from the
  399. // resume seed length (which is the whole stored log, not the original
  400. // boundary).
  401. const adapter2 = new MockAdapter([textResponse('b')])
  402. const ctx2 = new Context()
  403. await ctx2.plugin(LlmService)
  404. await ctx2.plugin(SessionStore)
  405. await ctx2.plugin(SystemPrompt)
  406. await ctx2.plugin(ToolRegistry)
  407. await ctx2.plugin(AgentRegistry)
  408. await ctx2.plugin(AgentLoop, { agents: [] })
  409. await ctx2.plugin(SessionPersistenceJsonl, { root })
  410. ctx2.llm.registerAdapter(['mock'], adapter2)
  411. const a2 = (await ctx2.agents.resume({ agentId: AgentId('m'), resumeSessionId: SessionId('forked-sess') })).agent as ReactLoopAgent
  412. expect(a2.session.header.parentSession).toBe('parent-sess')
  413. expect(a2.session.header.cwd).toBe('/w')
  414. expect(a2.session.header.seedLength).toBe(seed.length)
  415. await ctx2.fiber.dispose()
  416. })
  417. it('an idle inject() is flushed durably on its own (survives without explicit flush/dispose)', async () => {
  418. // Idle injection creates and flushes a one-shot turn. No explicit flush or
  419. // clean disposal follows, so disk presence proves its own checkpoint ran.
  420. const adapter1 = new MockAdapter([textResponse('answer')])
  421. const { ctx: ctx1, root } = await persistentHarness(adapter1)
  422. const a1 = (await ctx1.agents.create({ agentId: AgentId('m'), sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } })).agent as ReactLoopAgent
  423. a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
  424. await waitForIdle(ctx1, a1)
  425. a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } })
  426. // Let inject()'s fire-and-forget flush settle (NO explicit flush/dispose).
  427. await new Promise(r => setTimeout(r, 30))
  428. // A SEPARATE backend reads the on-disk log — proving the inject persisted
  429. // itself, not a later dispose drain.
  430. const probe = new Context()
  431. await probe.plugin(SessionStore)
  432. await probe.plugin(SessionPersistenceJsonl, { root })
  433. const loaded = await probe.sessionPersistence.load(SessionId('inject-sess'))
  434. expect(JSON.stringify(loaded.events)).toContain('background task 42 finished')
  435. await probe.fiber.dispose()
  436. await ctx1.fiber.dispose()
  437. })
  438. it('an idle inject() survives persist + resume (turn-enclosed, not dropped as crash tail)', async () => {
  439. // Turn enclosure keeps idle context out of crash-tail repair, so it must
  440. // survive persistence and resume.
  441. const adapter1 = new MockAdapter([textResponse('answer')])
  442. const { ctx: ctx1, root } = await persistentHarness(adapter1)
  443. const a1 = (await ctx1.agents.create({ agentId: AgentId('m'), sessionId: SessionId('inject-sess'), meta: { cwd: '/w' } })).agent as ReactLoopAgent
  444. a1.send([{ type: 'text', text: 'q' }], { source: { kind: 'user' } })
  445. await waitForIdle(ctx1, a1)
  446. a1.inject([{ type: 'text', text: 'background task 42 finished' }], { source: { kind: 'plugin', plugin: 'tool-bash' } })
  447. await ctx1.parallel('session/flush', a1.session)
  448. await ctx1.fiber.dispose()
  449. // Lifecycle 2: resume; the injected context is still in the derived history.
  450. const adapter2 = new MockAdapter([textResponse('next')])
  451. const ctx2 = new Context()
  452. await ctx2.plugin(LlmService)
  453. await ctx2.plugin(SessionStore)
  454. await ctx2.plugin(SystemPrompt)
  455. await ctx2.plugin(ToolRegistry)
  456. await ctx2.plugin(AgentRegistry)
  457. await ctx2.plugin(AgentLoop, { agents: [] })
  458. await ctx2.plugin(SessionPersistenceJsonl, { root })
  459. ctx2.llm.registerAdapter(['mock'], adapter2)
  460. const a2 = (await ctx2.agents.resume({ agentId: AgentId('m'), resumeSessionId: SessionId('inject-sess') })).agent as ReactLoopAgent
  461. const flat = JSON.stringify(a2.session.deriveMessages())
  462. expect(flat).toContain('background task 42 finished')
  463. await ctx2.fiber.dispose()
  464. })
  465. it('resume reloads a persisted session: history + turn numbering continue, no duplicate seqs', async () => {
  466. // Lifecycle 1: run one full turn, persisting it.
  467. const adapter1 = new MockAdapter([textResponse('first answer')])
  468. const { ctx: ctx1, root } = await persistentHarness(adapter1)
  469. const a1 = (await ctx1.agents.create({ agentId: AgentId('main'), sessionId: SessionId('sess-resume'), meta: { cwd: '/w' } })).agent as ReactLoopAgent
  470. a1.send([{ type: 'text', text: 'first question' }], { source: { kind: 'user' } })
  471. await waitForIdle(ctx1, a1)
  472. const events1 = [...a1.session.events]
  473. const seqs1 = events1.map(e => e.seq)
  474. expect(seqs1).toEqual([...seqs1].sort((x, y) => x - y)) // contiguous
  475. await ctx1.fiber.dispose()
  476. // Lifecycle 2: a brand-new context over the SAME root; resume the session.
  477. const adapter2 = new MockAdapter([textResponse('second answer')])
  478. const ctx2 = new Context()
  479. await ctx2.plugin(LlmService)
  480. await ctx2.plugin(SessionStore)
  481. await ctx2.plugin(SystemPrompt)
  482. await ctx2.plugin(ToolRegistry)
  483. await ctx2.plugin(AgentRegistry)
  484. await ctx2.plugin(AgentLoop, { agents: [] })
  485. await ctx2.plugin(SessionPersistenceJsonl, { root })
  486. ctx2.llm.registerAdapter(['mock'], adapter2)
  487. const a2 = (await ctx2.agents.resume({ agentId: AgentId('main'), resumeSessionId: SessionId('sess-resume') })).agent as ReactLoopAgent
  488. // The resumed session carries the prior history…
  489. expect(a2.session.id).toBe('sess-resume')
  490. expect(a2.session.events.length).toBe(events1.length)
  491. const replay = new Session(SessionId('replay'), events1)
  492. expect(a2.session.deriveMessages()).toEqual(replay.deriveMessages())
  493. // …and a new turn continues numbering (turn 2) with contiguous seqs.
  494. a2.send([{ type: 'text', text: 'second question' }], { source: { kind: 'user' } })
  495. await waitForIdle(ctx2, a2)
  496. const allSeqs = a2.session.events.map(e => e.seq)
  497. expect(allSeqs).toEqual(allSeqs.map((_, i) => i)) // 0..N contiguous, no duplicates
  498. const turnStarts = a2.session.events.filter(e => e.type === 'turn/start')
  499. expect(turnStarts.map(e => e.type === 'turn/start' && e.data.turn)).toEqual([1, 2])
  500. await ctx2.fiber.dispose()
  501. })
  502. it('resume rejects when session persistence is not configured', async () => {
  503. // A harness WITHOUT the persistence plugin.
  504. const adapter = new MockAdapter([textResponse('x')])
  505. const ctx = new Context()
  506. await ctx.plugin(LlmService)
  507. await ctx.plugin(SessionStore)
  508. await ctx.plugin(SystemPrompt)
  509. await ctx.plugin(ToolRegistry)
  510. await ctx.plugin(AgentRegistry)
  511. await ctx.plugin(AgentLoop, { agents: [] })
  512. ctx.llm.registerAdapter(['mock'], adapter)
  513. await expect(ctx.agents.resume({ agentId: AgentId('m'), resumeSessionId: SessionId('nope') }))
  514. .rejects.toThrow(/session persistence is not configured/)
  515. await ctx.fiber.dispose()
  516. })
  517. })