manager.spec.ts 56 KB

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