manager.spec.ts 51 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001100210031004100510061007100810091010101110121013101410151016101710181019102010211022102310241025102610271028102910301031103210331034103510361037103810391040104110421043104410451046104710481049105010511052105310541055105610571058105910601061106210631064106510661067106810691070107110721073107410751076107710781079108010811082108310841085108610871088108910901091109210931094109510961097109810991100110111021103110411051106110711081109111011111112
  1. /**
  2. * SessionManager orchestration: lazy resident instances, list lifecycle, host
  3. * frame routing, and the pending-frame buffer for uninstantiated sessions.
  4. */
  5. import { describe, expect, it, vi } from 'vitest'
  6. import type { SessionId } from '@deepseek-ai/dsh-client-connection/client'
  7. import { SessionManager } from '../src/client/sessions/manager.ts'
  8. import { FakeApiClient, deferred, err, ok } from './fake-api.ts'
  9. import { entries, plainTurn } from './event-script.ts'
  10. const S1 = 'fk-m1' as SessionId
  11. const S2 = 'fk-m2' as SessionId
  12. type SummaryOver = Partial<{
  13. updatedAt: number
  14. running: boolean
  15. blank: boolean
  16. parentSessionId: SessionId
  17. origin: 'subagent'
  18. }>
  19. function summary(sessionId: SessionId, over: SummaryOver = {}) {
  20. return { sessionId, updatedAt: 100, running: false, blank: false, ...over }
  21. }
  22. describe('instances', () => {
  23. it('lazily builds one resident instance per id and syncs the running bit from the list', async () => {
  24. const api = new FakeApiClient()
  25. api.onList = () => Promise.resolve(ok({ items: [summary(S1, { running: true })] as never[] }))
  26. const manager = new SessionManager(api)
  27. await manager.refreshList()
  28. const session = manager.get(S1)
  29. expect(manager.get(S1)).toBe(session) // resident: same instance forever
  30. expect(session.getSnapshot().running).toBe(true) // list preceded instantiation
  31. })
  32. it('replays buffered approval frames on instantiation and drops ordinary frames for uninstantiated sessions', () => {
  33. const api = new FakeApiClient()
  34. const manager = new SessionManager(api)
  35. // Uninstantiated: approval buffers, plain session/event drops.
  36. manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
  37. manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
  38. manager.handleMuxEnvelope({ rpcId: 're' as never, payload: { type: 'session/event', sessionId: S1, event: plainTurn(0, 0, 'x', 'y')[0] as never } })
  39. const session = manager.get(S1)
  40. expect(session.getSnapshot().pending).toMatchObject([{ kind: 'approval', payload: { approvalId: 'ap1' } }])
  41. // Buffer cleared: a second instantiation of another id gets nothing.
  42. expect(manager.get(S2).getSnapshot().pending).toEqual([])
  43. })
  44. it('retains every live answerable request and compacts resolutions before instantiation', () => {
  45. const api = new FakeApiClient()
  46. const manager = new SessionManager(api)
  47. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
  48. for (let i = 0; i < 40; i++) {
  49. manager.handleMuxEnvelope({ rpcId: `q${i}` as never, payload: { type: 'question/requested', sessionId: S1, questions: [] } })
  50. }
  51. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('question')
  52. for (let i = 0; i < 40; i++) {
  53. manager.handleMuxEnvelope({
  54. rpcId: `r${i}` as never,
  55. payload: { type: 'question/resolved', sessionId: S1, questionRpcId: `q${i}` as never, outcome: 'answered' },
  56. })
  57. }
  58. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  59. expect(manager.get(S1).getSnapshot().pending).toEqual([])
  60. })
  61. it('drops buffered answerable requests on session removal', () => {
  62. const manager = new SessionManager(new FakeApiClient())
  63. // Removed session: buffered frames must not replay on a future instantiation.
  64. manager.handleMuxEnvelope({ rpcId: 'qz' as never, payload: { type: 'question/requested', sessionId: S2, questions: [] } })
  65. manager.handleHostEnvelope({ rpcId: 'hz' as never, payload: { type: 'host/session-removed', sessionId: S2 } })
  66. expect(manager.get(S2).getSnapshot().pending).toEqual([])
  67. })
  68. })
  69. describe('list lifecycle', () => {
  70. it('single-flights refreshList and preserves the Host baseline order', async () => {
  71. const api = new FakeApiClient()
  72. const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
  73. api.onList = () => gate.promise
  74. const manager = new SessionManager(api)
  75. const first = manager.refreshList()
  76. const second = manager.refreshList()
  77. expect(manager.getListSnapshot().state).toBe('loading')
  78. gate.resolve(ok({ items: [summary(S2, { updatedAt: 200 }), summary(S1)] as never[] }))
  79. await Promise.all([first, second])
  80. expect(api.callsOf('session.list')).toHaveLength(1)
  81. const snapshot = manager.getListSnapshot()
  82. expect(snapshot.state).toBe('idle')
  83. expect(snapshot.items.map(i => i.sessionId)).toEqual([S2, S1])
  84. })
  85. it('replays incremental frames over hydration and never batch-reorders established ids', async () => {
  86. const api = new FakeApiClient()
  87. const first = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
  88. api.onList = () => first.promise
  89. const manager = new SessionManager(api)
  90. const hydration = manager.refreshList()
  91. manager.handleHostEnvelope({
  92. rpcId: 'during-first' as never,
  93. payload: { type: 'host/session-added', blank: true, sessionId: S2 },
  94. })
  95. first.resolve(ok({ items: [summary(S1)] as never[] }))
  96. await hydration
  97. expect(manager.getListSnapshot().items.map(item => item.sessionId)).toEqual([S2, S1])
  98. api.onList = () => Promise.resolve(ok({
  99. items: [summary(S1, { updatedAt: 900 }), summary(S2, { updatedAt: 800 })] as never[],
  100. }))
  101. await manager.refreshList()
  102. expect(manager.getListSnapshot().items.map(item => item.sessionId)).toEqual([S2, S1])
  103. })
  104. it('keeps the error in the list snapshot on failure', async () => {
  105. const api = new FakeApiClient()
  106. api.onList = () => Promise.resolve(err({ code: 'internal', message: 'boom', details: {} }))
  107. const manager = new SessionManager(api)
  108. await manager.refreshList()
  109. expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: { code: 'internal' } })
  110. // A failed pull does not step the arrival phase: still pending.
  111. expect(manager.getListSnapshot().phase).toBe('pending')
  112. })
  113. it('phase steps pending → ready on the first successful pull and never returns', async () => {
  114. const api = new FakeApiClient()
  115. const manager = new SessionManager(api)
  116. expect(manager.getListSnapshot().phase).toBe('pending')
  117. await manager.refreshList()
  118. expect(manager.getListSnapshot().phase).toBe('ready')
  119. // Sticky across later failures: the pull-activity axis reports the error,
  120. // the arrival phase holds.
  121. api.onList = () => Promise.resolve(err({ code: 'internal', message: 'down', details: {} }))
  122. await manager.refreshList()
  123. expect(manager.getListSnapshot()).toMatchObject({ state: 'error', phase: 'ready' })
  124. // And across an empty re-pull (empty-with-ready = truly no sessions).
  125. api.onList = () => Promise.resolve(ok({ items: [] as never[] }))
  126. await manager.refreshList()
  127. expect(manager.getListSnapshot()).toMatchObject({ state: 'idle', phase: 'ready' })
  128. expect(manager.getListSnapshot().items).toEqual([])
  129. })
  130. it('merges create into the list immediately without waiting for a refresh', async () => {
  131. const api = new FakeApiClient()
  132. api.onCreate = () => Promise.resolve(ok({ sessionId: S2 }))
  133. const manager = new SessionManager(api)
  134. const result = await manager.create()
  135. expect(result).toMatchObject({ ok: true, value: { sessionId: S2 } })
  136. expect(manager.getListSnapshot().items.map(i => i.sessionId)).toEqual([S2])
  137. })
  138. it('retains title projections before list arrival, keeps last-wins by seq, and clears them on removal', async () => {
  139. const api = new FakeApiClient()
  140. const manager = new SessionManager(api)
  141. const titleFrame = (rpcId: string, title: string, seq: number) => {
  142. manager.handleMuxEnvelope({
  143. rpcId: rpcId as never,
  144. payload: { type: 'session/projection', sessionId: S1, key: 'title', value: title, seq } as never,
  145. })
  146. }
  147. titleFrame('title-new', 'Newest', 4)
  148. titleFrame('title-stale', 'Stale', 3)
  149. titleFrame('title-equal', 'Equal', 4)
  150. api.onList = () => Promise.resolve(ok({
  151. items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[],
  152. }))
  153. await manager.refreshList()
  154. const titled = manager.getListSnapshot()
  155. expect(titled.items.map(item => item.sessionId)).toEqual([S1, S2])
  156. expect(titled.items[0]?.title).toBe('Newest')
  157. expect(titled.items[1]?.title).toBeUndefined()
  158. manager.handleHostEnvelope({ rpcId: 'removed' as never, payload: { type: 'host/session-removed', sessionId: S1 } })
  159. manager.handleHostEnvelope({ rpcId: 'readded' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } })
  160. expect(manager.getListSnapshot().items.find(item => item.sessionId === S1)?.title).toBeUndefined()
  161. })
  162. it('seeds cold titles from the list rows\' projections block under higher-seq-wins', async () => {
  163. const api = new FakeApiClient()
  164. const manager = new SessionManager(api)
  165. // A push frame landed before the list (S2's title is newer than the block's cut).
  166. manager.handleMuxEnvelope({
  167. rpcId: 'push-newer' as never,
  168. payload: { type: 'session/projection', sessionId: S2, key: 'title', value: 'Pushed', seq: 9 } as never,
  169. })
  170. api.onList = () => Promise.resolve(ok({
  171. items: [
  172. { ...summary(S1), projections: { asOfSeq: 4, values: { title: 'Cold cached' } } },
  173. { ...summary(S2, { updatedAt: 200 }), projections: { asOfSeq: 5, values: { title: 'List stale' } } },
  174. ] as never[],
  175. }))
  176. await manager.refreshList()
  177. const items = manager.getListSnapshot().items
  178. // Cold row: title surfaces straight from the list block — no open, no history.
  179. expect(items.find(item => item.sessionId === S1)?.title).toBe('Cold cached')
  180. // The stale list block (seq 5) cannot overwrite the newer push frame (seq 9).
  181. expect(items.find(item => item.sessionId === S2)?.title).toBe('Pushed')
  182. })
  183. it('drops a projection row beyond the subscription baseline before accepting its durable replay', async () => {
  184. const api = new FakeApiClient()
  185. api.onList = () => Promise.resolve(ok({ items: [summary(S1)] as never[] }))
  186. const manager = new SessionManager(api)
  187. await manager.refreshList()
  188. const frame = (rpcId: string, payload: object) => {
  189. manager.handleMuxEnvelope({ rpcId: rpcId as never, payload: payload as never })
  190. }
  191. frame('title-unflushed', { type: 'session/projection', sessionId: S1, key: 'title', value: 'Unflushed', seq: 4 })
  192. // The durable baseline says the host only knows up to seq 2: the phantom
  193. // row rode lost state and must drop, or last-wins pins it forever.
  194. frame('subscribed-recovered', { type: 'session/subscribed', sessionId: S1, lastSeq: 2 })
  195. expect(manager.getListSnapshot().items[0]?.title).toBeUndefined()
  196. frame('title-durable', { type: 'session/projection', sessionId: S1, key: 'title', value: 'Durable', seq: 2 })
  197. expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
  198. // A baseline at or past the row's seq keeps it (nothing phantom to drop).
  199. frame('subscribed-current', { type: 'session/subscribed', sessionId: S1, lastSeq: 2 })
  200. expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
  201. })
  202. })
  203. describe('search', () => {
  204. it('returns bounded Host results and forwards the caller signal', async () => {
  205. const api = new FakeApiClient()
  206. api.onSearch = () => Promise.resolve(ok({
  207. items: [{ sessionId: S1, snippet: 'matching excerpt' }],
  208. hasMore: true,
  209. }))
  210. const manager = new SessionManager(api)
  211. const signal = new AbortController().signal
  212. await expect(manager.search('exact phrase', signal)).resolves.toEqual({
  213. ok: true,
  214. value: {
  215. items: [{ sessionId: S1, snippet: 'matching excerpt' }],
  216. hasMore: true,
  217. },
  218. })
  219. expect(api.callsOf('session.search')).toEqual([{ query: 'exact phrase' }])
  220. expect(api.lastSearchSignal).toBe(signal)
  221. })
  222. it('preserves business errors and folds transport failures', async () => {
  223. const api = new FakeApiClient()
  224. const manager = new SessionManager(api)
  225. api.onSearch = () => Promise.resolve(err({
  226. code: 'internal',
  227. message: 'index unavailable',
  228. details: {},
  229. }))
  230. const signal = new AbortController().signal
  231. await expect(manager.search('first', signal)).resolves.toMatchObject({
  232. ok: false,
  233. error: { code: 'internal', message: 'index unavailable' },
  234. })
  235. api.onSearch = () => Promise.reject(new Error('wire down'))
  236. await expect(manager.search('second', signal)).resolves.toMatchObject({
  237. ok: false,
  238. error: { code: 'internal', message: 'wire down' },
  239. })
  240. })
  241. })
  242. describe('host frame routing', () => {
  243. it('adds/removes/flips sessions from host frames and keeps removed instances resident', async () => {
  244. const api = new FakeApiClient()
  245. const manager = new SessionManager(api)
  246. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } })
  247. manager.handleHostEnvelope({ rpcId: 'h2' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } }) // dup: ignored
  248. expect(manager.getListSnapshot().items).toHaveLength(1)
  249. const session = manager.get(S1)
  250. manager.handleHostEnvelope({ rpcId: 'h3' as never, payload: { type: 'host/session-status', sessionId: S1, running: true } })
  251. expect(session.getSnapshot().running).toBe(true)
  252. expect(manager.getListSnapshot().items[0]?.running).toBe(true)
  253. manager.handleHostEnvelope({ rpcId: 'h4' as never, payload: { type: 'host/agent-error', sessionId: S1, message: '炸了' } })
  254. expect(session.getSnapshot().lastAgentError).toBe('炸了')
  255. manager.handleHostEnvelope({ rpcId: 'h5' as never, payload: { type: 'host/session-removed', sessionId: S1 } })
  256. expect(manager.getListSnapshot().items).toHaveLength(0)
  257. expect(session.getSnapshot().removed).toBe(true)
  258. expect(manager.get(S1)).toBe(session) // resident-instance rule survives removal
  259. })
  260. })
  261. describe('subagent catalogs', () => {
  262. it('keeps a catalog-discovered child address across ordinary selection and status frames', async () => {
  263. const api = new FakeApiClient()
  264. api.onList = () => Promise.resolve(ok({ items: [
  265. summary(S1),
  266. summary(S2, { parentSessionId: S1, origin: 'subagent' }),
  267. ] as never[] }))
  268. api.onSubagentList = () => Promise.resolve(ok({
  269. entries: [{
  270. kind: 'child', id: S2, mode: 'continuable', label: 'worker',
  271. activity: 'running', hasChildren: false,
  272. }] as never[],
  273. parentAvailable: true,
  274. }))
  275. const manager = new SessionManager(api)
  276. await manager.refreshList()
  277. await manager.refreshSubagents(S1)
  278. manager.selectSubagent({ parentSessionId: S1, childSessionId: S2, mode: 'continuable' })
  279. expect(manager.getListSnapshot().currentAddress).toEqual({
  280. parentSessionId: S1, childSessionId: S2, mode: 'continuable',
  281. })
  282. expect(manager.get(S2).getSnapshot().subagent).toEqual({
  283. address: { parentSessionId: S1, childSessionId: S2, mode: 'continuable' },
  284. parentAvailable: true,
  285. })
  286. // Clicking the same child through an ordinary list-selection path must not
  287. // erase the catalog-derived address and fall back to session.* transport.
  288. manager.select(S2)
  289. expect(manager.getListSnapshot().currentAddress).toEqual({
  290. parentSessionId: S1, childSessionId: S2, mode: 'continuable',
  291. })
  292. expect(manager.get(S2).getSnapshot().subagent).toEqual({
  293. address: { parentSessionId: S1, childSessionId: S2, mode: 'continuable' },
  294. parentAvailable: true,
  295. })
  296. await manager.get(S2).open()
  297. await manager.get(S2).prompt([{ type: 'text', text: 'continue' }], 'queue')
  298. expect(api.callsOf('subagent.history')).toEqual([
  299. { parentSessionId: S1, childSessionId: S2, mode: 'continuable', maxMessages: 50 },
  300. ])
  301. expect(api.callsOf('subagent.prompt')).toEqual([
  302. {
  303. parentSessionId: S1, childSessionId: S2, mode: 'continuable',
  304. content: [{ type: 'text', text: 'continue' }],
  305. },
  306. ])
  307. expect(api.callsOf('session.history')).toEqual([])
  308. expect(api.callsOf('session.prompt')).toEqual([])
  309. const listCalls = api.callsOf('subagent.list').length
  310. manager.handleHostEnvelope({
  311. rpcId: 'child-complete' as never,
  312. payload: { type: 'host/session-status', sessionId: S2, running: false },
  313. })
  314. expect(manager.getListSnapshot().subagentsByParent[S1]?.entries[0]).toMatchObject({
  315. kind: 'child', id: S2, activity: 'inactive',
  316. })
  317. expect(api.callsOf('subagent.list')).toHaveLength(listCalls)
  318. manager.handleHostEnvelope({
  319. rpcId: 'child-detached' as never,
  320. payload: { type: 'host/session-removed', sessionId: S2 },
  321. })
  322. expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)).toMatchObject({
  323. origin: 'subagent', parentSessionId: S1, running: false,
  324. })
  325. expect(manager.get(S2).getSnapshot()).toMatchObject({
  326. removed: false,
  327. subagent: {
  328. address: { parentSessionId: S1, childSessionId: S2, mode: 'continuable' },
  329. },
  330. })
  331. })
  332. it('refetches debounced membership only while the parent catalog is open', async () => {
  333. vi.useFakeTimers()
  334. try {
  335. const api = new FakeApiClient()
  336. const manager = new SessionManager(api)
  337. await manager.refreshSubagents(S1)
  338. manager.setSubagentCatalogOpen(S1, true)
  339. await Promise.resolve()
  340. const baseline = api.callsOf('subagent.list').length
  341. manager.handleHostEnvelope({
  342. rpcId: 'child-added' as never,
  343. payload: {
  344. type: 'host/session-added', sessionId: S2, parentSessionId: S1, blank: false,
  345. },
  346. })
  347. manager.handleHostEnvelope({
  348. rpcId: 'child-added-again' as never,
  349. payload: {
  350. type: 'host/session-added', sessionId: 'fk-m3' as SessionId, parentSessionId: S1, blank: false,
  351. },
  352. })
  353. await vi.advanceTimersByTimeAsync(50)
  354. expect(api.callsOf('subagent.list')).toHaveLength(baseline + 1)
  355. manager.setSubagentCatalogOpen(S1, false)
  356. manager.handleHostEnvelope({
  357. rpcId: 'child-added-closed' as never,
  358. payload: {
  359. type: 'host/session-added', sessionId: 'fk-m4' as SessionId, parentSessionId: S1, blank: false,
  360. },
  361. })
  362. await vi.advanceTimersByTimeAsync(50)
  363. expect(api.callsOf('subagent.list')).toHaveLength(baseline + 1)
  364. } finally {
  365. vi.useRealTimers()
  366. }
  367. })
  368. it('marks a loaded parent row expandable only for a direct subagent publication', async () => {
  369. const api = new FakeApiClient()
  370. const root = 'fk-root' as SessionId
  371. api.onSubagentList = () => Promise.resolve(ok({
  372. entries: [
  373. {
  374. kind: 'child', id: S1, mode: 'continuable', label: 'parent',
  375. activity: 'inactive', hasChildren: false,
  376. },
  377. {
  378. kind: 'child', id: S2, mode: 'continuable', label: 'ordinary parent',
  379. activity: 'inactive', hasChildren: false,
  380. },
  381. ] as never[],
  382. parentAvailable: true,
  383. }))
  384. const manager = new SessionManager(api)
  385. await manager.refreshSubagents(root)
  386. manager.handleHostEnvelope({
  387. rpcId: 'nested-subagent' as never,
  388. payload: {
  389. type: 'host/session-added', sessionId: 'fk-grandchild' as SessionId,
  390. parentSessionId: S1, origin: 'subagent', blank: false,
  391. },
  392. })
  393. manager.handleHostEnvelope({
  394. rpcId: 'ordinary-fork' as never,
  395. payload: {
  396. type: 'host/session-added', sessionId: 'fk-fork' as SessionId,
  397. parentSessionId: S2, blank: false,
  398. },
  399. })
  400. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  401. { kind: 'child', id: S1, hasChildren: true },
  402. { kind: 'child', id: S2, hasChildren: false },
  403. ])
  404. })
  405. it('preserves a live expandability hint across only the older in-flight catalog response', async () => {
  406. const api = new FakeApiClient()
  407. const root = 'fk-root' as SessionId
  408. const response = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  409. api.onSubagentList = () => response.promise
  410. const manager = new SessionManager(api)
  411. const refresh = manager.refreshSubagents(root)
  412. manager.handleHostEnvelope({
  413. rpcId: 'nested-subagent' as never,
  414. payload: {
  415. type: 'host/session-added', sessionId: 'fk-grandchild' as SessionId,
  416. parentSessionId: S1, origin: 'subagent', blank: false,
  417. },
  418. })
  419. response.resolve(ok({
  420. entries: [{
  421. kind: 'child', id: S1, mode: 'continuable', label: 'parent',
  422. activity: 'inactive', hasChildren: false,
  423. }] as never[],
  424. parentAvailable: true,
  425. }))
  426. await refresh
  427. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  428. { kind: 'child', id: S1, hasChildren: true },
  429. ])
  430. api.onSubagentList = () => Promise.resolve(ok({
  431. entries: [{
  432. kind: 'child', id: S1, mode: 'continuable', label: 'parent',
  433. activity: 'inactive', hasChildren: false,
  434. }] as never[],
  435. parentAvailable: true,
  436. }))
  437. await manager.refreshSubagents(root)
  438. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  439. { kind: 'child', id: S1, hasChildren: false },
  440. ])
  441. })
  442. it('replays status frames over an older in-flight catalog response', async () => {
  443. const api = new FakeApiClient()
  444. const root = 'fk-root' as SessionId
  445. const response = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  446. api.onSubagentList = () => response.promise
  447. const manager = new SessionManager(api)
  448. const refresh = manager.refreshSubagents(root)
  449. manager.handleHostEnvelope({
  450. rpcId: 'child-stopped' as never,
  451. payload: { type: 'host/session-status', sessionId: S1, running: false },
  452. })
  453. manager.handleHostEnvelope({
  454. rpcId: 'child-started' as never,
  455. payload: { type: 'host/session-status', sessionId: S2, running: true },
  456. })
  457. response.resolve(ok({
  458. entries: [
  459. {
  460. kind: 'child', id: S1, mode: 'continuable', label: 'stopped',
  461. activity: 'running', hasChildren: false,
  462. },
  463. {
  464. kind: 'child', id: S2, mode: 'continuable', label: 'started',
  465. activity: 'inactive', hasChildren: false,
  466. },
  467. ] as never[],
  468. parentAvailable: true,
  469. }))
  470. await refresh
  471. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  472. { kind: 'child', id: S1, activity: 'inactive' },
  473. { kind: 'child', id: S2, activity: 'running' },
  474. ])
  475. })
  476. it('marks a detached catalog child inactive without requiring a selected address', async () => {
  477. const api = new FakeApiClient()
  478. api.onSubagentList = () => Promise.resolve(ok({
  479. entries: [{
  480. kind: 'child', id: S2, mode: 'continuable', label: 'worker',
  481. activity: 'running', hasChildren: false,
  482. }] as never[],
  483. parentAvailable: true,
  484. }))
  485. const manager = new SessionManager(api)
  486. await manager.refreshSubagents(S1)
  487. manager.handleHostEnvelope({
  488. rpcId: 'child-detached' as never,
  489. payload: { type: 'host/session-removed', sessionId: S2 },
  490. })
  491. expect(manager.getListSnapshot().subagentsByParent[S1]?.entries).toMatchObject([
  492. { kind: 'child', id: S2, activity: 'inactive' },
  493. ])
  494. })
  495. it('coalesces overlapping catalog reads without scheduling a trailing pull', async () => {
  496. const api = new FakeApiClient()
  497. const root = 'fk-root' as SessionId
  498. const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  499. api.onSubagentList = () => first.promise
  500. const manager = new SessionManager(api)
  501. const refresh = manager.refreshSubagents(root)
  502. expect(manager.refreshSubagents(root)).toBe(refresh)
  503. api.onSubagentList = () => Promise.resolve(ok({ entries: [], parentAvailable: true }))
  504. first.resolve(ok({ entries: [], parentAvailable: true }))
  505. await refresh
  506. expect(api.callsOf('subagent.list')).toHaveLength(1)
  507. })
  508. it('runs one trailing catalog refresh for a membership change coalesced into an in-flight pull', async () => {
  509. vi.useFakeTimers()
  510. try {
  511. const api = new FakeApiClient()
  512. const root = 'fk-root' as SessionId
  513. const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  514. const second = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  515. api.onSubagentList = () => first.promise
  516. const manager = new SessionManager(api, root)
  517. const refresh = manager.refreshSubagents(root)
  518. // A membership frame arrives while the pull is in flight; the debounced
  519. // refresh it schedules fires 50ms later and is coalesced into the pull —
  520. // which was requested before the new child existed. The stale mark must
  521. // queue one trailing pull carrying the change.
  522. manager.handleHostEnvelope({
  523. rpcId: 'child-added' as never,
  524. payload: {
  525. type: 'host/session-added', sessionId: S2, parentSessionId: root, blank: false,
  526. },
  527. })
  528. await vi.advanceTimersByTimeAsync(50)
  529. api.onSubagentList = () => second.promise
  530. first.resolve(ok({
  531. entries: [{
  532. kind: 'child', id: S1, mode: 'continuable', label: 'older',
  533. activity: 'inactive', hasChildren: false,
  534. }] as never[],
  535. parentAvailable: true,
  536. }))
  537. await refresh
  538. // The trailing pull is already in flight (kicked synchronously in finally).
  539. second.resolve(ok({
  540. entries: [
  541. {
  542. kind: 'child', id: S1, mode: 'continuable', label: 'older',
  543. activity: 'inactive', hasChildren: false,
  544. },
  545. {
  546. kind: 'child', id: S2, mode: 'continuable', label: 'new child',
  547. activity: 'inactive', hasChildren: false,
  548. },
  549. ] as never[],
  550. parentAvailable: true,
  551. }))
  552. await second.promise
  553. expect(api.callsOf('subagent.list')).toHaveLength(2)
  554. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  555. { kind: 'child', id: S1, label: 'older' },
  556. { kind: 'child', id: S2, label: 'new child' },
  557. ])
  558. } finally {
  559. vi.useRealTimers()
  560. }
  561. })
  562. it('keeps removal invalidation across a stale success and failed trailing pull', async () => {
  563. const api = new FakeApiClient()
  564. const root = 'fk-root' as SessionId
  565. const child = () => ({
  566. kind: 'child' as const, id: S2, mode: 'continuable' as const, label: 'worker',
  567. activity: 'inactive' as const, hasChildren: false,
  568. })
  569. const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  570. api.onSubagentList = () => first.promise
  571. const manager = new SessionManager(api)
  572. const refresh = manager.refreshSubagents(root)
  573. first.resolve(ok({ entries: [child()] as never[], parentAvailable: true }))
  574. await refresh
  575. manager.selectSubagent({ parentSessionId: root, childSessionId: S2, mode: 'continuable' })
  576. // The removal lands while a second pull is in flight: the invalidation
  577. // must survive the pre-removal ok response, so one trailing pull runs.
  578. const mid = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  579. api.onSubagentList = () => mid.promise
  580. const midRefresh = manager.refreshSubagents(root)
  581. manager.handleHostEnvelope({
  582. rpcId: 'parent-removed-mid-pull' as never,
  583. payload: { type: 'host/session-removed', sessionId: root },
  584. })
  585. const trailing = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  586. api.onSubagentList = () => trailing.promise
  587. mid.resolve(ok({ entries: [child()] as never[], parentAvailable: true }))
  588. await midRefresh
  589. expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
  590. expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
  591. trailing.resolve(err({ code: 'internal', message: 'trailing pull failed', details: {} }))
  592. await vi.waitFor(() => {
  593. expect(manager.getListSnapshot().subagentsByParent[root]).toMatchObject({
  594. state: 'error',
  595. parentAvailable: false,
  596. })
  597. })
  598. const rootCalls = api.callsOf('subagent.list')
  599. .filter(call => (call as { parentSessionId: SessionId }).parentSessionId === root)
  600. expect(rootCalls).toHaveLength(3)
  601. expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
  602. expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
  603. })
  604. it('invalidates catalog availability when the owning parent is removed', async () => {
  605. const api = new FakeApiClient()
  606. const root = 'fk-root' as SessionId
  607. api.onSubagentList = () => Promise.resolve(ok({
  608. entries: [{
  609. kind: 'child', id: S2, mode: 'continuable', label: 'worker',
  610. activity: 'inactive', hasChildren: false,
  611. }] as never[],
  612. parentAvailable: true,
  613. }))
  614. const manager = new SessionManager(api)
  615. await manager.refreshSubagents(root)
  616. manager.selectSubagent({ parentSessionId: root, childSessionId: S2, mode: 'continuable' })
  617. expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: true })
  618. manager.handleHostEnvelope({
  619. rpcId: 'parent-removed' as never,
  620. payload: { type: 'host/session-removed', sessionId: root },
  621. })
  622. expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
  623. expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
  624. })
  625. })
  626. describe('remaining branches', () => {
  627. it('refreshList folds a transport throw into the error state', async () => {
  628. const api = new FakeApiClient()
  629. api.onList = () => Promise.reject(new Error('list wire down'))
  630. const manager = new SessionManager(api)
  631. await manager.refreshList()
  632. expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: { code: 'internal', message: 'list wire down' } })
  633. })
  634. it('refreshList pushes running bits down to already-instantiated sessions', async () => {
  635. const api = new FakeApiClient()
  636. const manager = new SessionManager(api)
  637. const session = manager.get(S1)
  638. api.onList = () => Promise.resolve(ok({ items: [summary(S1, { running: true })] as never[] }))
  639. await manager.refreshList()
  640. expect(session.getSnapshot().running).toBe(true)
  641. })
  642. it('create passes cwd and a preallocated id, folds transport throws, and deduplicates the echo', async () => {
  643. const api = new FakeApiClient()
  644. api.onCreate = () => Promise.resolve(ok({ sessionId: S1 }))
  645. const manager = new SessionManager(api)
  646. await manager.create({ cwd: '/tmp/w', sessionId: S1 })
  647. expect(api.callsOf('session.create')).toEqual([{ cwd: '/tmp/w', sessionId: S1 }])
  648. expect(manager.getListSnapshot().items[0]).toMatchObject({ sessionId: S1, cwd: '/tmp/w' })
  649. await manager.create({ cwd: '/tmp/w' }) // same id returned: no duplicate row
  650. expect(manager.getListSnapshot().items).toHaveLength(1)
  651. api.onCreate = () => Promise.reject(new Error('create wire down'))
  652. expect(await manager.create()).toMatchObject({ ok: false, error: { code: 'internal' } })
  653. // Business error passes through untouched.
  654. api.onCreate = () => Promise.resolve(err({ code: 'internal', message: 'no', details: {} }))
  655. expect(await manager.create()).toMatchObject({ ok: false })
  656. })
  657. it('publishes a real Ungrouped summary from workspace-attach-failed', async () => {
  658. const api = new FakeApiClient()
  659. api.onCreate = () => Promise.resolve(err({
  660. code: 'workspace-attach-failed',
  661. message: 'published but unattached',
  662. details: { sessionId: S1, workspaceId: 'w1' },
  663. } as never))
  664. const manager = new SessionManager(api)
  665. const result = await manager.create({ workspaceId: 'w1' as never, sessionId: S1 })
  666. expect(result).toMatchObject({ ok: false, error: { code: 'workspace-attach-failed' } })
  667. expect(manager.getListSnapshot().items).toEqual([expect.objectContaining({ sessionId: S1 })])
  668. expect(manager.getListSnapshot().items[0]).not.toHaveProperty('cwd')
  669. })
  670. it('reconciles a fork child published before workspace attachment fails', async () => {
  671. const api = new FakeApiClient()
  672. api.onFork = () => Promise.resolve(err({
  673. code: 'workspace-attach-failed',
  674. message: 'forked but unattached',
  675. details: { sessionId: S2, workspaceId: 'w1' },
  676. } as never))
  677. const manager = new SessionManager(api)
  678. const result = await manager.fork({ sessionId: S1 })
  679. expect(result).toMatchObject({ ok: false, error: { code: 'workspace-attach-failed' } })
  680. expect(manager.getListSnapshot().items).toEqual([expect.objectContaining({
  681. sessionId: S2,
  682. parentSessionId: S1,
  683. blank: false,
  684. })])
  685. })
  686. it('reconciles a preallocated id after an ordinary transport failure', async () => {
  687. const api = new FakeApiClient()
  688. api.onCreate = () => Promise.reject(new Error('response lost'))
  689. const manager = new SessionManager(api)
  690. const failed = await manager.create({ workspaceId: 'w1' as never, sessionId: S1 })
  691. expect(failed).toMatchObject({ ok: false, error: { message: 'response lost' } })
  692. expect(manager.getListSnapshot().items).toEqual([])
  693. manager.handleHostEnvelope({
  694. rpcId: 'published-later' as never,
  695. payload: { type: 'host/session-added', blank: true, sessionId: S1, cwd: '/w/one' },
  696. })
  697. expect(manager.getListSnapshot().items).toEqual([
  698. expect.objectContaining({ sessionId: S1, cwd: '/w/one' }),
  699. ])
  700. manager.handleHostEnvelope({
  701. rpcId: 'duplicate-frame' as never,
  702. payload: { type: 'host/session-added', blank: true, sessionId: S1, cwd: '/w/one' },
  703. })
  704. expect(manager.getListSnapshot().items).toHaveLength(1)
  705. })
  706. it('subscribe notifies on list changes and stops after unsubscribe', async () => {
  707. const api = new FakeApiClient()
  708. const manager = new SessionManager(api)
  709. let notified = 0
  710. const unsubscribe = manager.subscribe(() => { notified++ })
  711. await manager.refreshList()
  712. await new Promise(resolve => setTimeout(resolve, 0))
  713. expect(notified).toBeGreaterThan(0)
  714. const seen = notified
  715. unsubscribe()
  716. manager.handleHostEnvelope({ rpcId: 'h' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } })
  717. await new Promise(resolve => setTimeout(resolve, 0))
  718. expect(notified).toBe(seen)
  719. })
  720. it('routes stream/error and unknown frames to the documented drops, and dispatches to instantiated sessions', () => {
  721. const api = new FakeApiClient()
  722. const manager = new SessionManager(api)
  723. manager.handleMuxEnvelope({ rpcId: 'e' as never, payload: { type: 'stream/error', error: { code: 'internal', message: 'x', details: {} } } })
  724. manager.handleHostEnvelope({ rpcId: 'e2' as never, payload: { type: 'stream/error', error: { code: 'internal', message: 'x', details: {} } } })
  725. manager.handleHostEnvelope({ rpcId: 'e3' as never, payload: { type: 'future/host-frame' } as never })
  726. const session = manager.get(S1)
  727. manager.handleMuxEnvelope({ rpcId: 'q1' as never, payload: { type: 'question/requested', sessionId: S1, questions: [] } })
  728. expect(session.getSnapshot().pending).toMatchObject([{ kind: 'question' }])
  729. // status flip for an unknown session only touches summaries (no crash).
  730. manager.handleHostEnvelope({ rpcId: 'h9' as never, payload: { type: 'host/session-status', sessionId: S2, running: true } })
  731. manager.handleHostEnvelope({ rpcId: 'ha' as never, payload: { type: 'host/agent-error', sessionId: S2, message: '无实例' } })
  732. })
  733. it('keeps list-entry identity for unchanged rows across an unrelated list change', async () => {
  734. const api = new FakeApiClient()
  735. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
  736. const manager = new SessionManager(api)
  737. await manager.refreshList()
  738. const before = manager.getListSnapshot()
  739. manager.handleHostEnvelope({ rpcId: 'h' as never, payload: { type: 'host/session-status', sessionId: S2, running: true } })
  740. const after = manager.getListSnapshot()
  741. expect(after.items).not.toBe(before.items)
  742. const beforeS1 = before.items.find(e => e.sessionId === S1)
  743. const afterS1 = after.items.find(e => e.sessionId === S1)
  744. expect(afterS1).toBe(beforeS1) // untouched entry keeps identity (entryCache)
  745. // Same-order same-entries snapshot reuses the items array.
  746. manager.handleHostEnvelope({ rpcId: 'h2' as never, payload: { type: 'host/agent-error', sessionId: S1, message: 'x' } })
  747. expect(manager.getListSnapshot().items).toBe(after.items)
  748. })
  749. it('carries parentSessionId from host/session-added into the lineage row', () => {
  750. const api = new FakeApiClient()
  751. const manager = new SessionManager(api)
  752. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', blank: true, sessionId: S1 } })
  753. manager.handleHostEnvelope({
  754. rpcId: 'h2' as never,
  755. payload: {
  756. type: 'host/session-added', blank: true, sessionId: S2,
  757. parentSessionId: S1, origin: 'subagent',
  758. },
  759. })
  760. const items = manager.getListSnapshot().items
  761. expect(items.find(e => e.sessionId === S2)).toMatchObject({
  762. parentSessionId: S1, origin: 'subagent', depth: 1,
  763. })
  764. })
  765. })
  766. describe('connected generation', () => {
  767. it('refreshes the list and resyncs only opened instances', async () => {
  768. const api = new FakeApiClient()
  769. api.onHistory = () => Promise.resolve(ok({
  770. events: entries(plainTurn(0, 0, 'a', 'b')) as never[],
  771. hasMore: false,
  772. modelTarget: { provider: 'deepseek-official', model: 'deepseek-chat' },
  773. }))
  774. const manager = new SessionManager(api)
  775. const openedSession = manager.get(S1)
  776. await openedSession.open()
  777. manager.get(S2) // instantiated but never opened
  778. const historyCallsBefore = api.callsOf('session.history').length
  779. manager.handleConnected()
  780. await vi.waitFor(() => {
  781. expect(api.callsOf('session.list').length).toBe(1)
  782. // Only the opened instance repulls history; the cold one stays silent.
  783. expect(api.callsOf('session.history').length).toBe(historyCallsBefore + 1)
  784. })
  785. })
  786. it('reloads the durable parent address for a restored child selection', async () => {
  787. const api = new FakeApiClient()
  788. const address = {
  789. parentSessionId: S1, childSessionId: S2, mode: 'continuable' as const,
  790. }
  791. const manager = new SessionManager(api, S2, address)
  792. manager.handleConnected()
  793. await vi.waitFor(() => {
  794. expect(api.callsOf('subagent.list')).toContainEqual({ parentSessionId: S1 })
  795. })
  796. expect(manager.getListSnapshot().currentAddress).toEqual(address)
  797. })
  798. })
  799. describe('pending-interaction list status', () => {
  800. it('tracks approval requests through replay and resolution without instantiation', () => {
  801. const manager = new SessionManager(new FakeApiClient())
  802. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
  803. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  804. manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
  805. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
  806. // Mux-open replay of the same question (same approvalId) is idempotent.
  807. manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
  808. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
  809. manager.handleMuxEnvelope({ rpcId: 'rx' as never, payload: { type: 'approval/resolved', sessionId: S1, approvalId: 'ap1' as never, outcome: 'allowed-once' as never } })
  810. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  811. })
  812. it('classifies ordinary questions and renderable plan reviews, then clears by question rpcId', () => {
  813. const manager = new SessionManager(new FakeApiClient())
  814. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
  815. manager.handleMuxEnvelope({
  816. rpcId: 'q1' as never,
  817. payload: { type: 'question/requested', sessionId: S1, questions: [{ id: 'name', question: 'Name?' }] },
  818. })
  819. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('question')
  820. manager.handleMuxEnvelope({ rpcId: 'qx' as never, payload: { type: 'question/resolved', sessionId: S1, questionRpcId: 'q1' as never, outcome: 'answered' } })
  821. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  822. manager.handleMuxEnvelope({
  823. rpcId: 'q2' as never,
  824. payload: {
  825. type: 'question/requested',
  826. sessionId: S1,
  827. questions: [{
  828. id: 'plan', question: 'Approve?', detail: '# Plan',
  829. options: [{ label: 'Approve' }, { label: 'Refuse' }],
  830. intent: { kind: 'plan-review', approve: 'Approve' },
  831. }],
  832. },
  833. })
  834. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('plan-review')
  835. manager.handleMuxEnvelope({ rpcId: 'qy' as never, payload: { type: 'question/resolved', sessionId: S1, questionRpcId: 'q2' as never, outcome: 'cancelled' } })
  836. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  837. })
  838. it.each([
  839. ['missing detail', {}],
  840. ['multi-select', { detail: '# Plan', multiSelect: true }],
  841. ['more than two options', { detail: '# Plan', options: [{ label: 'Approve' }, { label: 'Refuse' }, { label: 'Revise' }] }],
  842. ['missing approve option', { detail: '# Plan', options: [{ label: 'Refuse' }] }],
  843. ])('keeps an unrenderable %s plan intent on the ordinary question flow', (_name, over) => {
  844. const manager = new SessionManager(new FakeApiClient())
  845. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
  846. manager.handleMuxEnvelope({
  847. rpcId: 'q-plan' as never,
  848. payload: {
  849. type: 'question/requested', sessionId: S1,
  850. questions: [{
  851. id: 'plan', question: 'Approve?', options: [{ label: 'Approve' }],
  852. intent: { kind: 'plan-review', approve: 'Approve' },
  853. ...over,
  854. }],
  855. },
  856. })
  857. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('question')
  858. })
  859. it('the first question outranks sibling approvals and resolving it reveals the remaining wait', () => {
  860. const manager = new SessionManager(new FakeApiClient())
  861. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
  862. manager.handleMuxEnvelope({ rpcId: 'r1' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'a1' as never, toolName: 'rm' } })
  863. manager.handleMuxEnvelope({
  864. rpcId: 'q1' as never,
  865. payload: { type: 'question/requested', sessionId: S1, questions: [{ id: 'name', question: 'Name?' }] },
  866. })
  867. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('question')
  868. manager.handleMuxEnvelope({ rpcId: 'qy' as never, payload: { type: 'question/resolved', sessionId: S1, questionRpcId: 'q1' as never, outcome: 'answered' } })
  869. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
  870. manager.handleMuxEnvelope({ rpcId: 'rx' as never, payload: { type: 'approval/resolved', sessionId: S1, approvalId: 'a1' as never, outcome: 'rejected' as never } })
  871. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  872. manager.handleMuxEnvelope({ rpcId: 'r2' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'a2' as never, toolName: 'rm' } })
  873. manager.handleHostEnvelope({ rpcId: 'h2' as never, payload: { type: 'host/session-removed', sessionId: S1 } })
  874. expect(manager.getListSnapshot().items).toHaveLength(0)
  875. })
  876. it('drops stale status at generation death before replay re-adds live interactions', () => {
  877. const manager = new SessionManager(new FakeApiClient())
  878. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
  879. manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
  880. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
  881. // Generation death clears (resolved-while-disconnected questions send no frame)…
  882. manager.handleDisconnected()
  883. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBeUndefined()
  884. // …and a replayed frame arriving before onConnected (stream open precedes
  885. // the readiness handshake) survives the later handleConnected untouched.
  886. manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
  887. manager.handleConnected()
  888. expect(manager.getListSnapshot().items[0]?.pendingInteraction).toBe('approval')
  889. })
  890. it('generation death drops buffered answerable frames (a dead generation cannot be answered)', () => {
  891. const manager = new SessionManager(new FakeApiClient())
  892. manager.handleHostEnvelope({ rpcId: 'h1' as never, payload: { type: 'host/session-added', sessionId: S1, blank: false } })
  893. // Buffered pre-instantiation: an approval pair and a queued row.
  894. manager.handleMuxEnvelope({ rpcId: 'ra' as never, payload: { type: 'approval/requested', sessionId: S1, approvalId: 'ap1' as never, toolName: 'rm' } })
  895. manager.handleMuxEnvelope({ rpcId: 'q1' as never, payload: { type: 'question/requested', sessionId: S1, questions: [] } })
  896. manager.handleDisconnected()
  897. // Instantiate after the death sweep: no zombie interaction replays (the
  898. // pendingBuffers held only dead-generation rpcIds), so the session mints
  899. // no pending waits.
  900. const session = manager.get(S1)
  901. expect(session.getSnapshot().pending).toEqual([])
  902. })
  903. })
  904. describe('completed reminder', () => {
  905. const status = (rpcId: string, sessionId: SessionId, running: boolean) => ({
  906. rpcId: rpcId as never,
  907. payload: { type: 'host/session-status' as const, sessionId, running },
  908. })
  909. const added = (rpcId: string, sessionId: SessionId) => ({
  910. rpcId: rpcId as never,
  911. payload: { type: 'host/session-added' as const, sessionId, blank: false },
  912. })
  913. const entry = (manager: SessionManager, sessionId: SessionId) =>
  914. manager.getListSnapshot().items.find(item => item.sessionId === sessionId)
  915. it('arms on a running→idle flip of a non-selected session and clears on select', () => {
  916. const manager = new SessionManager(new FakeApiClient())
  917. manager.handleHostEnvelope(added('h1', S1))
  918. manager.handleHostEnvelope(added('h2', S2))
  919. manager.select(S1)
  920. expect(entry(manager, S2)?.completed).toBe(false)
  921. manager.handleHostEnvelope(status('s1', S2, true))
  922. manager.handleHostEnvelope(status('s2', S2, false))
  923. expect(entry(manager, S2)?.completed).toBe(true)
  924. // Opening the session consumes the reminder.
  925. manager.select(S2)
  926. expect(entry(manager, S2)?.completed).toBe(false)
  927. })
  928. it('never arms for the session being watched and re-arms after a switch-away re-run', () => {
  929. const manager = new SessionManager(new FakeApiClient())
  930. manager.handleHostEnvelope(added('h1', S1))
  931. manager.handleHostEnvelope(added('h2', S2))
  932. manager.select(S2)
  933. manager.handleHostEnvelope(status('s1', S2, true))
  934. manager.handleHostEnvelope(status('s2', S2, false))
  935. expect(entry(manager, S2)?.completed).toBe(false) // watched to completion: no reminder
  936. // Switch away; a fresh run completing again arms the reminder.
  937. manager.select(S1)
  938. manager.handleHostEnvelope(status('s3', S2, true))
  939. manager.handleHostEnvelope(status('s4', S2, false))
  940. expect(entry(manager, S2)?.completed).toBe(true)
  941. })
  942. it('a re-run disarms the reminder while running and re-arms on its completion', () => {
  943. const manager = new SessionManager(new FakeApiClient())
  944. manager.handleHostEnvelope(added('h1', S1))
  945. manager.handleHostEnvelope(added('h2', S2))
  946. manager.select(S1)
  947. manager.handleHostEnvelope(status('s1', S2, true))
  948. manager.handleHostEnvelope(status('s2', S2, false))
  949. expect(entry(manager, S2)?.completed).toBe(true)
  950. // The user starts a new run without opening the session: running wins.
  951. manager.handleHostEnvelope(status('s3', S2, true))
  952. expect(entry(manager, S2)?.completed).toBe(false)
  953. manager.handleHostEnvelope(status('s4', S2, false))
  954. expect(entry(manager, S2)?.completed).toBe(true)
  955. })
  956. it('session-removed drops the reminder and a re-add starts clean', () => {
  957. const manager = new SessionManager(new FakeApiClient())
  958. manager.handleHostEnvelope(added('h1', S1))
  959. manager.handleHostEnvelope(added('h2', S2))
  960. manager.select(S1)
  961. manager.handleHostEnvelope(status('s1', S2, true))
  962. manager.handleHostEnvelope(status('s2', S2, false))
  963. expect(entry(manager, S2)?.completed).toBe(true)
  964. manager.handleHostEnvelope({ rpcId: 'rm' as never, payload: { type: 'host/session-removed', sessionId: S2 } })
  965. expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)).toBeUndefined()
  966. manager.handleHostEnvelope(added('h3', S2))
  967. expect(entry(manager, S2)?.completed).toBe(false)
  968. })
  969. it('a list refresh carrying the running→idle transition arms the reminder', async () => {
  970. const api = new FakeApiClient()
  971. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200, running: true })] as never[] }))
  972. const manager = new SessionManager(api)
  973. await manager.refreshList()
  974. manager.select(S1)
  975. expect(entry(manager, S2)?.completed).toBe(false)
  976. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200, running: false })] as never[] }))
  977. await manager.refreshList()
  978. expect(entry(manager, S2)?.completed).toBe(true)
  979. })
  980. it('never arms for sessions already idle at first observation', async () => {
  981. const api = new FakeApiClient()
  982. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
  983. const manager = new SessionManager(api)
  984. await manager.refreshList()
  985. manager.select(S1)
  986. expect(entry(manager, S2)?.completed).toBe(false)
  987. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 201 })] as never[] }))
  988. await manager.refreshList()
  989. expect(entry(manager, S2)?.completed).toBe(false)
  990. })
  991. it('arms a completion that happened during an in-flight first pull (baseline running, replayed idle)', async () => {
  992. const api = new FakeApiClient()
  993. const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
  994. api.onList = () => gate.promise
  995. const manager = new SessionManager(api)
  996. const refresh = manager.refreshList()
  997. // The session finishes while the first pull is still in flight; the pull
  998. // response recorded it as running at pull time.
  999. manager.handleHostEnvelope(status('s-mid', S2, false))
  1000. gate.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200, running: true })] as never[] }))
  1001. await refresh
  1002. expect(entry(manager, S2)?.completed).toBe(true)
  1003. })
  1004. it('arms when a session ran and completed entirely between in-flight mutations (baseline idle)', async () => {
  1005. const api = new FakeApiClient()
  1006. const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
  1007. api.onList = () => gate.promise
  1008. const manager = new SessionManager(api)
  1009. const refresh = manager.refreshList()
  1010. // The unknown session starts and finishes while the first pull is in
  1011. // flight; the pull-time baseline recorded it idle, so the running→idle
  1012. // edge lives entirely inside the replayed mutations.
  1013. manager.handleHostEnvelope(status('s-start', S2, true))
  1014. manager.handleHostEnvelope(status('s-finish', S2, false))
  1015. gate.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
  1016. await refresh
  1017. expect(entry(manager, S2)?.completed).toBe(true)
  1018. })
  1019. })