manager.client.spec.ts 42 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182838485868788899091929394959697989910010110210310410510610710810911011111211311411511611711811912012112212312412512612712812913013113213313413513613713813914014114214314414514614714814915015115215315415515615715815916016116216316416516616716816917017117217317417517617717817918018118218318418518618718818919019119219319419519619719819920020120220320420520620720820921021121221321421521621721821922022122222322422522622722822923023123223323423523623723823924024124224324424524624724824925025125225325425525625725825926026126226326426526626726826927027127227327427527627727827928028128228328428528628728828929029129229329429529629729829930030130230330430530630730830931031131231331431531631731831932032132232332432532632732832933033133233333433533633733833934034134234334434534634734834935035135235335435535635735835936036136236336436536636736836937037137237337437537637737837938038138238338438538638738838939039139239339439539639739839940040140240340440540640740840941041141241341441541641741841942042142242342442542642742842943043143243343443543643743843944044144244344444544644744844945045145245345445545645745845946046146246346446546646746846947047147247347447547647747847948048148248348448548648748848949049149249349449549649749849950050150250350450550650750850951051151251351451551651751851952052152252352452552652752852953053153253353453553653753853954054154254354454554654754854955055155255355455555655755855956056156256356456556656756856957057157257357457557657757857958058158258358458558658758858959059159259359459559659759859960060160260360460560660760860961061161261361461561661761861962062162262362462562662762862963063163263363463563663763863964064164264364464564664764864965065165265365465565665765865966066166266366466566666766866967067167267367467567667767867968068168268368468568668768868969069169269369469569669769869970070170270370470570670770870971071171271371471571671771871972072172272372472572672772872973073173273373473573673773873974074174274374474574674774874975075175275375475575675775875976076176276376476576676776876977077177277377477577677777877978078178278378478578678778878979079179279379479579679779879980080180280380480580680780880981081181281381481581681781881982082182282382482582682782882983083183283383483583683783883984084184284384484584684784884985085185285385485585685785885986086186286386486586686786886987087187287387487587687787887988088188288388488588688788888989089189289389489589689789889990090190290390490590690790890991091191291391491591691791891992092192292392492592692792892993093193293393493593693793893994094194294394494594694794894995095195295395495595695795895996096196296396496596696796896997097197297397497597697797897998098198298398498598698798898999099199299399499599699799899910001001
  1. /**
  2. * SessionManager orchestration: lazy resident instances, list lifecycle, host
  3. * frame routing, and control baselines for uninstantiated sessions.
  4. */
  5. import { describe, expect, it, vi } from 'vitest'
  6. import type { SessionId } from '@deepseek-ai/dsh-api-remotes/client'
  7. import type { SessionControlFrame } from '@deepseek-ai/dsh-api-session-controller/types'
  8. import type {} from '@deepseek-ai/dsh-session-title/client'
  9. import { SessionManager } from '../src/client/sessions/manager.ts'
  10. import { FakeApiClient, deferred, err, fakeRemote, ok } from './fake-api.client.ts'
  11. import { entries, plainTurn } from './event-script.client.ts'
  12. const S1 = 'fk-m1' as SessionId
  13. const S2 = 'fk-m2' as SessionId
  14. type SummaryOver = Partial<{
  15. updatedAt: number
  16. running: boolean
  17. blank: boolean
  18. cwd: string
  19. parentSessionId: SessionId
  20. origin: 'subagent'
  21. }>
  22. function summary(sessionId: SessionId, over: SummaryOver = {}) {
  23. return { sessionId, updatedAt: 100, running: false, blank: false, ...over }
  24. }
  25. function makeManager(): SessionManager {
  26. const api = new FakeApiClient()
  27. return new SessionManager(api, fakeRemote(api))
  28. }
  29. describe('SessionManager instances', () => {
  30. it('lazily builds one resident instance per id and syncs the running bit from the list', async () => {
  31. const api = new FakeApiClient()
  32. api.onList = () => Promise.resolve(ok({ items: [summary(S1, { running: true })] as never[] }))
  33. const manager = new SessionManager(api, fakeRemote(api))
  34. await manager.refreshList()
  35. const session = manager.get(S1)
  36. expect(manager.get(S1)).toBe(session) // resident: same instance forever
  37. expect(session.getSnapshot().running).toBe(true) // list preceded instantiation
  38. })
  39. })
  40. describe('list lifecycle', () => {
  41. it('single-flights refreshList and preserves the Host baseline order', async () => {
  42. const api = new FakeApiClient()
  43. const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
  44. api.onList = () => gate.promise
  45. const manager = new SessionManager(api, fakeRemote(api))
  46. const first = manager.refreshList()
  47. const second = manager.refreshList()
  48. expect(manager.getListSnapshot().state).toBe('loading')
  49. gate.resolve(ok({ items: [summary(S2, { updatedAt: 200 }), summary(S1)] as never[] }))
  50. await Promise.all([first, second])
  51. expect(api.callsOf('session.list')).toHaveLength(1)
  52. const snapshot = manager.getListSnapshot()
  53. expect(snapshot.state).toBe('idle')
  54. expect(snapshot.items.map(i => i.sessionId)).toEqual([S2, S1])
  55. })
  56. it('replays incremental frames over hydration and never batch-reorders established ids', async () => {
  57. const api = new FakeApiClient()
  58. const first = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
  59. api.onList = () => first.promise
  60. const manager = new SessionManager(api, fakeRemote(api))
  61. const hydration = manager.refreshList()
  62. manager.handleSessionAdded(summary(S2, { blank: true }))
  63. first.resolve(ok({ items: [summary(S1)] as never[] }))
  64. await hydration
  65. expect(manager.getListSnapshot().items.map(item => item.sessionId)).toEqual([S2, S1])
  66. api.onList = () => Promise.resolve(ok({
  67. items: [summary(S1, { updatedAt: 900 }), summary(S2, { updatedAt: 800 })] as never[],
  68. }))
  69. await manager.refreshList()
  70. expect(manager.getListSnapshot().items.map(item => item.sessionId)).toEqual([S2, S1])
  71. })
  72. it('advances list activity from the filtered Host notification', async () => {
  73. const api = new FakeApiClient()
  74. api.onList = () => Promise.resolve(ok({ items: [summary(S1)] as never[] }))
  75. const manager = new SessionManager(api, fakeRemote(api))
  76. await manager.refreshList()
  77. manager.handleSessionActivity(S1, 500)
  78. expect(manager.getListSnapshot().items[0]?.updatedAt).toBe(500)
  79. })
  80. it('keeps the error in the list snapshot on failure', async () => {
  81. const api = new FakeApiClient()
  82. api.onList = () => Promise.resolve(err({ code: 'internal', message: 'boom', details: {} }))
  83. const manager = new SessionManager(api, fakeRemote(api))
  84. await manager.refreshList()
  85. expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: { code: 'internal' } })
  86. // A failed pull does not step the arrival phase: still pending.
  87. expect(manager.getListSnapshot().phase).toBe('pending')
  88. })
  89. it('phase steps pending → ready on the first successful pull and never returns', async () => {
  90. const api = new FakeApiClient()
  91. const manager = new SessionManager(api, fakeRemote(api))
  92. expect(manager.getListSnapshot().phase).toBe('pending')
  93. await manager.refreshList()
  94. expect(manager.getListSnapshot().phase).toBe('ready')
  95. // Sticky across later failures: the pull-activity axis reports the error,
  96. // the arrival phase holds.
  97. api.onList = () => Promise.resolve(err({ code: 'internal', message: 'down', details: {} }))
  98. await manager.refreshList()
  99. expect(manager.getListSnapshot()).toMatchObject({ state: 'error', phase: 'ready' })
  100. // And across an empty re-pull (empty-with-ready = truly no sessions).
  101. api.onList = () => Promise.resolve(ok({ items: [] as never[] }))
  102. await manager.refreshList()
  103. expect(manager.getListSnapshot()).toMatchObject({ state: 'idle', phase: 'ready' })
  104. expect(manager.getListSnapshot().items).toEqual([])
  105. })
  106. it('merges create into the list immediately without waiting for a refresh', async () => {
  107. const api = new FakeApiClient()
  108. api.onCreate = () => Promise.resolve(ok({ sessionId: S2 }))
  109. const manager = new SessionManager(api, fakeRemote(api))
  110. const result = await manager.create()
  111. expect(result).toMatchObject({ ok: true, value: { sessionId: S2 } })
  112. expect(manager.getListSnapshot().items.map(i => i.sessionId)).toEqual([S2])
  113. })
  114. it('retains title projections before list arrival, keeps last-wins by seq, and clears them on removal', async () => {
  115. const api = new FakeApiClient()
  116. const manager = new SessionManager(api, fakeRemote(api))
  117. const titleFrame = (title: string, seq: number) => {
  118. manager.handleControlFrame({ type: 'projection', sessionId: S1, key: 'title', value: title, seq })
  119. }
  120. titleFrame('Newest', 4)
  121. titleFrame('Stale', 3)
  122. titleFrame('Equal', 4)
  123. api.onList = () => Promise.resolve(ok({
  124. items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[],
  125. }))
  126. await manager.refreshList()
  127. const titled = manager.getListSnapshot()
  128. expect(titled.items.map(item => item.sessionId)).toEqual([S1, S2])
  129. expect(titled.items[0]?.title).toBe('Newest')
  130. expect(titled.items[1]?.title).toBeUndefined()
  131. manager.handleSessionRemoved(S1)
  132. manager.handleSessionAdded(summary(S1, { blank: true }))
  133. expect(manager.getListSnapshot().items.find(item => item.sessionId === S1)?.title).toBeUndefined()
  134. })
  135. it('prewarms cold titles from list and session-added hints without replacing authoritative values', async () => {
  136. const api = new FakeApiClient()
  137. const manager = new SessionManager(api, fakeRemote(api))
  138. // A push frame landed before the list. Even a later cache watermark stays
  139. // tentative and cannot replace this authoritative value.
  140. manager.handleControlFrame({
  141. type: 'projection', sessionId: S2, key: 'title', value: 'Pushed', seq: 9,
  142. })
  143. api.onList = () => Promise.resolve(ok({
  144. items: [
  145. { ...summary(S1), projections: { asOfSeq: 4, values: { title: 'Cold cached' } } },
  146. { ...summary(S2, { updatedAt: 200 }), projections: { asOfSeq: 12, values: { title: 'List stale' } } },
  147. ] as never[],
  148. }))
  149. await manager.refreshList()
  150. const items = manager.getListSnapshot().items
  151. // Cold row: title surfaces straight from the list block — no open, no history.
  152. expect(items.find(item => item.sessionId === S1)?.title).toBe('Cold cached')
  153. expect(items.find(item => item.sessionId === S2)?.title).toBe('Pushed')
  154. manager.handleSessionAdded({
  155. ...summary(S2, { updatedAt: 300 }),
  156. projections: { asOfSeq: 15, values: { title: 'Added stale' } },
  157. })
  158. expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)?.title).toBe('Pushed')
  159. })
  160. it('drops a projection row beyond the subscription baseline before accepting its durable replay', async () => {
  161. const api = new FakeApiClient()
  162. api.onList = () => Promise.resolve(ok({ items: [summary(S1)] as never[] }))
  163. const manager = new SessionManager(api, fakeRemote(api))
  164. await manager.refreshList()
  165. const frame = (payload: SessionControlFrame) => { manager.handleControlFrame(payload) }
  166. frame({ type: 'projection', sessionId: S1, key: 'title', value: 'Unflushed', seq: 4 })
  167. // The durable baseline says the host only knows up to seq 2: the phantom
  168. // row rode lost state and must drop, or last-wins pins it forever.
  169. frame({
  170. type: 'baseline',
  171. value: {
  172. queues: {}, jobs: {},
  173. projections: { [S1]: { asOfSeq: 2, values: {} } },
  174. },
  175. })
  176. expect(manager.getListSnapshot().items[0]?.title).toBeUndefined()
  177. frame({ type: 'projection', sessionId: S1, key: 'title', value: 'Durable', seq: 2 })
  178. expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
  179. // A baseline at or past the row's seq keeps it (nothing phantom to drop).
  180. frame({
  181. type: 'baseline',
  182. value: {
  183. queues: {}, jobs: {},
  184. projections: { [S1]: { asOfSeq: 2, values: { title: 'Durable' } } },
  185. },
  186. })
  187. expect(manager.getListSnapshot().items[0]?.title).toBe('Durable')
  188. })
  189. })
  190. describe('search', () => {
  191. it('returns bounded Host results and forwards the caller signal', async () => {
  192. const api = new FakeApiClient()
  193. api.onSearch = () => Promise.resolve(ok({
  194. items: [{ sessionId: S1, snippet: 'matching excerpt' }],
  195. hasMore: true,
  196. }))
  197. const manager = new SessionManager(api, fakeRemote(api))
  198. const signal = new AbortController().signal
  199. await expect(manager.search('exact phrase', signal)).resolves.toEqual({
  200. ok: true,
  201. value: {
  202. items: [{ sessionId: S1, snippet: 'matching excerpt' }],
  203. hasMore: true,
  204. },
  205. })
  206. expect(api.callsOf('session.search')).toEqual([{ query: 'exact phrase' }])
  207. expect(api.lastSearchSignal).toBe(signal)
  208. })
  209. it('preserves business errors and folds transport failures', async () => {
  210. const api = new FakeApiClient()
  211. const manager = new SessionManager(api, fakeRemote(api))
  212. api.onSearch = () => Promise.resolve(err({
  213. code: 'internal',
  214. message: 'index unavailable',
  215. details: {},
  216. }))
  217. const signal = new AbortController().signal
  218. await expect(manager.search('first', signal)).resolves.toMatchObject({
  219. ok: false,
  220. error: { code: 'internal', message: 'index unavailable' },
  221. })
  222. api.onSearch = () => Promise.reject(new Error('wire down'))
  223. await expect(manager.search('second', signal)).resolves.toMatchObject({
  224. ok: false,
  225. error: { code: 'internal', message: 'wire down' },
  226. })
  227. })
  228. })
  229. describe('Host Remote event routing', () => {
  230. it('adds/removes/flips sessions and keeps removed instances resident', async () => {
  231. const api = new FakeApiClient()
  232. const manager = new SessionManager(api, fakeRemote(api))
  233. manager.handleSessionAdded(summary(S1, { blank: true }))
  234. manager.handleSessionAdded(summary(S1, { blank: true })) // dup: ignored
  235. expect(manager.getListSnapshot().items).toHaveLength(1)
  236. const session = manager.get(S1)
  237. manager.handleSessionStatus(S1, true)
  238. expect(session.getSnapshot().running).toBe(true)
  239. expect(manager.getListSnapshot().items[0]?.running).toBe(true)
  240. manager.handleSessionError(S1, '炸了')
  241. expect(session.getSnapshot().lastAgentError).toBe('炸了')
  242. manager.handleSessionRemoved(S1)
  243. expect(manager.getListSnapshot().items).toHaveLength(0)
  244. expect(session.getSnapshot().removed).toBe(true)
  245. expect(manager.get(S1)).toBe(session) // resident-instance rule survives removal
  246. })
  247. })
  248. describe('subagent catalogs', () => {
  249. it('keeps a catalog-discovered child address across ordinary selection and status frames', async () => {
  250. const api = new FakeApiClient()
  251. api.onList = () => Promise.resolve(ok({ items: [
  252. summary(S1),
  253. summary(S2, { parentSessionId: S1, origin: 'subagent' }),
  254. ] as never[] }))
  255. api.onSubagentList = () => Promise.resolve(ok({
  256. entries: [{
  257. kind: 'child', id: S2, mode: 'continuable', label: 'worker',
  258. activity: 'running', hasChildren: false,
  259. }] as never[],
  260. parentAvailable: true,
  261. }))
  262. const manager = new SessionManager(api, fakeRemote(api))
  263. await manager.refreshList()
  264. await manager.refreshSubagents(S1)
  265. manager.selectSubagent({ parentSessionId: S1, childSessionId: S2, mode: 'continuable' })
  266. expect(manager.getListSnapshot().currentAddress).toEqual({
  267. parentSessionId: S1, childSessionId: S2, mode: 'continuable',
  268. })
  269. expect(manager.get(S2).getSnapshot().subagent).toEqual({
  270. address: { parentSessionId: S1, childSessionId: S2, mode: 'continuable' },
  271. parentAvailable: true,
  272. })
  273. // Clicking the same child through an ordinary list-selection path must not
  274. // erase the catalog-derived address and fall back to session.* transport.
  275. manager.select(S2)
  276. expect(manager.getListSnapshot().currentAddress).toEqual({
  277. parentSessionId: S1, childSessionId: S2, mode: 'continuable',
  278. })
  279. expect(manager.get(S2).getSnapshot().subagent).toEqual({
  280. address: { parentSessionId: S1, childSessionId: S2, mode: 'continuable' },
  281. parentAvailable: true,
  282. })
  283. await manager.get(S2).open()
  284. await manager.get(S2).prompt([{ type: 'text', text: 'continue' }], 'queue')
  285. expect(api.callsOf('session.follow')).toEqual([
  286. {
  287. address: {
  288. kind: 'subagent', parentSessionId: S1, childSessionId: S2, mode: 'continuable',
  289. },
  290. maxMessages: 50,
  291. },
  292. ])
  293. expect(api.callsOf('subagent.history')).toEqual([])
  294. expect(api.callsOf('subagent.prompt')).toEqual([
  295. {
  296. parentSessionId: S1, childSessionId: S2, mode: 'continuable',
  297. content: [{ type: 'text', text: 'continue' }],
  298. clientTimeZone: new Intl.DateTimeFormat().resolvedOptions().timeZone,
  299. },
  300. ])
  301. expect(api.callsOf('session.history')).toEqual([])
  302. expect(api.callsOf('session.prompt')).toEqual([])
  303. const listCalls = api.callsOf('subagent.list').length
  304. manager.handleSessionStatus(S2, false)
  305. expect(manager.getListSnapshot().subagentsByParent[S1]?.entries[0]).toMatchObject({
  306. kind: 'child', id: S2, activity: 'inactive',
  307. })
  308. expect(api.callsOf('subagent.list')).toHaveLength(listCalls)
  309. manager.handleSessionRemoved(S2)
  310. expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)).toMatchObject({
  311. origin: 'subagent', parentSessionId: S1, running: false,
  312. })
  313. expect(manager.get(S2).getSnapshot()).toMatchObject({
  314. removed: false,
  315. subagent: {
  316. address: { parentSessionId: S1, childSessionId: S2, mode: 'continuable' },
  317. },
  318. })
  319. })
  320. it('refetches debounced membership only while the parent catalog is open', async () => {
  321. vi.useFakeTimers()
  322. try {
  323. const api = new FakeApiClient()
  324. const manager = new SessionManager(api, fakeRemote(api))
  325. await manager.refreshSubagents(S1)
  326. manager.setSubagentCatalogOpen(S1, true)
  327. await Promise.resolve()
  328. const baseline = api.callsOf('subagent.list').length
  329. manager.handleSessionAdded(summary(S2, { parentSessionId: S1 }))
  330. manager.handleSessionAdded(summary('fk-m3' as SessionId, { parentSessionId: S1 }))
  331. await vi.advanceTimersByTimeAsync(50)
  332. expect(api.callsOf('subagent.list')).toHaveLength(baseline + 1)
  333. manager.setSubagentCatalogOpen(S1, false)
  334. manager.handleSessionAdded(summary('fk-m4' as SessionId, { parentSessionId: S1 }))
  335. await vi.advanceTimersByTimeAsync(50)
  336. expect(api.callsOf('subagent.list')).toHaveLength(baseline + 1)
  337. } finally {
  338. vi.useRealTimers()
  339. }
  340. })
  341. it('marks a loaded parent row expandable only for a direct subagent publication', async () => {
  342. const api = new FakeApiClient()
  343. const root = 'fk-root' as SessionId
  344. api.onSubagentList = () => Promise.resolve(ok({
  345. entries: [
  346. {
  347. kind: 'child', id: S1, mode: 'continuable', label: 'parent',
  348. activity: 'inactive', hasChildren: false,
  349. },
  350. {
  351. kind: 'child', id: S2, mode: 'continuable', label: 'ordinary parent',
  352. activity: 'inactive', hasChildren: false,
  353. },
  354. ] as never[],
  355. parentAvailable: true,
  356. }))
  357. const manager = new SessionManager(api, fakeRemote(api))
  358. await manager.refreshSubagents(root)
  359. manager.handleSessionAdded(summary('fk-grandchild' as SessionId, {
  360. parentSessionId: S1, origin: 'subagent',
  361. }))
  362. manager.handleSessionAdded(summary('fk-fork' as SessionId, { parentSessionId: S2 }))
  363. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  364. { kind: 'child', id: S1, hasChildren: true },
  365. { kind: 'child', id: S2, hasChildren: false },
  366. ])
  367. })
  368. it('preserves a live expandability hint across only the older in-flight catalog response', async () => {
  369. const api = new FakeApiClient()
  370. const root = 'fk-root' as SessionId
  371. const response = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  372. api.onSubagentList = () => response.promise
  373. const manager = new SessionManager(api, fakeRemote(api))
  374. const refresh = manager.refreshSubagents(root)
  375. manager.handleSessionAdded(summary('fk-grandchild' as SessionId, {
  376. parentSessionId: S1, origin: 'subagent',
  377. }))
  378. response.resolve(ok({
  379. entries: [{
  380. kind: 'child', id: S1, mode: 'continuable', label: 'parent',
  381. activity: 'inactive', hasChildren: false,
  382. }] as never[],
  383. parentAvailable: true,
  384. }))
  385. await refresh
  386. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  387. { kind: 'child', id: S1, hasChildren: true },
  388. ])
  389. api.onSubagentList = () => Promise.resolve(ok({
  390. entries: [{
  391. kind: 'child', id: S1, mode: 'continuable', label: 'parent',
  392. activity: 'inactive', hasChildren: false,
  393. }] as never[],
  394. parentAvailable: true,
  395. }))
  396. await manager.refreshSubagents(root)
  397. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  398. { kind: 'child', id: S1, hasChildren: false },
  399. ])
  400. })
  401. it('replays status frames over an older in-flight catalog response', async () => {
  402. const api = new FakeApiClient()
  403. const root = 'fk-root' as SessionId
  404. const response = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  405. api.onSubagentList = () => response.promise
  406. const manager = new SessionManager(api, fakeRemote(api))
  407. const refresh = manager.refreshSubagents(root)
  408. manager.handleSessionStatus(S1, false)
  409. manager.handleSessionStatus(S2, true)
  410. response.resolve(ok({
  411. entries: [
  412. {
  413. kind: 'child', id: S1, mode: 'continuable', label: 'stopped',
  414. activity: 'running', hasChildren: false,
  415. },
  416. {
  417. kind: 'child', id: S2, mode: 'continuable', label: 'started',
  418. activity: 'inactive', hasChildren: false,
  419. },
  420. ] as never[],
  421. parentAvailable: true,
  422. }))
  423. await refresh
  424. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  425. { kind: 'child', id: S1, activity: 'inactive' },
  426. { kind: 'child', id: S2, activity: 'running' },
  427. ])
  428. })
  429. it('marks a detached catalog child inactive without requiring a selected address', async () => {
  430. const api = new FakeApiClient()
  431. api.onSubagentList = () => Promise.resolve(ok({
  432. entries: [{
  433. kind: 'child', id: S2, mode: 'continuable', label: 'worker',
  434. activity: 'running', hasChildren: false,
  435. }] as never[],
  436. parentAvailable: true,
  437. }))
  438. const manager = new SessionManager(api, fakeRemote(api))
  439. await manager.refreshSubagents(S1)
  440. manager.handleSessionRemoved(S2)
  441. expect(manager.getListSnapshot().subagentsByParent[S1]?.entries).toMatchObject([
  442. { kind: 'child', id: S2, activity: 'inactive' },
  443. ])
  444. })
  445. it('coalesces overlapping catalog reads without scheduling a trailing pull', async () => {
  446. const api = new FakeApiClient()
  447. const root = 'fk-root' as SessionId
  448. const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  449. api.onSubagentList = () => first.promise
  450. const manager = new SessionManager(api, fakeRemote(api))
  451. const refresh = manager.refreshSubagents(root)
  452. expect(manager.refreshSubagents(root)).toBe(refresh)
  453. api.onSubagentList = () => Promise.resolve(ok({ entries: [], parentAvailable: true }))
  454. first.resolve(ok({ entries: [], parentAvailable: true }))
  455. await refresh
  456. expect(api.callsOf('subagent.list')).toHaveLength(1)
  457. })
  458. it('runs one trailing catalog refresh for a membership change coalesced into an in-flight pull', async () => {
  459. vi.useFakeTimers()
  460. try {
  461. const api = new FakeApiClient()
  462. const root = 'fk-root' as SessionId
  463. const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  464. const second = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  465. api.onSubagentList = () => first.promise
  466. const manager = new SessionManager(api, fakeRemote(api), root)
  467. const refresh = manager.refreshSubagents(root)
  468. manager.setSubagentCatalogOpen(root, true)
  469. // A membership frame arrives while the pull is in flight; the debounced
  470. // refresh it schedules fires 50ms later and is coalesced into the pull —
  471. // which was requested before the new child existed. The stale mark must
  472. // queue one trailing pull carrying the change.
  473. manager.handleSessionAdded(summary(S2, { parentSessionId: root }))
  474. await vi.advanceTimersByTimeAsync(50)
  475. api.onSubagentList = () => second.promise
  476. first.resolve(ok({
  477. entries: [{
  478. kind: 'child', id: S1, mode: 'continuable', label: 'older',
  479. activity: 'inactive', hasChildren: false,
  480. }] as never[],
  481. parentAvailable: true,
  482. }))
  483. await refresh
  484. // The trailing pull is already in flight (kicked synchronously in finally).
  485. second.resolve(ok({
  486. entries: [
  487. {
  488. kind: 'child', id: S1, mode: 'continuable', label: 'older',
  489. activity: 'inactive', hasChildren: false,
  490. },
  491. {
  492. kind: 'child', id: S2, mode: 'continuable', label: 'new child',
  493. activity: 'inactive', hasChildren: false,
  494. },
  495. ] as never[],
  496. parentAvailable: true,
  497. }))
  498. await second.promise
  499. expect(api.callsOf('subagent.list')).toHaveLength(2)
  500. expect(manager.getListSnapshot().subagentsByParent[root]?.entries).toMatchObject([
  501. { kind: 'child', id: S1, label: 'older' },
  502. { kind: 'child', id: S2, label: 'new child' },
  503. ])
  504. } finally {
  505. vi.useRealTimers()
  506. }
  507. })
  508. it('keeps removal invalidation across a stale success and failed trailing pull', async () => {
  509. const api = new FakeApiClient()
  510. const root = 'fk-root' as SessionId
  511. const child = () => ({
  512. kind: 'child' as const, id: S2, mode: 'continuable' as const, label: 'worker',
  513. activity: 'inactive' as const, hasChildren: false,
  514. })
  515. const first = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  516. api.onSubagentList = () => first.promise
  517. const manager = new SessionManager(api, fakeRemote(api))
  518. const refresh = manager.refreshSubagents(root)
  519. first.resolve(ok({ entries: [child()] as never[], parentAvailable: true }))
  520. await refresh
  521. manager.selectSubagent({ parentSessionId: root, childSessionId: S2, mode: 'continuable' })
  522. // The removal lands while a second pull is in flight: the invalidation
  523. // must survive the pre-removal ok response, so one trailing pull runs.
  524. const mid = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  525. api.onSubagentList = () => mid.promise
  526. const midRefresh = manager.refreshSubagents(root)
  527. manager.handleSessionRemoved(root)
  528. const trailing = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  529. api.onSubagentList = () => trailing.promise
  530. mid.resolve(ok({ entries: [child()] as never[], parentAvailable: true }))
  531. await midRefresh
  532. expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
  533. expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
  534. trailing.resolve(err({ code: 'internal', message: 'trailing pull failed', details: {} }))
  535. await vi.waitFor(() => {
  536. expect(manager.getListSnapshot().subagentsByParent[root]).toMatchObject({
  537. state: 'error',
  538. parentAvailable: false,
  539. })
  540. })
  541. const rootCalls = api.callsOf('subagent.list')
  542. .filter(call => (call as { parentSessionId: SessionId }).parentSessionId === root)
  543. expect(rootCalls).toHaveLength(3)
  544. expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
  545. expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
  546. })
  547. it('invalidates catalog availability when the owning parent is removed', async () => {
  548. const api = new FakeApiClient()
  549. const root = 'fk-root' as SessionId
  550. api.onSubagentList = () => Promise.resolve(ok({
  551. entries: [{
  552. kind: 'child', id: S2, mode: 'continuable', label: 'worker',
  553. activity: 'inactive', hasChildren: false,
  554. }] as never[],
  555. parentAvailable: true,
  556. }))
  557. const manager = new SessionManager(api, fakeRemote(api))
  558. await manager.refreshSubagents(root)
  559. manager.selectSubagent({ parentSessionId: root, childSessionId: S2, mode: 'continuable' })
  560. expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: true })
  561. manager.handleSessionRemoved(root)
  562. expect(manager.getListSnapshot().subagentsByParent[root]?.parentAvailable).toBe(false)
  563. expect(manager.get(S2).getSnapshot().subagent).toMatchObject({ parentAvailable: false })
  564. })
  565. })
  566. describe('remaining branches', () => {
  567. it('refreshList folds a transport throw into the error state', async () => {
  568. const api = new FakeApiClient()
  569. api.onList = () => Promise.reject(new Error('list wire down'))
  570. const manager = new SessionManager(api, fakeRemote(api))
  571. await manager.refreshList()
  572. expect(manager.getListSnapshot()).toMatchObject({ state: 'error', error: { code: 'internal', message: 'list wire down' } })
  573. })
  574. it('refreshList pushes running bits down to already-instantiated sessions', async () => {
  575. const api = new FakeApiClient()
  576. const manager = new SessionManager(api, fakeRemote(api))
  577. const session = manager.get(S1)
  578. api.onList = () => Promise.resolve(ok({ items: [summary(S1, { running: true })] as never[] }))
  579. await manager.refreshList()
  580. expect(session.getSnapshot().running).toBe(true)
  581. })
  582. it('create passes cwd and a preallocated id, folds transport throws, and deduplicates the echo', async () => {
  583. const api = new FakeApiClient()
  584. api.onCreate = () => Promise.resolve(ok({ sessionId: S1 }))
  585. const manager = new SessionManager(api, fakeRemote(api))
  586. await manager.create({ cwd: '/tmp/w', sessionId: S1 })
  587. expect(api.callsOf('session.create')).toEqual([{ cwd: '/tmp/w', sessionId: S1 }])
  588. expect(manager.getListSnapshot().items[0]).toMatchObject({ sessionId: S1, cwd: '/tmp/w' })
  589. await manager.create({ cwd: '/tmp/w' }) // same id returned: no duplicate row
  590. expect(manager.getListSnapshot().items).toHaveLength(1)
  591. api.onCreate = () => Promise.reject(new Error('create wire down'))
  592. expect(await manager.create()).toMatchObject({ ok: false, error: { code: 'internal' } })
  593. // Business error passes through untouched.
  594. api.onCreate = () => Promise.resolve(err({ code: 'internal', message: 'no', details: {} }))
  595. expect(await manager.create()).toMatchObject({ ok: false })
  596. })
  597. it('publishes a real Ungrouped summary from workspace-attach-failed', async () => {
  598. const api = new FakeApiClient()
  599. api.onCreate = () => Promise.resolve(err({
  600. code: 'workspace-attach-failed',
  601. message: 'published but unattached',
  602. details: { sessionId: S1, workspaceId: 'w1' },
  603. } as never))
  604. const manager = new SessionManager(api, fakeRemote(api))
  605. const result = await manager.create({ workspaceId: 'w1' as never, sessionId: S1 })
  606. expect(result).toMatchObject({ ok: false, error: { code: 'workspace-attach-failed' } })
  607. expect(manager.getListSnapshot().items).toEqual([expect.objectContaining({ sessionId: S1 })])
  608. expect(manager.getListSnapshot().items[0]).not.toHaveProperty('cwd')
  609. })
  610. it('reconciles a fork child published before workspace attachment fails', async () => {
  611. const api = new FakeApiClient()
  612. api.onFork = () => Promise.resolve(err({
  613. code: 'workspace-attach-failed',
  614. message: 'forked but unattached',
  615. details: { sessionId: S2, workspaceId: 'w1' },
  616. } as never))
  617. const manager = new SessionManager(api, fakeRemote(api))
  618. const result = await manager.fork({ sessionId: S1 })
  619. expect(result).toMatchObject({ ok: false, error: { code: 'workspace-attach-failed' } })
  620. expect(manager.getListSnapshot().items).toEqual([expect.objectContaining({
  621. sessionId: S2,
  622. parentSessionId: S1,
  623. blank: false,
  624. })])
  625. })
  626. it('reconciles a preallocated id after an ordinary transport failure', async () => {
  627. const api = new FakeApiClient()
  628. api.onCreate = () => Promise.reject(new Error('response lost'))
  629. const manager = new SessionManager(api, fakeRemote(api))
  630. const failed = await manager.create({ workspaceId: 'w1' as never, sessionId: S1 })
  631. expect(failed).toMatchObject({ ok: false, error: { message: 'response lost' } })
  632. expect(manager.getListSnapshot().items).toEqual([])
  633. manager.handleSessionAdded(summary(S1, { blank: true, cwd: '/w/one' }))
  634. expect(manager.getListSnapshot().items).toEqual([
  635. expect.objectContaining({ sessionId: S1, cwd: '/w/one' }),
  636. ])
  637. manager.handleSessionAdded(summary(S1, { blank: true, cwd: '/w/one' }))
  638. expect(manager.getListSnapshot().items).toHaveLength(1)
  639. })
  640. it('subscribe notifies on list changes and stops after unsubscribe', async () => {
  641. const api = new FakeApiClient()
  642. const manager = new SessionManager(api, fakeRemote(api))
  643. let notified = 0
  644. const unsubscribe = manager.subscribe(() => { notified++ })
  645. await manager.refreshList()
  646. await new Promise(resolve => setTimeout(resolve, 0))
  647. expect(notified).toBeGreaterThan(0)
  648. const seen = notified
  649. unsubscribe()
  650. manager.handleSessionAdded(summary(S1, { blank: true }))
  651. await new Promise(resolve => setTimeout(resolve, 0))
  652. expect(notified).toBe(seen)
  653. })
  654. it('ignores Host status and error events for sessions without an instance', () => {
  655. const api = new FakeApiClient()
  656. const manager = new SessionManager(api, fakeRemote(api))
  657. manager.handleSessionStatus(S2, true)
  658. manager.handleSessionError(S2, '无实例')
  659. })
  660. it('keeps list-entry identity for unchanged rows across an unrelated list change', async () => {
  661. const api = new FakeApiClient()
  662. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
  663. const manager = new SessionManager(api, fakeRemote(api))
  664. await manager.refreshList()
  665. const before = manager.getListSnapshot()
  666. manager.handleSessionStatus(S2, true)
  667. const after = manager.getListSnapshot()
  668. expect(after.items).not.toBe(before.items)
  669. const beforeS1 = before.items.find(e => e.sessionId === S1)
  670. const afterS1 = after.items.find(e => e.sessionId === S1)
  671. expect(afterS1).toBe(beforeS1) // untouched entry keeps identity (entryCache)
  672. // Same-order same-entries snapshot reuses the items array.
  673. manager.handleSessionError(S1, 'x')
  674. expect(manager.getListSnapshot().items).toBe(after.items)
  675. })
  676. it('carries parentSessionId from the added event into the lineage row', () => {
  677. const api = new FakeApiClient()
  678. const manager = new SessionManager(api, fakeRemote(api))
  679. manager.handleSessionAdded(summary(S1, { blank: true }))
  680. manager.handleSessionAdded(summary(S2, {
  681. blank: true, parentSessionId: S1, origin: 'subagent',
  682. }))
  683. const items = manager.getListSnapshot().items
  684. expect(items.find(e => e.sessionId === S2)).toMatchObject({
  685. parentSessionId: S1, origin: 'subagent', depth: 1,
  686. })
  687. })
  688. })
  689. describe('connected generation', () => {
  690. it('refreshes query baselines without rebuilding independently resumed Session sources', async () => {
  691. const api = new FakeApiClient()
  692. api.onHistory = () => Promise.resolve(ok({
  693. records: entries(plainTurn(0, 0, 'a', 'b')) as never[],
  694. hasMore: false,
  695. modelSelection: { provider: 'deepseek-official', model: 'deepseek-chat' },
  696. }))
  697. const manager = new SessionManager(api, fakeRemote(api))
  698. const openedSession = manager.get(S1)
  699. await openedSession.open()
  700. manager.get(S2) // instantiated but never opened
  701. const historyCallsBefore = api.callsOf('session.history').length
  702. manager.handleConnected()
  703. await vi.waitFor(() => {
  704. expect(api.callsOf('session.list').length).toBe(1)
  705. })
  706. expect(api.callsOf('session.history')).toHaveLength(historyCallsBefore)
  707. })
  708. it('retains the durable parent address and refreshes its catalogs across reconnect', async () => {
  709. const api = new FakeApiClient()
  710. const address = {
  711. parentSessionId: S1, childSessionId: S2, mode: 'continuable' as const,
  712. }
  713. const parent = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  714. const child = deferred<Awaited<ReturnType<FakeApiClient['onSubagentList']>>>()
  715. api.onSubagentList = payload => (
  716. (payload as { parentSessionId: SessionId }).parentSessionId === S1
  717. ? parent.promise
  718. : child.promise
  719. )
  720. const manager = new SessionManager(api, fakeRemote(api), S2, address)
  721. manager.handleConnected()
  722. expect(manager.get(S2).getSnapshot().subagent).toEqual({ address })
  723. parent.resolve(ok({ entries: [], parentAvailable: true }))
  724. child.resolve(ok({ entries: [], parentAvailable: true }))
  725. await vi.waitFor(() => {
  726. expect(api.callsOf('session.list')).toHaveLength(1)
  727. })
  728. await vi.waitFor(() => {
  729. expect(api.callsOf('subagent.list')).toEqual([
  730. { parentSessionId: S1 },
  731. { parentSessionId: S2 },
  732. ])
  733. })
  734. expect(manager.get(S2).getSnapshot().subagent).toEqual({
  735. address,
  736. parentAvailable: true,
  737. })
  738. expect(manager.getListSnapshot().currentAddress).toEqual(address)
  739. })
  740. })
  741. describe('completed reminder', () => {
  742. const status = (manager: SessionManager, sessionId: SessionId, running: boolean): void => {
  743. manager.handleSessionStatus(sessionId, running)
  744. }
  745. const added = (manager: SessionManager, sessionId: SessionId): void => {
  746. manager.handleSessionAdded(summary(sessionId))
  747. }
  748. const entry = (manager: SessionManager, sessionId: SessionId) =>
  749. manager.getListSnapshot().items.find(item => item.sessionId === sessionId)
  750. it('arms on a running→idle flip of a non-selected session and clears on select', () => {
  751. const manager = makeManager()
  752. added(manager, S1)
  753. added(manager, S2)
  754. manager.select(S1)
  755. expect(entry(manager, S2)?.completed).toBe(false)
  756. status(manager, S2, true)
  757. status(manager, S2, false)
  758. expect(entry(manager, S2)?.completed).toBe(true)
  759. // Opening the session consumes the reminder.
  760. manager.select(S2)
  761. expect(entry(manager, S2)?.completed).toBe(false)
  762. })
  763. it('never arms for the session being watched and re-arms after a switch-away re-run', () => {
  764. const manager = makeManager()
  765. added(manager, S1)
  766. added(manager, S2)
  767. manager.select(S2)
  768. status(manager, S2, true)
  769. status(manager, S2, false)
  770. expect(entry(manager, S2)?.completed).toBe(false) // watched to completion: no reminder
  771. // Switch away; a fresh run completing again arms the reminder.
  772. manager.select(S1)
  773. status(manager, S2, true)
  774. status(manager, S2, false)
  775. expect(entry(manager, S2)?.completed).toBe(true)
  776. })
  777. it('a re-run disarms the reminder while running and re-arms on its completion', () => {
  778. const manager = makeManager()
  779. added(manager, S1)
  780. added(manager, S2)
  781. manager.select(S1)
  782. status(manager, S2, true)
  783. status(manager, S2, false)
  784. expect(entry(manager, S2)?.completed).toBe(true)
  785. // The user starts a new run without opening the session: running wins.
  786. status(manager, S2, true)
  787. expect(entry(manager, S2)?.completed).toBe(false)
  788. status(manager, S2, false)
  789. expect(entry(manager, S2)?.completed).toBe(true)
  790. })
  791. it('session-removed drops the reminder and a re-add starts clean', () => {
  792. const manager = makeManager()
  793. added(manager, S1)
  794. added(manager, S2)
  795. manager.select(S1)
  796. status(manager, S2, true)
  797. status(manager, S2, false)
  798. expect(entry(manager, S2)?.completed).toBe(true)
  799. manager.handleSessionRemoved(S2)
  800. expect(manager.getListSnapshot().items.find(item => item.sessionId === S2)).toBeUndefined()
  801. added(manager, S2)
  802. expect(entry(manager, S2)?.completed).toBe(false)
  803. })
  804. it('a list refresh carrying the running→idle transition arms the reminder', async () => {
  805. const api = new FakeApiClient()
  806. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200, running: true })] as never[] }))
  807. const manager = new SessionManager(api, fakeRemote(api))
  808. await manager.refreshList()
  809. manager.select(S1)
  810. expect(entry(manager, S2)?.completed).toBe(false)
  811. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200, running: false })] as never[] }))
  812. await manager.refreshList()
  813. expect(entry(manager, S2)?.completed).toBe(true)
  814. })
  815. it('never arms for sessions already idle at first observation', async () => {
  816. const api = new FakeApiClient()
  817. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
  818. const manager = new SessionManager(api, fakeRemote(api))
  819. await manager.refreshList()
  820. manager.select(S1)
  821. expect(entry(manager, S2)?.completed).toBe(false)
  822. api.onList = () => Promise.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 201 })] as never[] }))
  823. await manager.refreshList()
  824. expect(entry(manager, S2)?.completed).toBe(false)
  825. })
  826. it('arms a completion that happened during an in-flight first pull (baseline running, replayed idle)', async () => {
  827. const api = new FakeApiClient()
  828. const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
  829. api.onList = () => gate.promise
  830. const manager = new SessionManager(api, fakeRemote(api))
  831. const refresh = manager.refreshList()
  832. // The session finishes while the first pull is still in flight; the pull
  833. // response recorded it as running at pull time.
  834. status(manager, S2, false)
  835. gate.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200, running: true })] as never[] }))
  836. await refresh
  837. expect(entry(manager, S2)?.completed).toBe(true)
  838. })
  839. it('arms when a session ran and completed entirely between in-flight mutations (baseline idle)', async () => {
  840. const api = new FakeApiClient()
  841. const gate = deferred<Awaited<ReturnType<FakeApiClient['onList']>>>()
  842. api.onList = () => gate.promise
  843. const manager = new SessionManager(api, fakeRemote(api))
  844. const refresh = manager.refreshList()
  845. // The unknown session starts and finishes while the first pull is in
  846. // flight; the pull-time baseline recorded it idle, so the running→idle
  847. // edge lives entirely inside the replayed mutations.
  848. status(manager, S2, true)
  849. status(manager, S2, false)
  850. gate.resolve(ok({ items: [summary(S1), summary(S2, { updatedAt: 200 })] as never[] }))
  851. await refresh
  852. expect(entry(manager, S2)?.completed).toBe(true)
  853. })
  854. })
  855. describe('background-job mirror', () => {
  856. const view = (over: Partial<{ id: string; status: string; label: string }> = {}) => ({
  857. id: 'bash-1', kind: 'bash', label: 'pnpm run build', status: 'running', startedAt: 5, ...over,
  858. })
  859. const tasksFrame = (
  860. sessionId: SessionId,
  861. jobs: unknown[],
  862. ): Extract<SessionControlFrame, { type: 'jobs' }> => ({
  863. type: 'jobs', sessionId, jobs: jobs as never,
  864. })
  865. it('mirrors the whole set last-wins, keyed per session, with no Session instance needed', () => {
  866. const manager = makeManager()
  867. manager.handleControlFrame(tasksFrame(S1, [view()]))
  868. manager.handleControlFrame(tasksFrame(S2, [view({ id: 'pwsh-1', label: 'other' })]))
  869. const first = manager.getListSnapshot().jobsBySession
  870. expect(first[S1]).toEqual([view()])
  871. expect(first[S2]?.[0]?.label).toBe('other')
  872. // Last-wins: the newer whole set replaces, it does not merge.
  873. manager.handleControlFrame(tasksFrame(S1, [view({ status: 'completed' })]))
  874. expect(manager.getListSnapshot().jobsBySession[S1]).toEqual([view({ status: 'completed' })])
  875. })
  876. it('stores an emptied set as an absent key so absence and [] read alike', () => {
  877. const manager = makeManager()
  878. manager.handleControlFrame(tasksFrame(S1, [view()]))
  879. expect(S1 in manager.getListSnapshot().jobsBySession).toBe(true)
  880. manager.handleControlFrame(tasksFrame(S1, []))
  881. expect(S1 in manager.getListSnapshot().jobsBySession).toBe(false)
  882. })
  883. it('clears the mirror when the next control baseline has no jobs', () => {
  884. const manager = makeManager()
  885. manager.handleControlFrame(tasksFrame(S1, [view()]))
  886. manager.handleControlFrame({
  887. type: 'baseline',
  888. value: { queues: {}, jobs: {}, projections: {} },
  889. })
  890. expect(S1 in manager.getListSnapshot().jobsBySession).toBe(false)
  891. })
  892. it('drops the rows when the session is removed, whichever stream lands first', () => {
  893. const manager = makeManager()
  894. manager.handleSessionAdded(summary(S1, { blank: true }))
  895. manager.handleControlFrame(tasksFrame(S1, [view()]))
  896. manager.handleSessionRemoved(S1)
  897. expect(S1 in manager.getListSnapshot().jobsBySession).toBe(false)
  898. })
  899. it('notifies list subscribers so an open header re-renders without a poll', async () => {
  900. const manager = makeManager()
  901. const seen = vi.fn()
  902. manager.subscribe(seen)
  903. manager.handleControlFrame(tasksFrame(S1, [view()]))
  904. // The notifier batches on a microtask; the frame itself is already applied.
  905. await Promise.resolve()
  906. expect(seen).toHaveBeenCalled()
  907. })
  908. })