diff --git a/.changeset/swarm-roster-refresh.md b/.changeset/swarm-roster-refresh.md new file mode 100644 index 0000000000..41af827fd0 --- /dev/null +++ b/.changeset/swarm-roster-refresh.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Fix the AgentSwarm member list disappearing after a page refresh while subagents are still running. diff --git a/apps/kimi-web/src/api/daemon/client.ts b/apps/kimi-web/src/api/daemon/client.ts index 67cf1c4804..291eceb4c5 100644 --- a/apps/kimi-web/src/api/daemon/client.ts +++ b/apps/kimi-web/src/api/daemon/client.ts @@ -495,6 +495,8 @@ export class DaemonKimiWebApi implements KimiWebApi { }, pendingApprovals: data.pending_approvals.map(toAppApprovalRequest), pendingQuestions: data.pending_questions.map(toAppQuestionRequest), + // Older servers omit the roster entirely; treat as an empty roster. + subagents: (data.subagents ?? []).map(toAppTask), }; } diff --git a/apps/kimi-web/src/api/daemon/eventReducer.ts b/apps/kimi-web/src/api/daemon/eventReducer.ts index 970fbc925a..5d6de6a307 100644 --- a/apps/kimi-web/src/api/daemon/eventReducer.ts +++ b/apps/kimi-web/src/api/daemon/eventReducer.ts @@ -31,6 +31,12 @@ const OPTIMISTIC_USER_MESSAGE_METADATA_KEY = 'kimiWeb.optimisticUserMessage'; * in full (small synthesized lines). */ const MAX_BACKGROUND_OUTPUT_LINES = 40; +/** Skeleton description used by `patchSubagent` in agentEventProjector.ts when + * a lifecycle event re-projects a subagent the projector never saw spawn + * (e.g. after a page refresh, where the snapshot roster — not the WS stream — + * carried the real description). */ +const PLACEHOLDER_SUBAGENT_DESCRIPTION = 'Sub Agent'; + // --------------------------------------------------------------------------- // State // --------------------------------------------------------------------------- @@ -558,6 +564,14 @@ export function reduceAppEvent( ...event.task, outputLines: previous.outputLines, text: previous.text, + // A post-refresh lifecycle event re-projects the task with skeleton + // metadata; don't let its placeholder clobber the roster-seeded + // description. + description: + event.task.description === PLACEHOLDER_SUBAGENT_DESCRIPTION && + previous.description !== PLACEHOLDER_SUBAGENT_DESCRIPTION + ? previous.description + : event.task.description, swarmIndex: event.task.swarmIndex ?? previous.swarmIndex, parentToolCallId: event.task.parentToolCallId ?? previous.parentToolCallId, subagentType: event.task.subagentType ?? previous.subagentType, diff --git a/apps/kimi-web/src/api/daemon/mappers.ts b/apps/kimi-web/src/api/daemon/mappers.ts index 85b81421a1..a22ecf1afb 100644 --- a/apps/kimi-web/src/api/daemon/mappers.ts +++ b/apps/kimi-web/src/api/daemon/mappers.ts @@ -382,9 +382,11 @@ export function toAppTask(wire: WireTask): AppTask { parentToolCallId: wire.parent_tool_call_id, suspendedReason: wire.suspended_reason, swarmIndex: wire.swarm_index, - // The background task store only holds detached tasks, so any subagent it - // returns is a background subagent (foreground ones never persist here). - runInBackground: wire.kind === 'subagent' ? true : undefined, + // The snapshot's subagent roster carries the explicit flag. REST `/tasks` + // does not, but its background-task store only holds detached tasks, so any + // subagent it returns is a background subagent (foreground ones never + // persist there) — hence the `?? true` fallback for that path. + runInBackground: wire.run_in_background ?? (wire.kind === 'subagent' ? true : undefined), // outputLines starts undefined; populated by eventReducer via task.progress events }; } diff --git a/apps/kimi-web/src/api/daemon/wire.ts b/apps/kimi-web/src/api/daemon/wire.ts index e03f37a94a..fb5289ec12 100644 --- a/apps/kimi-web/src/api/daemon/wire.ts +++ b/apps/kimi-web/src/api/daemon/wire.ts @@ -309,6 +309,7 @@ export interface WireTask { parent_tool_call_id?: string; suspended_reason?: string; swarm_index?: number; + run_in_background?: boolean; } // --------------------------------------------------------------------------- @@ -561,6 +562,8 @@ export interface WireSessionSnapshot { session: WireSession; messages: { items: WireMessage[]; has_more: boolean }; in_flight_turn: WireInFlightTurn | null; + /** Live subagent roster at the watermark (absent on older servers). */ + subagents?: WireTask[]; pending_approvals: WireApprovalRequest[]; pending_questions: WireQuestionRequest[]; } diff --git a/apps/kimi-web/src/api/types.ts b/apps/kimi-web/src/api/types.ts index 8f0309c956..f24a1b8e00 100644 --- a/apps/kimi-web/src/api/types.ts +++ b/apps/kimi-web/src/api/types.ts @@ -498,6 +498,8 @@ export interface AppSessionSnapshot { messages: AppMessage[]; hasMoreMessages: boolean; inFlightTurn: AppInFlightTurn | null; + /** Live subagent roster at the watermark — rebuilds swarm cards on refresh. */ + subagents: AppTask[]; pendingApprovals: AppApprovalRequest[]; pendingQuestions: AppQuestionRequest[]; } diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index 03fb9d4f68..126b05a39e 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -15,6 +15,7 @@ import { } from '../lib/workspaceOrder'; import { mergeWorkspaces } from '../lib/mergeWorkspaces'; import { mergeSnapshotMessages } from '../lib/snapshotMessages'; +import { mergeSnapshotSubagents } from '../lib/taskMerge'; import { createCoalescedAsyncRunner } from '../lib/snapshotSync'; import { loadUnread, @@ -1275,6 +1276,17 @@ async function syncSessionFromSnapshot(sessionId: string): Promise t.kind === 'subagent' && !restIds.has(t.id)); return liveSubagents.length === 0 ? restBased : [...restBased, ...liveSubagents]; } + +/** + * Seed the task store from the snapshot's subagent roster. The roster is + * authoritative for identity/status/phase; keep reducer-owned accumulated + * output (outputLines/text) from any already-live task, and keep tasks the + * roster does not know about (background bash tasks from REST). + */ +export function mergeSnapshotSubagents(roster: AppTask[], existing: AppTask[]): AppTask[] { + if (roster.length === 0) return existing; + const existingById = new Map(existing.map((t) => [t.id, t] as const)); + const rosterIds = new Set(roster.map((t) => t.id)); + const merged = roster.map((task) => { + const live = existingById.get(task.id); + if (!live) return task; + return { ...task, outputLines: live.outputLines, text: live.text }; + }); + const kept = existing.filter((t) => !rosterIds.has(t.id)); + return kept.length === 0 ? merged : [...merged, ...kept]; +} diff --git a/apps/kimi-web/test/event-reducer.test.ts b/apps/kimi-web/test/event-reducer.test.ts index 860a927f09..3aa471c2aa 100644 --- a/apps/kimi-web/test/event-reducer.test.ts +++ b/apps/kimi-web/test/event-reducer.test.ts @@ -265,6 +265,47 @@ describe('reduceAppEvent taskProgress', () => { text: 'partial', }); }); + + it('keeps the roster-seeded description when a re-projected task carries the placeholder', () => { + // After a page refresh the snapshot roster seeds the real description; a + // later subagent.* lifecycle event re-projects the task with the + // projector's skeleton default ('Sub Agent') — it must not clobber it. + const state = { + ...createInitialState(), + tasksBySession: { + 's1': [{ ...makeSubagentTask('t1', 's1'), description: 'explore the auth flow' }], + }, + }; + const next = reduceAppEvent( + state, + { + type: 'taskCreated', + sessionId: 's1', + task: { ...makeSubagentTask('t1', 's1'), description: 'Sub Agent' }, + }, + { sessionId: 's1', seq: 1 }, + ); + expect(next.tasksBySession['s1']?.[0]?.description).toBe('explore the auth flow'); + }); + + it('takes the incoming description when it is a real one', () => { + const state = { + ...createInitialState(), + tasksBySession: { + 's1': [{ ...makeSubagentTask('t1', 's1'), description: 'Sub Agent' }], + }, + }; + const next = reduceAppEvent( + state, + { + type: 'taskCreated', + sessionId: 's1', + task: { ...makeSubagentTask('t1', 's1'), description: 'write the tests' }, + }, + { sessionId: 's1', seq: 1 }, + ); + expect(next.tasksBySession['s1']?.[0]?.description).toBe('write the tests'); + }); }); describe('reduceAppEvent sessions reference stability', () => { diff --git a/apps/kimi-web/test/lib-logic.test.ts b/apps/kimi-web/test/lib-logic.test.ts index ab2d517e52..1e7f3dc44e 100644 --- a/apps/kimi-web/test/lib-logic.test.ts +++ b/apps/kimi-web/test/lib-logic.test.ts @@ -9,6 +9,7 @@ import { buildDiffLines } from '../src/lib/diffLines'; import { buildEditDiffLines } from '../src/lib/toolDiff'; import { createCoalescedAsyncRunner } from '../src/lib/snapshotSync'; import { mergeSnapshotMessages } from '../src/lib/snapshotMessages'; +import { mergeSnapshotSubagents } from '../src/lib/taskMerge'; import { normalizeToolName, toolSummary } from '../src/lib/toolMeta'; import { collapsePrompt, humanizeCron } from '../src/lib/cronHumanize'; import { @@ -25,7 +26,7 @@ import { modelThinkingAvailability, segmentsFor, } from '../src/lib/modelThinking'; -import type { AppMessage, AppModel } from '../src/api/types'; +import type { AppMessage, AppModel, AppTask } from '../src/api/types'; import { resolveToolRenderer } from '../src/components/chat/tool-calls/toolRegistry'; import AgentTool from '../src/components/chat/tool-calls/AgentTool.vue'; import EditTool from '../src/components/chat/tool-calls/EditTool.vue'; @@ -534,3 +535,59 @@ describe('mergeSnapshotMessages', () => { expect(mergeSnapshotMessages(snapshot, [])).toEqual([]); }); }); + +describe('mergeSnapshotSubagents', () => { + function subagent(id: string, overrides: Partial = {}): AppTask { + return { + id, + sessionId: 's1', + kind: 'subagent', + description: `task ${id}`, + status: 'running', + createdAt: '2026-01-01T00:00:00.000Z', + ...overrides, + }; + } + + it('seeds an empty store from the roster', () => { + const roster = [ + subagent('a1', { subagentPhase: 'working', swarmIndex: 0, parentToolCallId: 'call-1' }), + subagent('a2', { subagentPhase: 'queued', swarmIndex: 1, parentToolCallId: 'call-1' }), + ]; + expect(mergeSnapshotSubagents(roster, [])).toEqual(roster); + }); + + it('keeps reducer-owned accumulated output from an already-live task', () => { + const live = subagent('a1', { + subagentPhase: 'queued', + outputLines: ['line 1'], + text: 'partial answer', + }); + const roster = [subagent('a1', { subagentPhase: 'working' })]; + const [merged] = mergeSnapshotSubagents(roster, [live]); + // Roster is authoritative for identity/status/phase… + expect(merged?.subagentPhase).toBe('working'); + // …but the accumulated output survives the seed. + expect(merged?.outputLines).toEqual(['line 1']); + expect(merged?.text).toBe('partial answer'); + }); + + it('keeps tasks the roster does not know about', () => { + const background: AppTask = { + id: 'bash-1', + sessionId: 's1', + kind: 'bash', + description: 'npm test', + status: 'running', + createdAt: '2026-01-01T00:00:00.000Z', + }; + const roster = [subagent('a1')]; + const merged = mergeSnapshotSubagents(roster, [background, subagent('a1')]); + expect(merged.map((t) => t.id)).toEqual(['a1', 'bash-1']); + }); + + it('returns the existing list untouched when the roster is empty', () => { + const existing = [subagent('a1')]; + expect(mergeSnapshotSubagents([], existing)).toBe(existing); + }); +}); diff --git a/packages/protocol/src/__tests__/snapshot.test.ts b/packages/protocol/src/__tests__/snapshot.test.ts index 017481c191..30e752aabd 100644 --- a/packages/protocol/src/__tests__/snapshot.test.ts +++ b/packages/protocol/src/__tests__/snapshot.test.ts @@ -106,6 +106,52 @@ describe('rest/snapshot — session snapshot', () => { } }); + it('parses a snapshot with a subagent roster', () => { + const result = sessionSnapshotResponseSchema.safeParse({ + as_of_seq: 12, + epoch: 'ep_01ABC', + session: SESSION, + messages: { items: [], has_more: false }, + in_flight_turn: null, + subagents: [ + { + id: 'agent_1', + session_id: 'sess_1', + kind: 'subagent', + description: 'explore the auth flow', + status: 'running', + created_at: TS, + started_at: TS, + subagent_phase: 'working', + subagent_type: 'explore', + parent_tool_call_id: 'call_1', + swarm_index: 0, + run_in_background: false, + }, + { + id: 'agent_2', + session_id: 'sess_1', + kind: 'subagent', + description: 'write tests', + status: 'completed', + created_at: TS, + completed_at: TS, + output_preview: 'done', + subagent_phase: 'completed', + swarm_index: 1, + }, + ], + pending_approvals: [], + pending_questions: [], + }); + expect(result.success).toBe(true); + if (result.success) { + expect(result.data.subagents).toHaveLength(2); + expect(result.data.subagents?.[0]?.parent_tool_call_id).toBe('call_1'); + expect(result.data.subagents?.[1]?.subagent_phase).toBe('completed'); + } + }); + it('rejects a snapshot missing the watermark', () => { const result = sessionSnapshotResponseSchema.safeParse({ epoch: 'ep_01ABC', diff --git a/packages/protocol/src/rest/snapshot.ts b/packages/protocol/src/rest/snapshot.ts index 642772b695..618193834c 100644 --- a/packages/protocol/src/rest/snapshot.ts +++ b/packages/protocol/src/rest/snapshot.ts @@ -27,6 +27,7 @@ import { approvalRequestSchema } from '../approval'; import { messageSchema } from '../message'; import { questionRequestSchema } from '../question'; import { sessionSchema } from '../session'; +import { taskSchema } from '../task'; export const inFlightToolCallSchema = z.object({ tool_call_id: z.string().min(1), @@ -59,6 +60,21 @@ export const inFlightTurnSchema = z.object({ }); export type InFlightTurn = z.infer; +/** + * A live subagent task as of the snapshot watermark. Extends the base task + * wire shape with the swarm identity metadata that otherwise only rides the + * (non-replayed) `subagent.spawned` WS event. + */ +export const snapshotSubagentSchema = taskSchema.extend({ + subagent_phase: z.enum(['queued', 'working', 'suspended', 'completed', 'failed']).optional(), + subagent_type: z.string().optional(), + parent_tool_call_id: z.string().optional(), + suspended_reason: z.string().optional(), + swarm_index: z.number().int().nonnegative().optional(), + run_in_background: z.boolean().optional(), +}); +export type SnapshotSubagent = z.infer; + export const sessionSnapshotResponseSchema = z.object({ /** Durable event watermark this snapshot is consistent with. */ as_of_seq: z.number().int().nonnegative(), @@ -71,6 +87,12 @@ export const sessionSnapshotResponseSchema = z.object({ has_more: z.boolean(), }), in_flight_turn: inFlightTurnSchema.nullable(), + /** + * Roster of live subagent tasks at the watermark, so a reconnecting client + * can rebuild swarm cards before the swarm's tool result lands. Optional + * for cross-version tolerance: older servers do not send it. + */ + subagents: z.array(snapshotSubagentSchema).optional(), pending_approvals: z.array(approvalRequestSchema), pending_questions: z.array(questionRequestSchema), }); diff --git a/packages/server/AGENTS.md b/packages/server/AGENTS.md index bd72782bde..99b03a8740 100644 --- a/packages/server/AGENTS.md +++ b/packages/server/AGENTS.md @@ -16,7 +16,7 @@ The Kimi Code server. It hosts `agent-core` sessions and exposes them over REST - Top level: `start.ts`, `index.ts`, `envelope.ts`, `error-handler.ts`, `lock.ts`, `request-id.ts`, `version.ts`. - `routes/` — REST domain modules, the `registerApiV1Routes.ts` aggregator, `webAssets.ts`, and `action-suffix.ts`. -- `services/` — server-owned DI adapters: `approval/`, `question/`, `gateway/` (`rest`/`ws`/`broadcast`/`connectionRegistry`/`sessionClients`/`sessionEventJournal`/`inFlightTurnTracker`), `pinoLoggerService.ts`, `serviceCollection.ts`. +- `services/` — server-owned DI adapters: `approval/`, `question/`, `gateway/` (`rest`/`ws`/`broadcast`/`connectionRegistry`/`sessionClients`/`sessionEventJournal`/`inFlightTurnTracker`/`subagentRosterTracker`), `pinoLoggerService.ts`, `serviceCollection.ts`. - `ws/` — `connection.ts` (`WsConnection`), `protocol.ts` (frame builders), `rawData.ts`. - `middleware/` — `defineRoute.ts`, `schema.ts`, `validate.ts`. `openapi/transforms.ts`. - `svc/` — OS service managers (launchd / systemd / schtasks) backing `kimi server install/start`. diff --git a/packages/server/src/routes/snapshot.ts b/packages/server/src/routes/snapshot.ts index 7172c64b91..9db6d4c422 100644 --- a/packages/server/src/routes/snapshot.ts +++ b/packages/server/src/routes/snapshot.ts @@ -169,6 +169,7 @@ async function readViaLegacyAssembly(ix: IInstantiationService, sid: string) { session: session!, messages: { items, has_more: hasMore }, in_flight_turn: inFlightTurn, + subagents: snapState.subagents, pending_approvals: approvals.listPending(sid), pending_questions: questions.listPending(sid), }; diff --git a/packages/server/src/services/gateway/subagentRosterTracker.ts b/packages/server/src/services/gateway/subagentRosterTracker.ts new file mode 100644 index 0000000000..76f774fbca --- /dev/null +++ b/packages/server/src/services/gateway/subagentRosterTracker.ts @@ -0,0 +1,113 @@ +/** + * `SubagentRosterTracker` — accumulates the per-session roster of live + * subagent tasks so a reconnecting client can rebuild swarm cards from the + * session snapshot. The refresh flow subscribes at the snapshot watermark, so + * earlier `subagent.spawned` events — the only carriers of the swarm identity + * metadata — are never replayed to it. + * + * Without this roster a mid-swarm page refresh loses the swarm card's member + * list: REST `/tasks` only serves the main agent's background-task store + * (foreground swarm subagents never persist there), and later `subagent.*` + * events carry only the `subagentId`, so the identity metadata + * (`parentToolCallId` / `swarmIndex` / `description`) is unrecoverable until + * the swarm's `` tool output lands. + * + * Owned by `WSBroadcastService` and updated INSIDE its per-session dispatch + * queue — same pattern as `InFlightTurnTracker`, keeping the roster, the + * journal watermark, and the fan-out order mutually consistent. + * + * Lifetime: the roster is dropped on `turn.ended`. After turn end the swarm's + * `` tool output is in the wire transcript and becomes + * the restore source; this also bounds the roster's lifetime (background + * subagents that outlive a turn are a known, pre-existing bound — same + * trade-off as `InFlightTurnTracker`). + */ + +import type { Event, SnapshotSubagent } from '@moonshot-ai/protocol'; + +export class SubagentRosterTracker { + private readonly bySession = new Map>(); + + apply(sessionId: string, event: Event): void { + switch (event.type) { + case 'subagent.spawned': { + let roster = this.bySession.get(sessionId); + if (!roster) { + roster = new Map(); + this.bySession.set(sessionId, roster); + } + roster.set(event.subagentId, { + id: event.subagentId, + session_id: sessionId, + kind: 'subagent', + description: event.description ?? event.subagentName ?? 'Sub Agent', + status: 'running', + subagent_phase: 'queued', + ...(event.subagentName !== undefined ? { subagent_type: event.subagentName } : {}), + ...(event.parentToolCallId !== undefined + ? { parent_tool_call_id: event.parentToolCallId } + : {}), + ...(event.swarmIndex !== undefined ? { swarm_index: event.swarmIndex } : {}), + run_in_background: event.runInBackground, + created_at: new Date().toISOString(), + }); + return; + } + case 'subagent.started': { + const entry = this.bySession.get(sessionId)?.get(event.subagentId); + if (!entry) return; + entry.subagent_phase = 'working'; + entry.suspended_reason = undefined; + // Keep an existing started_at: a resumed (previously suspended) + // subagent re-fires `subagent.started`. + entry.started_at ??= new Date().toISOString(); + return; + } + case 'subagent.suspended': { + const entry = this.bySession.get(sessionId)?.get(event.subagentId); + if (!entry) return; + entry.subagent_phase = 'suspended'; + entry.suspended_reason = event.reason; + return; + } + case 'subagent.completed': { + const entry = this.bySession.get(sessionId)?.get(event.subagentId); + if (!entry) return; + entry.subagent_phase = 'completed'; + entry.status = 'completed'; + entry.completed_at = new Date().toISOString(); + entry.output_preview = event.resultSummary; + return; + } + case 'subagent.failed': { + const entry = this.bySession.get(sessionId)?.get(event.subagentId); + if (!entry) return; + entry.subagent_phase = 'failed'; + entry.status = 'failed'; + entry.completed_at = new Date().toISOString(); + entry.output_preview = event.error; + return; + } + case 'turn.ended': { + // After turn end the swarm's `` tool output is in + // the wire transcript and becomes the restore source; dropping the + // roster here also bounds its lifetime. + this.bySession.delete(sessionId); + return; + } + default: + return; + } + } + + /** Fresh copies — callers must not mutate the tracked entries. */ + get(sessionId: string): SnapshotSubagent[] { + const roster = this.bySession.get(sessionId); + if (!roster) return []; + return Array.from(roster.values(), (entry) => ({ ...entry })); + } + + clear(sessionId: string): void { + this.bySession.delete(sessionId); + } +} diff --git a/packages/server/src/services/gateway/wsBroadcast.ts b/packages/server/src/services/gateway/wsBroadcast.ts index 759401cce0..0f23f817a3 100644 --- a/packages/server/src/services/gateway/wsBroadcast.ts +++ b/packages/server/src/services/gateway/wsBroadcast.ts @@ -33,7 +33,7 @@ */ import { createDecorator } from '@moonshot-ai/agent-core'; -import type { InFlightTurn, SessionCursor } from '@moonshot-ai/protocol'; +import type { InFlightTurn, SessionCursor, SnapshotSubagent } from '@moonshot-ai/protocol'; import type { EventEnvelope } from '#/ws/protocol'; @@ -43,6 +43,8 @@ export interface SessionSnapshotState { seq: number; epoch: string; inFlightTurn: InFlightTurn | null; + /** Live subagent roster at the watermark (see `SubagentRosterTracker`). */ + subagents: SnapshotSubagent[]; } export interface BufferedSinceResult { diff --git a/packages/server/src/services/gateway/wsBroadcastService.ts b/packages/server/src/services/gateway/wsBroadcastService.ts index d488a35a66..5d5953fdb9 100644 --- a/packages/server/src/services/gateway/wsBroadcastService.ts +++ b/packages/server/src/services/gateway/wsBroadcastService.ts @@ -8,6 +8,7 @@ import { IConnectionRegistry } from './connectionRegistry'; import { InFlightTurnTracker } from './inFlightTurnTracker'; import { ISessionClientsService } from './sessionClients'; import { SessionEventJournal } from './sessionEventJournal'; +import { SubagentRosterTracker } from './subagentRosterTracker'; import { DEFAULT_MAX_BUFFER_SIZE, IWSBroadcastService, @@ -40,6 +41,7 @@ export class WSBroadcastService extends Disposable implements IWSBroadcastServic private readonly _maxBufferSize: number; private readonly _journalDir: string; private readonly _turnTracker = new InFlightTurnTracker(); + private readonly _rosterTracker = new SubagentRosterTracker(); constructor( @IEventService eventService: IEventService, @@ -87,6 +89,9 @@ export class WSBroadcastService extends Disposable implements IWSBroadcastServic // text, the journal watermark, and fan-out order stay consistent. For // text deltas this also yields the pre-append offset for the envelope. const annotation = this._turnTracker.apply(sid, event); + // Same queue-discipline for the subagent roster: snapshot rebuilds must + // see exactly the roster as of the durable watermark. + this._rosterTracker.apply(sid, event); let envelope: EventEnvelope; if (isVolatileEventType(evType)) { @@ -172,6 +177,7 @@ export class WSBroadcastService extends Disposable implements IWSBroadcastServic seq: journal.seq, epoch: journal.epoch, inFlightTurn: this._turnTracker.get(sid), + subagents: this._rosterTracker.get(sid), }; } diff --git a/packages/server/src/services/snapshot/snapshotService.ts b/packages/server/src/services/snapshot/snapshotService.ts index db794e4261..254e626357 100644 --- a/packages/server/src/services/snapshot/snapshotService.ts +++ b/packages/server/src/services/snapshot/snapshotService.ts @@ -159,6 +159,7 @@ export class SnapshotService extends Disposable implements ISnapshotService { session, messages: { items: sliced, has_more: hasMore }, in_flight_turn: inFlightTurn, + subagents: snapState.subagents, pending_approvals: [...approvals], pending_questions: [...questions], }; diff --git a/packages/server/test/services.test.ts b/packages/server/test/services.test.ts index 186338697e..6f1af78f44 100644 --- a/packages/server/test/services.test.ts +++ b/packages/server/test/services.test.ts @@ -16,6 +16,7 @@ import { ISessionClientsService, type ISessionClientsService as ISessionClientsServiceT, } from '#/services/gateway'; +import { SubagentRosterTracker } from '#/services/gateway/subagentRosterTracker'; import { WSBroadcastService } from '#/services/gateway/wsBroadcastService'; import type { WsConnection } from '../src/ws/connection'; @@ -454,6 +455,161 @@ describe('WSBroadcastService (WS transport pump)', () => { broadcast.dispose(); bus.dispose(); }); + + it('getSnapshotState exposes the live subagent roster until turn.ended', async () => { + const bus = new EventService(); + const broadcast = new WSBroadcastService(bus, testLogger, new FakeSessionClients(), new FakeConnectionRegistry(), makeEnv()); + bus.publish({ + type: 'subagent.spawned', + sessionId: 'sid_r', + agentId: 'main', + subagentId: 'agent_1', + subagentName: 'explore', + parentToolCallId: 'call_1', + description: 'explore the auth flow', + swarmIndex: 0, + runInBackground: false, + } as unknown as Event); + bus.publish({ + type: 'subagent.started', + sessionId: 'sid_r', + agentId: 'agent_1', + subagentId: 'agent_1', + } as unknown as Event); + + const mid = await broadcast.getSnapshotState('sid_r'); + expect(mid.subagents).toHaveLength(1); + expect(mid.subagents[0]).toMatchObject({ + id: 'agent_1', + session_id: 'sid_r', + kind: 'subagent', + description: 'explore the auth flow', + status: 'running', + subagent_phase: 'working', + subagent_type: 'explore', + parent_tool_call_id: 'call_1', + swarm_index: 0, + run_in_background: false, + }); + expect(mid.subagents[0]?.started_at).toBeDefined(); + + bus.publish({ + type: 'subagent.completed', + sessionId: 'sid_r', + agentId: 'agent_1', + subagentId: 'agent_1', + resultSummary: 'done', + } as unknown as Event); + const done = await broadcast.getSnapshotState('sid_r'); + expect(done.subagents[0]).toMatchObject({ + subagent_phase: 'completed', + status: 'completed', + output_preview: 'done', + }); + expect(done.subagents[0]?.completed_at).toBeDefined(); + + // turn.ended drops the roster: the swarm result tool output in the wire + // transcript becomes the restore source from then on. + bus.publish({ + type: 'turn.ended', + sessionId: 'sid_r', + agentId: 'main', + turnId: 1, + reason: 'completed', + } as unknown as Event); + const after = await broadcast.getSnapshotState('sid_r'); + expect(after.subagents).toEqual([]); + broadcast.dispose(); + bus.dispose(); + }); +}); + +describe('SubagentRosterTracker', () => { + const spawned = (overrides: Record = {}): Event => + ({ + type: 'subagent.spawned', + sessionId: 'sid', + agentId: 'main', + subagentId: 'agent_1', + subagentName: 'explore', + parentToolCallId: 'call_1', + runInBackground: false, + ...overrides, + }) as unknown as Event; + + it('ignores lifecycle events for unknown subagent ids', () => { + const tracker = new SubagentRosterTracker(); + tracker.apply('sid', { + type: 'subagent.completed', + subagentId: 'ghost', + resultSummary: 'x', + } as unknown as Event); + tracker.apply('sid', { type: 'subagent.started', subagentId: 'ghost' } as unknown as Event); + tracker.apply('sid', { + type: 'subagent.suspended', + subagentId: 'ghost', + reason: 'approval', + } as unknown as Event); + expect(tracker.get('sid')).toEqual([]); + }); + + it('returns fresh copies that do not alias the tracked entries', () => { + const tracker = new SubagentRosterTracker(); + tracker.apply('sid', spawned()); + const first = tracker.get('sid'); + first[0]!.status = 'failed'; + first.push({} as never); + const second = tracker.get('sid'); + expect(second).toHaveLength(1); + expect(second[0]?.status).toBe('running'); + }); + + it('tracks suspend and resume, keeping the original started_at', () => { + const tracker = new SubagentRosterTracker(); + tracker.apply('sid', spawned()); + tracker.apply('sid', { type: 'subagent.started', subagentId: 'agent_1' } as unknown as Event); + const startedAt = tracker.get('sid')[0]?.started_at; + expect(startedAt).toBeDefined(); + + tracker.apply('sid', { + type: 'subagent.suspended', + subagentId: 'agent_1', + reason: 'awaiting approval', + } as unknown as Event); + expect(tracker.get('sid')[0]).toMatchObject({ + subagent_phase: 'suspended', + suspended_reason: 'awaiting approval', + }); + + tracker.apply('sid', { type: 'subagent.started', subagentId: 'agent_1' } as unknown as Event); + const resumed = tracker.get('sid')[0]!; + expect(resumed.subagent_phase).toBe('working'); + expect(resumed.started_at).toBe(startedAt); + expect(resumed.suspended_reason).toBeUndefined(); + }); + + it('records failure with the error as output preview', () => { + const tracker = new SubagentRosterTracker(); + tracker.apply('sid', spawned()); + tracker.apply('sid', { + type: 'subagent.failed', + subagentId: 'agent_1', + error: 'boom', + } as unknown as Event); + expect(tracker.get('sid')[0]).toMatchObject({ + subagent_phase: 'failed', + status: 'failed', + output_preview: 'boom', + }); + }); + + it('clear drops the roster', () => { + const tracker = new SubagentRosterTracker(); + tracker.apply('sid', spawned()); + expect(tracker.get('sid')).toHaveLength(1); + tracker.clear('sid'); + expect(tracker.get('sid')).toEqual([]); + }); }); describe('FsWatcherService', () => { diff --git a/packages/server/test/snapshotService.unit.test.ts b/packages/server/test/snapshotService.unit.test.ts index be50822ebc..2c709f3395 100644 --- a/packages/server/test/snapshotService.unit.test.ts +++ b/packages/server/test/snapshotService.unit.test.ts @@ -64,6 +64,7 @@ class StubBroadcast implements IWSBroadcastServiceT { readonly _serviceBrand: undefined; snapshotCalls = 0; inFlight: SessionSnapshotState['inFlightTurn'] = null; + subagents: SessionSnapshotState['subagents'] = []; seq = 0; epoch = 'ep_test'; @@ -75,7 +76,12 @@ class StubBroadcast implements IWSBroadcastServiceT { } async getSnapshotState(): Promise { this.snapshotCalls++; - return { seq: this.seq, epoch: this.epoch, inFlightTurn: this.inFlight }; + return { + seq: this.seq, + epoch: this.epoch, + inFlightTurn: this.inFlight, + subagents: this.subagents, + }; } currentSeq(): number { return this.seq; @@ -375,6 +381,34 @@ describe('SnapshotService.read', () => { 'prompt_xyz', ); }); + + it('passes the broadcast subagent roster through to the response', async () => { + const f = await makeFixture(); + const sid = 'sess_roster'; + await f.store.create({ id: sid, workDir: f.workDir }); + + f.broadcast.subagents = [ + { + id: 'agent_1', + session_id: sid, + kind: 'subagent', + description: 'explore the auth flow', + status: 'running', + created_at: new Date().toISOString(), + subagent_phase: 'working', + swarm_index: 0, + run_in_background: false, + }, + ]; + + const snap = await f.service.read(sid); + expect(snap.subagents).toHaveLength(1); + expect(snap.subagents?.[0]).toMatchObject({ + id: 'agent_1', + subagent_phase: 'working', + swarm_index: 0, + }); + }); }); describe('SnapshotService status FSM (bus subscriber)', () => {