diff --git a/.changeset/stale-running-subagents.md b/.changeset/stale-running-subagents.md new file mode 100644 index 0000000000..8f88506457 --- /dev/null +++ b/.changeset/stale-running-subagents.md @@ -0,0 +1,5 @@ +--- +"@moonshot-ai/kimi-code": patch +--- + +web: Fix finished foreground subagents remaining stuck as running after turn boundaries or reconnects — reconcile live rows from the authoritative snapshot roster, and never let a late/replayed progress frame downgrade a terminal subagent row back to running. diff --git a/apps/kimi-web/src/api/daemon/agentEventProjector.ts b/apps/kimi-web/src/api/daemon/agentEventProjector.ts index d4ebbd09a0..365077371a 100644 --- a/apps/kimi-web/src/api/daemon/agentEventProjector.ts +++ b/apps/kimi-web/src/api/daemon/agentEventProjector.ts @@ -26,6 +26,7 @@ import type { AppTask, } from '../types'; import { i18n } from '../../i18n'; +import { isLiveForegroundSubagent } from '../../lib/taskMerge'; import { toolLabel, toolSummary } from '../../lib/toolMeta'; import { toAppMessageContent } from './mappers'; import type { WireMessageContent } from './wire'; @@ -60,6 +61,7 @@ const MAIN_AGENT_TRANSCRIPT_FRAMES = new Set([ 'prompt.aborted', 'error', ]); +const MAX_RETAINED_SUBAGENT_HISTORY = 256; // --------------------------------------------------------------------------- // Internal helpers @@ -127,6 +129,11 @@ interface SessionState { // Subagent lifecycle deltas after spawned only carry subagentId. Keep the // spawned metadata here so later updates can replace the full AppTask. subagentMeta: Map; + // Live foreground ids seen before reset, awaiting the next snapshot roster. + subagentResetCandidates: Set; + // Ids authoritatively absent from a present roster. Bare progress must not + // recreate them; an explicit spawn/start lifecycle event re-opens them. + subagentTombstones: Set; // Bubble cleared by turn.step.retrying, to be reused by the retried // step.started (same turn) instead of stacking a new bubble. @@ -151,6 +158,8 @@ function createSessionState(): SessionState { model: '', messages: [], subagentMeta: new Map(), + subagentResetCandidates: new Set(), + subagentTombstones: new Set(), retryReuseMsgId: undefined, }; } @@ -207,6 +216,7 @@ function patchSubagent( patch: Partial, ): AppTask | null { if (typeof subagentId !== 'string' || subagentId.length === 0) return null; + state.subagentResetCandidates.delete(subagentId); const prev = state.subagentMeta.get(subagentId) ?? { id: subagentId, sessionId, @@ -269,6 +279,30 @@ function toolArgSummary(name: string, args: unknown): string { return toolSummary(name, arg); } +/** Statuses a bare progress frame must never downgrade back to 'running'. A + * late/replayed subagent-scoped frame arriving after the lifecycle terminal + * would otherwise resurrect the row (and its elapsed timer) permanently — no + * later event corrects it. Explicit lifecycle events (`subagent.started` on a + * resume) still re-open the row through their own projector cases. */ +function isTerminalSubagentStatus(status: AppTask['status'] | undefined): boolean { + return status === 'completed' || status === 'failed' || status === 'cancelled'; +} + +function trimRetainedSubagentHistory(state: SessionState): void { + const terminalIds: string[] = []; + for (const [id, task] of state.subagentMeta) { + if (isTerminalSubagentStatus(task.status)) terminalIds.push(id); + } + for (let index = 0; index < terminalIds.length - MAX_RETAINED_SUBAGENT_HISTORY; index += 1) { + state.subagentMeta.delete(terminalIds[index]!); + } + while (state.subagentTombstones.size > MAX_RETAINED_SUBAGENT_HISTORY) { + const oldest = state.subagentTombstones.values().next().value; + if (oldest === undefined) break; + state.subagentTombstones.delete(oldest); + } +} + function projectSubagentProgress( state: SessionState, sessionId: string, @@ -280,7 +314,9 @@ function projectSubagentProgress( // Side-channel agents (e.g. BTW side chat) stream their own transcript via // agentDelta events; don't pollute the main task output with generic step // placeholders like "Started a step". - if (sideChannelAgents.has(subagentId) && rawType === 'turn.step.started') return []; + const isSideChannel = sideChannelAgents.has(subagentId); + if (isSideChannel && rawType === 'turn.step.started') return []; + if (!isSideChannel && state.subagentTombstones.has(subagentId)) return []; // The subagent's own streamed text: forward each delta as a `text`-kind // progress chunk so the reducer concatenates it into `AppTask.text`, letting @@ -295,13 +331,15 @@ function projectSubagentProgress( // taskProgress to existing tasks — without this, the deltas are dropped and // the live detail stays blank until a non-text frame recreates the task. const previous = state.subagentMeta.get(subagentId); - const task = patchSubagent(state, sessionId, subagentId, { - status: 'running', - subagentPhase: 'working', - startedAt: previous?.startedAt ?? new Date().toISOString(), - }); const out: AppEvent[] = []; - if (task) out.push({ type: 'taskCreated', sessionId, task }); + if (!isTerminalSubagentStatus(previous?.status)) { + const task = patchSubagent(state, sessionId, subagentId, { + status: 'running', + subagentPhase: 'working', + startedAt: previous?.startedAt ?? new Date().toISOString(), + }); + if (task) out.push({ type: 'taskCreated', sessionId, task }); + } out.push({ type: 'taskProgress', sessionId, @@ -316,13 +354,15 @@ function projectSubagentProgress( const text = subagentProgressText(rawType, payload); if (text === null || text.length === 0) return []; const previous = state.subagentMeta.get(subagentId); - const task = patchSubagent(state, sessionId, subagentId, { - status: 'running', - subagentPhase: 'working', - startedAt: previous?.startedAt ?? new Date().toISOString(), - }); const out: AppEvent[] = []; - if (task) out.push({ type: 'taskCreated', sessionId, task }); + if (!isTerminalSubagentStatus(previous?.status)) { + const task = patchSubagent(state, sessionId, subagentId, { + status: 'running', + subagentPhase: 'working', + startedAt: previous?.startedAt ?? new Date().toISOString(), + }); + if (task) out.push({ type: 'taskCreated', sessionId, task }); + } out.push({ type: 'taskProgress', sessionId, taskId: subagentId, outputChunk: text, stream: 'stdout' }); return out; } @@ -514,7 +554,21 @@ export interface AgentProjector { * snapshot's `session.status` is the authoritative value. */ seedInFlight(sessionId: string, turn: AppInFlightTurn): AppEvent[]; - /** Reset all per-session state (call on re-subscribe / resync). */ + /** + * Seed `subagentMeta` from a session snapshot's subagent roster (v2 sync). + * Call after `reset` / `seedInFlight`, or independently when only the roster + * is fresh enough to reconcile. This keeps the reducer's task rows and the + * projector's meta aligned: a roster-completed row stays terminal, and a + * present roster tombstones live foreground ids that it omits. `undefined` + * preserves the pending decision for a future authoritative roster. + */ + seedSubagents(sessionId: string, roster: AppTask[] | undefined): void; + /** + * Reset all per-session state (call on re-subscribe / resync). Recent + * terminal, background, and side-channel subagent meta is retained, while + * live foreground ids wait for the snapshot roster to decide whether they + * remain live or become absence tombstones. + */ reset(sessionId: string): void; /** * Mark an agent id as a side-channel (e.g. BTW side chat) rather than a @@ -538,11 +592,64 @@ export function createAgentProjector(): AgentProjector { } function reset(sessionId: string): void { - sessions.set(sessionId, createSessionState()); + const previous = sessions.get(sessionId); + const fresh = createSessionState(); + if (previous !== undefined) { + for (const id of previous.subagentResetCandidates) { + if (!sideChannelAgents.has(id)) fresh.subagentResetCandidates.add(id); + } + for (const id of previous.subagentTombstones) { + if (!sideChannelAgents.has(id)) fresh.subagentTombstones.add(id); + } + // Retain terminal knowledge so roster-less servers cannot let a late + // progress frame resurrect a finished row. Background and side-channel + // entries also stay: the foreground-only roster never owns them. Only + // live foreground ids wait for the next present roster to decide whether + // they still exist. + for (const [id, task] of previous.subagentMeta) { + if (sideChannelAgents.has(id) || isTerminalSubagentStatus(task.status)) { + fresh.subagentResetCandidates.delete(id); + fresh.subagentTombstones.delete(id); + fresh.subagentMeta.set(id, task); + } else if (isLiveForegroundSubagent(task)) { + fresh.subagentResetCandidates.add(id); + } else { + fresh.subagentResetCandidates.delete(id); + fresh.subagentMeta.set(id, task); + } + } + trimRetainedSubagentHistory(fresh); + } + sessions.set(sessionId, fresh); + } + + function seedSubagents(sessionId: string, roster: AppTask[] | undefined): void { + const s = getOrCreate(sessionId); + if (roster === undefined) return; + const rosterIds = new Set(roster.map((task) => task.id)); + for (const [id, task] of s.subagentMeta) { + if (isLiveForegroundSubagent(task)) s.subagentResetCandidates.add(id); + } + for (const id of s.subagentResetCandidates) { + if (!sideChannelAgents.has(id) && !rosterIds.has(id)) { + s.subagentMeta.delete(id); + s.subagentTombstones.add(id); + } + } + s.subagentResetCandidates.clear(); + for (const task of roster) { + s.subagentTombstones.delete(task.id); + s.subagentMeta.set(task.id, { ...task }); + } + trimRetainedSubagentHistory(s); } function markSideChannelAgent(agentId: string): void { sideChannelAgents.add(agentId); + for (const state of sessions.values()) { + state.subagentResetCandidates.delete(agentId); + state.subagentTombstones.delete(agentId); + } } function bindNextPromptId(sessionId: string, promptId: string): void { @@ -1093,6 +1200,8 @@ export function createAgentProjector(): AgentProjector { // ----------------------------------------------------------------------- case 'subagent.spawned': { const taskId = typeof p?.subagentId === 'string' && p.subagentId.length > 0 ? p.subagentId : ulid('task_'); + s.subagentResetCandidates.delete(taskId); + s.subagentTombstones.delete(taskId); const task: AppTask = { id: taskId, sessionId, @@ -1116,6 +1225,7 @@ export function createAgentProjector(): AgentProjector { } case 'subagent.started': { + if (typeof p?.subagentId === 'string') s.subagentTombstones.delete(p.subagentId); const task = patchSubagent(s, sessionId, p?.subagentId, { subagentPhase: 'working', status: 'running', @@ -1126,6 +1236,12 @@ export function createAgentProjector(): AgentProjector { } case 'subagent.suspended': { + if ( + typeof p?.subagentId === 'string' && + s.subagentTombstones.has(p.subagentId) + ) { + break; + } const task = patchSubagent(s, sessionId, p?.subagentId, { subagentPhase: 'suspended', status: 'running', @@ -1236,6 +1352,7 @@ export function createAgentProjector(): AgentProjector { // client (subscribed late): later agent-scoped progress frames are // routed by agent id, and seeding subagentMeta here keeps them on // this one row instead of synthesizing a second one. + s.subagentTombstones.delete(agentId); const task = patchSubagent(s, sessionId, agentId, { description, backgroundTaskId: taskId, @@ -1406,7 +1523,7 @@ export function createAgentProjector(): AgentProjector { return out; } - return { project, bindNextPromptId, seedInFlight, reset, markSideChannelAgent }; + return { project, bindNextPromptId, seedInFlight, seedSubagents, reset, markSideChannelAgent }; } // --------------------------------------------------------------------------- diff --git a/apps/kimi-web/src/api/daemon/client.ts b/apps/kimi-web/src/api/daemon/client.ts index d8b1833edf..d0f44a2918 100644 --- a/apps/kimi-web/src/api/daemon/client.ts +++ b/apps/kimi-web/src/api/daemon/client.ts @@ -558,8 +558,9 @@ 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), + // Preserve `undefined`: older servers omit the roster, while an empty + // array from a newer server authoritatively means no live subagents. + subagents: data.subagents?.map(toAppTask), }; traceKeyEvent('session:snapshot:accepted', { sessionId, @@ -1500,13 +1501,22 @@ export class DaemonKimiWebApi implements KimiWebApi { // message list. if (snapshot.inFlightTurn === null) { projector.reset(sessionId); + // Seed the roster AFTER the reset so the projector's subagentMeta and + // the reducer's task rows come from the same snapshot — otherwise a + // roster-completed row forgets its terminal status here and a late + // progress frame would resurrect it as running. + projector.seedSubagents(sessionId, snapshot.subagents); return; } const appEvents = projector.seedInFlight(sessionId, snapshot.inFlightTurn); + projector.seedSubagents(sessionId, snapshot.subagents); for (const appEvent of appEvents) { handlers.onEvent(appEvent, { sessionId, seq: snapshot.asOfSeq }); } }, + reconcileSubagents(sessionId: string, roster: AppTask[]): void { + projector.seedSubagents(sessionId, roster); + }, bindNextPromptId(sessionId: string, promptId: string): void { // Wire the real daemon prompt_id into the projector so turn.started // uses it instead of a synthetic ulid('pr_'). Without this, the diff --git a/apps/kimi-web/src/api/types.ts b/apps/kimi-web/src/api/types.ts index 6c4679f165..715134bb09 100644 --- a/apps/kimi-web/src/api/types.ts +++ b/apps/kimi-web/src/api/types.ts @@ -526,8 +526,8 @@ export interface AppSessionSnapshot { messages: AppMessage[]; hasMoreMessages: boolean; inFlightTurn: AppInFlightTurn | null; - /** Live subagent roster at the watermark — rebuilds swarm cards on refresh. */ - subagents: AppTask[]; + /** Live subagent roster at the watermark — absent on older servers. */ + subagents?: AppTask[]; pendingApprovals: AppApprovalRequest[]; pendingQuestions: AppQuestionRequest[]; } @@ -569,6 +569,11 @@ export interface KimiEventConnection { * state first — call BEFORE subscribe(), with the snapshot's cursor. */ seedSnapshot(sessionId: string, snapshot: AppSessionSnapshot): void; + /** + * Reconcile only the foreground-subagent roster from a snapshot that is + * fresh for roster state but stale for unrelated session events. + */ + reconcileSubagents(sessionId: string, roster: AppTask[]): void; abort(sessionId: string, promptId: string): void; terminalAttach(sessionId: string, terminalId: string, sinceSeq?: number): void; terminalInput(sessionId: string, terminalId: string, data: string): void; diff --git a/apps/kimi-web/src/composables/client/useTaskPoller.ts b/apps/kimi-web/src/composables/client/useTaskPoller.ts index f4c9dfa973..3cd5cc1fdb 100644 --- a/apps/kimi-web/src/composables/client/useTaskPoller.ts +++ b/apps/kimi-web/src/composables/client/useTaskPoller.ts @@ -22,6 +22,7 @@ export interface UseTaskPoller { export function useTaskPoller( rawState: ExtendedState, activeAppTasks: ComputedRef, + onTaskListLoaded?: (sessionId: string) => void, ): UseTaskPoller { let taskOutputPollTimer: ReturnType | null = null; let lastPolledSessionId: string | undefined; @@ -36,6 +37,7 @@ export function useTaskPoller( // Keep WS-delivered swarm subagents that REST /tasks omits (see keepLiveSubagents). [sessionId]: keepLiveSubagents(taskList, rawState.tasksBySession[sessionId] ?? []), }; + onTaskListLoaded?.(sessionId); // Completed tasks may have real terminal output that never streamed over // WS. Fetch it once now so the rows are expandable when the session opens. await fetchTerminalTaskOutputs(sessionId, taskList); @@ -186,6 +188,7 @@ export function useTaskPoller( // Keep WS-delivered swarm subagents that REST /tasks omits (see keepLiveSubagents). [sessionId]: keepLiveSubagents(refreshed, existing), }; + onTaskListLoaded?.(sessionId); } function startTaskOutputPolling(sessionId: string): void { diff --git a/apps/kimi-web/src/composables/useKimiWebClient.ts b/apps/kimi-web/src/composables/useKimiWebClient.ts index f6e3f6a694..2675ef6aa2 100644 --- a/apps/kimi-web/src/composables/useKimiWebClient.ts +++ b/apps/kimi-web/src/composables/useKimiWebClient.ts @@ -16,7 +16,7 @@ import { import { mergeWorkspaces } from '../lib/mergeWorkspaces'; import { workspaceRootKey } from '../lib/rootKey'; import { mergeSnapshotMessages } from '../lib/snapshotMessages'; -import { mergeSnapshotSubagents } from '../lib/taskMerge'; +import { isLiveForegroundSubagent, mergeSnapshotSubagents } from '../lib/taskMerge'; import { createCoalescedAsyncRunner } from '../lib/snapshotSync'; import { loadUnread, @@ -603,6 +603,7 @@ function forgetSession(sessionId: string): void { delete rawState.messagesLoadMoreErrorBySession[sessionId]; delete epochBySession[sessionId]; sessionsRequiringSnapshot.delete(sessionId); + pendingSubagentReconciles.delete(sessionId); sessionsRetryingStaleSnapshot.delete(sessionId); sessionsKnownEmpty.delete(sessionId); // In-flight / queued prompt state: drop these too so a queued follow-up @@ -892,11 +893,15 @@ function processEvent(appEvent: AppEvent, meta: KimiEventMeta): void { // advanced past it) must not drain a second queued message. const prevSeq = rawState.lastSeqBySession[meta.sessionId] ?? 0; const wasMainTurnActive = rawState.turnActiveBySession[meta.sessionId] ?? false; + const tasksBeforeEvent = rawState.tasksBySession[meta.sessionId] ?? []; // meta carries wire-level seq/sessionId so the reducer can advance // lastSeqBySession[sessionId] = seq. Compaction completion appends a // persistent divider marker in the reducer (TUI parity: the scrollback // is kept, only a marker line records the compaction). applyEvent(appEvent, meta.sessionId, meta.seq); + if (appEventAdvancesSubagentRoster(appEvent, meta.sessionId, tasksBeforeEvent)) { + advancePendingSubagentReconcile(meta.sessionId, meta.seq); + } const sideTarget = sideChat.sideChatTargetBySession.value[meta.sessionId]; if (sideTarget) { @@ -961,6 +966,7 @@ function processEvent(appEvent: AppEvent, meta: KimiEventMeta): void { reason === 'cancelled' || reason === 'failed' || reason === 'blocked' ? 'aborted' : 'idle', wasMainTurnActive, ); + requestSubagentReconcileIfNeeded(appEvent.sessionId, meta.seq); } if ( @@ -970,6 +976,13 @@ function processEvent(appEvent: AppEvent, meta: KimiEventMeta): void { meta.seq > prevSeq ) { clearWorkingFlags(appEvent.sessionId); + // Same lost-turn.ended fallback as above: the turn boundary is the moment + // a stale foreground subagent row must be reconciled, so the quiet signal + // requests the same authoritative snapshot. A legacy quiet event still + // clears optimistic working flags when turn.started was missed, but only a + // previously-active main turn may request reconciliation — otherwise the + // normal turn-end path could request it twice. + if (wasMainTurnActive) requestSubagentReconcileIfNeeded(appEvent.sessionId, meta.seq); } // A prompt that never produced a turn gets no turn.ended and no session @@ -1111,6 +1124,20 @@ const epochBySession: Record = {}; // onResync resets the event projector, so that path must apply a snapshot even // if a newer global event advances the local cursor while the GET is in flight. const sessionsRequiringSnapshot = new Set(); +interface PendingSubagentReconcile { + /** Snapshot watermark that must include every roster-affecting event seen locally. */ + requiredSeq: number; + /** One immediate catch-up is allowed for each required watermark. */ + immediateRetrySeq?: number; + /** A later task-list refresh may retry this watermark without a tight loop. */ + deferredRetrySeq?: number; +} +// Foreground roster convergence has its own watermark. Background/BTW events +// may advance the session-wide cursor without making an older full snapshot's +// roster stale; in that case only the roster is applied. A genuinely older +// roster stays pending and retries once immediately, then on the existing +// one-second task-poll cadence. +const pendingSubagentReconciles = new Map(); // A normal foreground refresh may race one newer event. Retry once with a // fresh snapshot so volatile text missed during sleep is still restored. const sessionsRetryingStaleSnapshot = new Set(); @@ -1373,9 +1400,11 @@ async function syncSessionFromSnapshot(sessionId: string): Promise snap.asOfSeq ) { + const pendingReconcile = pendingSubagentReconciles.get(sessionId); + if (pendingReconcile !== undefined) { + if (!hasSubagentReconcileCandidate(sessionId) || snap.subagents === undefined) { + pendingSubagentReconciles.delete(sessionId); + } else if (snap.asOfSeq >= pendingReconcile.requiredSeq) { + applySnapshotSubagentRoster(sessionId, snap.subagents); + pendingSubagentReconciles.delete(sessionId); + return 'ok'; + } else if (pendingReconcile.immediateRetrySeq !== pendingReconcile.requiredSeq) { + pendingReconcile.immediateRetrySeq = pendingReconcile.requiredSeq; + pendingReconcile.deferredRetrySeq = undefined; + snapshotSyncRunner.request(sessionId); + return 'ok'; + } else { + pendingReconcile.deferredRetrySeq = pendingReconcile.requiredSeq; + return 'ok'; + } + } if (sessionsRetryingStaleSnapshot.delete(sessionId)) return 'ok'; sessionsRetryingStaleSnapshot.add(sessionId); snapshotSyncRunner.request(sessionId); @@ -1419,12 +1466,14 @@ async function syncSessionFromSnapshot(sessionId: string): Promise task.id !== sideChannelAgentId && isLiveForegroundSubagent(task), + ); +} + +function appEventAdvancesSubagentRoster( + event: AppEvent, + sessionId: string, + tasksBeforeEvent: AppTask[], +): boolean { + if (event.type === 'turnActiveChanged') return true; + const taskId = + event.type === 'taskCreated' + ? event.task.id + : event.type === 'taskCompleted' + ? event.taskId + : undefined; + if (taskId === undefined) return false; + const sideChannelAgentId = sideChat.sideChatTargetBySession.value[sessionId]?.agentId; + if (taskId === sideChannelAgentId) return false; + const previous = tasksBeforeEvent.find((task) => task.id === taskId); + const current = (rawState.tasksBySession[sessionId] ?? []).find((task) => task.id === taskId); + if (current?.kind !== 'subagent') return false; + if (previous === undefined) return isLiveForegroundSubagent(current); + const wasLiveForeground = isLiveForegroundSubagent(previous); + const isLiveForeground = isLiveForegroundSubagent(current); + if (!wasLiveForeground && !isLiveForeground) return false; + return ( + previous.description !== current.description || + previous.status !== current.status || + previous.subagentPhase !== current.subagentPhase || + previous.subagentType !== current.subagentType || + previous.parentToolCallId !== current.parentToolCallId || + previous.swarmIndex !== current.swarmIndex || + previous.runInBackground !== current.runInBackground || + previous.backgroundTaskId !== current.backgroundTaskId || + previous.suspendedReason !== current.suspendedReason || + previous.startedAt !== current.startedAt || + previous.outputPreview !== current.outputPreview || + previous.outputBytes !== current.outputBytes || + previous.completedAt !== current.completedAt + ); +} + +function advancePendingSubagentReconcile(sessionId: string, seq: number): void { + const pending = pendingSubagentReconciles.get(sessionId); + if (pending === undefined || seq <= pending.requiredSeq) return; + if (!hasSubagentReconcileCandidate(sessionId)) { + pendingSubagentReconciles.delete(sessionId); + return; + } + pending.requiredSeq = seq; + pending.immediateRetrySeq = undefined; + pending.deferredRetrySeq = undefined; + snapshotSyncRunner.request(sessionId); +} + +function applySnapshotSubagentRoster(sessionId: string, roster: AppTask[]): void { + rawState.tasksBySession = { + ...rawState.tasksBySession, + [sessionId]: mergeSnapshotSubagents( + roster, + rawState.tasksBySession[sessionId] ?? [], + sideChat.sideChatTargetBySession.value[sessionId]?.agentId, + ), + }; + eventConn?.reconcileSubagents(sessionId, roster); +} + +function requestSubagentReconcileIfNeeded(sessionId: string, requiredSeq: number): void { + if (!hasSubagentReconcileCandidate(sessionId)) return; + const pending = pendingSubagentReconciles.get(sessionId); + if (pending === undefined) { + pendingSubagentReconciles.set(sessionId, { requiredSeq }); + } else if (requiredSeq > pending.requiredSeq) { + pending.requiredSeq = requiredSeq; + pending.immediateRetrySeq = undefined; + pending.deferredRetrySeq = undefined; + } + snapshotSyncRunner.request(sessionId); +} + +function retryPendingSubagentReconcile(sessionId: string): void { + const pending = pendingSubagentReconciles.get(sessionId); + if (pending === undefined || pending.deferredRetrySeq !== pending.requiredSeq) { + return; + } + if (!hasSubagentReconcileCandidate(sessionId)) { + pendingSubagentReconciles.delete(sessionId); + return; + } + pending.deferredRetrySeq = undefined; + snapshotSyncRunner.request(sessionId); +} + function hasLoadedMessages(sessionId: string): boolean { return Object.prototype.hasOwnProperty.call(rawState.messagesBySession, sessionId); } @@ -1959,7 +2116,11 @@ const activeAppTasks = computed(() => { return (rawState.tasksBySession[sid] ?? []).filter((task) => task.id !== hiddenBtwAgentId); }); -const taskPoller = useTaskPoller(rawState, activeAppTasks); +const taskPoller = useTaskPoller( + rawState, + activeAppTasks, + retryPendingSubagentReconcile, +); const turns = computed(() => { const sid = rawState.activeSessionId; diff --git a/apps/kimi-web/src/lib/taskMerge.ts b/apps/kimi-web/src/lib/taskMerge.ts index d3c41b9868..9da5816272 100644 --- a/apps/kimi-web/src/lib/taskMerge.ts +++ b/apps/kimi-web/src/lib/taskMerge.ts @@ -61,14 +61,31 @@ export function keepLiveSubagents(restBased: AppTask[], existing: AppTask[]): Ap return [...rest, ...merged]; } +/** True for a still-live foreground row that only the event stream/roster owns. */ +export function isLiveForegroundSubagent(t: AppTask): boolean { + return ( + t.kind === 'subagent' && + t.status === 'running' && + t.runInBackground !== true && + t.backgroundTaskId === undefined + ); +} + /** * 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). + * roster does not own (background tasks from REST and terminal history). + * Older servers omit the roster; only a present roster, including `[]`, may + * remove a stale live foreground row. A side-channel agent id is also outside + * roster ownership and survives an authoritative merge. */ -export function mergeSnapshotSubagents(roster: AppTask[], existing: AppTask[]): AppTask[] { - if (roster.length === 0) return existing; +export function mergeSnapshotSubagents( + roster: AppTask[] | undefined, + existing: AppTask[], + sideChannelAgentId?: string, +): AppTask[] { + if (roster === undefined) 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) => { @@ -76,6 +93,11 @@ export function mergeSnapshotSubagents(roster: AppTask[], existing: AppTask[]): if (!live) return task; return { ...task, outputLines: live.outputLines, text: live.text }; }); - const kept = existing.filter((t) => !rosterIds.has(t.id)); + const kept = existing.filter( + (task) => + !rosterIds.has(task.id) && + (task.id === sideChannelAgentId || !isLiveForegroundSubagent(task)), + ); + if (merged.length === 0 && kept.length === existing.length) return existing; return kept.length === 0 ? merged : [...merged, ...kept]; } diff --git a/apps/kimi-web/test/agent-event-projector.test.ts b/apps/kimi-web/test/agent-event-projector.test.ts index 763e2fe2b9..c0a75a70cc 100644 --- a/apps/kimi-web/test/agent-event-projector.test.ts +++ b/apps/kimi-web/test/agent-event-projector.test.ts @@ -580,3 +580,468 @@ describe('background subagent task registration', () => { ]); }); }); + +describe('subagent terminal stickiness', () => { + function spawn(projector: ReturnType): void { + projector.project( + 'subagent.spawned', + { subagentId: 'sub-1', description: 'Explore repo', runInBackground: false }, + 's1', + ); + } + + function spawnAndComplete(projector: ReturnType): void { + spawn(projector); + projector.project( + 'subagent.completed', + { subagentId: 'sub-1', resultSummary: 'done' }, + 's1', + ); + } + + it('a late assistant.delta after completion keeps the row terminal', () => { + const projector = createAgentProjector(); + spawnAndComplete(projector); + + const events = projector.project('assistant.delta', { agentId: 'sub-1', delta: 'late' }, 's1'); + + // The progress chunk is still forwarded (it may be legitimate trailing + // output), but no running-stamped taskCreated may resurrect the row. + expect(events.filter((e) => e.type === 'taskCreated')).toEqual([]); + expect(events).toContainEqual( + expect.objectContaining({ type: 'taskProgress', taskId: 'sub-1', outputChunk: 'late' }), + ); + }); + + it('a late tool.call.started after completion keeps the row terminal', () => { + const projector = createAgentProjector(); + spawnAndComplete(projector); + + const events = projector.project( + 'tool.call.started', + { agentId: 'sub-1', turnId: 1, toolCallId: 'tc-1', name: 'Read', args: { path: '/x' } }, + 's1', + ); + + expect(events.filter((e) => e.type === 'taskCreated')).toEqual([]); + expect(events).toContainEqual(expect.objectContaining({ type: 'taskProgress', taskId: 'sub-1' })); + }); + + it('an explicit subagent.started (resume) still re-opens a completed row', () => { + const projector = createAgentProjector(); + spawnAndComplete(projector); + + const resumed = projector.project('subagent.started', { subagentId: 'sub-1' }, 's1'); + expect(resumed).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ id: 'sub-1', status: 'running' }), + }), + ); + + // Progress during the resumed run is projected normally again. + const events = projector.project('assistant.delta', { agentId: 'sub-1', delta: 'again' }, 's1'); + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ id: 'sub-1', status: 'running' }), + }), + ); + }); + + it('a roster seed after a resync reset keeps the terminal knowledge', () => { + const projector = createAgentProjector(); + spawnAndComplete(projector); + + // Resync/snapshot rebuild: the reset wipes subagentMeta; the snapshot's + // roster must restore it, or a late progress frame resurrects the row. + projector.reset('s1'); + projector.seedSubagents('s1', [ + { + id: 'sub-1', + sessionId: 's1', + kind: 'subagent', + description: 'Explore repo', + status: 'completed', + createdAt: '2026-01-01T00:00:00.000Z', + subagentPhase: 'completed', + }, + ]); + + const events = projector.project('assistant.delta', { agentId: 'sub-1', delta: 'late' }, 's1'); + expect(events.filter((e) => e.type === 'taskCreated')).toEqual([]); + expect(events).toContainEqual( + expect.objectContaining({ type: 'taskProgress', taskId: 'sub-1', outputChunk: 'late' }), + ); + }); + + it('an omitted roster seed (older server) still keeps terminal knowledge across the reset', () => { + const projector = createAgentProjector(); + spawnAndComplete(projector); + projector.reset('s1'); + projector.seedSubagents('s1', undefined); + + // The roster cannot re-seed terminal meta on an older server, so reset + // itself retains it: a late progress frame must not resurrect the row. + const events = projector.project('assistant.delta', { agentId: 'sub-1', delta: 'late' }, 's1'); + expect(events.filter((e) => e.type === 'taskCreated')).toEqual([]); + expect(events).toContainEqual( + expect.objectContaining({ type: 'taskProgress', taskId: 'sub-1', outputChunk: 'late' }), + ); + }); + + it('an omitted legacy roster lets live progress recreate a running row after reset', () => { + const projector = createAgentProjector(); + spawn(projector); + + projector.reset('s1'); + projector.seedSubagents('s1', undefined); + const events = projector.project('assistant.delta', { agentId: 'sub-1', delta: 'live' }, 's1'); + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ id: 'sub-1', status: 'running' }), + }), + ); + }); + + it('a later authoritative empty roster still tombstones a row after a legacy omission', () => { + const projector = createAgentProjector(); + spawn(projector); + projector.reset('s1'); + projector.seedSubagents('s1', undefined); + projector.reset('s1'); + projector.seedSubagents('s1', []); + + const events = projector.project('assistant.delta', { agentId: 'sub-1', delta: 'late' }, 's1'); + + expect(events).toEqual([]); + }); + + it('an authoritative foreground roster omission keeps a background subagent live', () => { + const projector = createAgentProjector(); + projector.project( + 'subagent.spawned', + { subagentId: 'sub-1', description: 'Background work', runInBackground: true }, + 's1', + ); + projector.reset('s1'); + projector.seedSubagents('s1', []); + + const events = projector.project( + 'assistant.delta', + { agentId: 'sub-1', delta: 'still running' }, + 's1', + ); + + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ + id: 'sub-1', + status: 'running', + runInBackground: true, + }), + }), + ); + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskProgress', + taskId: 'sub-1', + outputChunk: 'still running', + }), + ); + }); + + it('an authoritative foreground roster omission keeps a detached subagent live', () => { + const projector = createAgentProjector(); + spawn(projector); + projector.project( + 'task.started', + { + info: { + taskId: 'task-1', + kind: 'agent', + agentId: 'sub-1', + detached: true, + description: 'Detached work', + }, + }, + 's1', + ); + projector.reset('s1'); + projector.seedSubagents('s1', []); + + const events = projector.project( + 'assistant.delta', + { agentId: 'sub-1', delta: 'still detached' }, + 's1', + ); + + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ + id: 'sub-1', + status: 'running', + runInBackground: true, + backgroundTaskId: 'task-1', + }), + }), + ); + }); + + it('an authoritative foreground roster omission keeps side-channel tool progress live', () => { + const projector = createAgentProjector(); + projector.markSideChannelAgent('btw-1'); + projector.project( + 'tool.call.started', + { + agentId: 'btw-1', + turnId: 1, + toolCallId: 'tc-1', + name: 'Read', + args: { path: '/before' }, + }, + 's1', + ); + projector.reset('s1'); + projector.seedSubagents('s1', []); + + const events = projector.project( + 'tool.call.started', + { + agentId: 'btw-1', + turnId: 2, + toolCallId: 'tc-2', + name: 'Read', + args: { path: '/after' }, + }, + 's1', + ); + + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ id: 'btw-1', status: 'running' }), + }), + ); + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskProgress', + taskId: 'btw-1', + outputChunk: expect.stringContaining('/after'), + }), + ); + }); + + it('an authoritative roster omission prevents late progress from recreating a running row', () => { + const projector = createAgentProjector(); + spawn(projector); + projector.reset('s1'); + // A resync reset can be followed by another reset while applying its + // snapshot; the pending authoritative-absence decision must survive both. + projector.reset('s1'); + projector.seedSubagents('s1', []); + + const events = projector.project('assistant.delta', { agentId: 'sub-1', delta: 'late' }, 's1'); + + expect(events.filter((event) => event.type === 'taskCreated')).toEqual([]); + }); + + it('an authoritative roster omission tombstones a live row without a reset', () => { + const projector = createAgentProjector(); + spawn(projector); + projector.seedSubagents('s1', []); + + expect( + projector.project('assistant.delta', { agentId: 'sub-1', delta: 'late' }, 's1'), + ).toEqual([]); + }); + + it('an in-flight seed tombstones a running row omitted from its authoritative roster', () => { + const projector = createAgentProjector(); + spawn(projector); + projector.seedInFlight('s1', { + turnId: 2, + assistantText: '', + thinkingText: '', + runningTools: [], + promptId: 'prompt-2', + }); + projector.seedSubagents('s1', []); + + expect( + projector.project('assistant.delta', { agentId: 'sub-1', delta: 'late' }, 's1'), + ).toEqual([]); + }); + + it('bounds terminal metadata retained across resets', () => { + const projector = createAgentProjector(); + for (let index = 0; index < 257; index += 1) { + const subagentId = `sub-${String(index)}`; + projector.project( + 'subagent.spawned', + { subagentId, description: 'Explore repo', runInBackground: false }, + 's1', + ); + projector.project( + 'subagent.completed', + { subagentId, resultSummary: 'done' }, + 's1', + ); + } + + projector.reset('s1'); + + expect( + projector.project('assistant.delta', { agentId: 'sub-0', delta: 'old' }, 's1'), + ).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ id: 'sub-0', status: 'running' }), + }), + ); + expect( + projector.project('assistant.delta', { agentId: 'sub-256', delta: 'recent' }, 's1'), + ).not.toContainEqual(expect.objectContaining({ type: 'taskCreated' })); + }); + + it('bounds authoritative-absence tombstones retained across resets', () => { + const projector = createAgentProjector(); + for (let index = 0; index < 257; index += 1) { + const subagentId = `sub-${String(index)}`; + projector.project( + 'subagent.spawned', + { subagentId, description: 'Explore repo', runInBackground: false }, + 's1', + ); + projector.reset('s1'); + projector.seedSubagents('s1', []); + } + + expect( + projector.project('assistant.delta', { agentId: 'sub-0', delta: 'old' }, 's1'), + ).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ id: 'sub-0', status: 'running' }), + }), + ); + expect( + projector.project('assistant.delta', { agentId: 'sub-256', delta: 'recent' }, 's1'), + ).toEqual([]); + }); + + it('a background registration reopens a roster tombstone for later progress', () => { + const projector = createAgentProjector(); + spawn(projector); + projector.reset('s1'); + projector.seedSubagents('s1', []); + projector.project( + 'task.started', + { + info: { + taskId: 'task-1', + kind: 'agent', + agentId: 'sub-1', + detached: true, + description: 'Detached work', + }, + }, + 's1', + ); + + const events = projector.project( + 'assistant.delta', + { agentId: 'sub-1', delta: 'detached' }, + 's1', + ); + + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ + id: 'sub-1', + status: 'running', + backgroundTaskId: 'task-1', + }), + }), + ); + }); + + it('a later authoritative roster entry reopens an earlier absence tombstone', () => { + const projector = createAgentProjector(); + spawn(projector); + projector.reset('s1'); + projector.seedSubagents('s1', []); + projector.reset('s1'); + projector.seedSubagents('s1', [ + { + id: 'sub-1', + sessionId: 's1', + kind: 'subagent', + description: 'Explore repo', + status: 'running', + createdAt: '2026-01-01T00:00:00.000Z', + subagentPhase: 'working', + }, + ]); + + const events = projector.project( + 'assistant.delta', + { agentId: 'sub-1', delta: 'resumed' }, + 's1', + ); + + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ id: 'sub-1', status: 'running' }), + }), + ); + }); + + it.each(['subagent.spawned', 'subagent.started'] as const)( + '%s explicitly reopens an authoritative-roster tombstone for later progress', + (lifecycleType) => { + const projector = createAgentProjector(); + spawn(projector); + projector.reset('s1'); + projector.seedSubagents('s1', []); + projector.project( + lifecycleType, + { subagentId: 'sub-1', description: 'Explore repo', runInBackground: false }, + 's1', + ); + + const events = projector.project( + 'assistant.delta', + { agentId: 'sub-1', delta: 'resumed' }, + 's1', + ); + + expect(events).toContainEqual( + expect.objectContaining({ + type: 'taskCreated', + task: expect.objectContaining({ id: 'sub-1', status: 'running' }), + }), + ); + }, + ); + + it('a suspended lifecycle cannot reopen an authoritative-roster tombstone', () => { + const projector = createAgentProjector(); + spawn(projector); + projector.reset('s1'); + projector.seedSubagents('s1', []); + + const events = projector.project( + 'subagent.suspended', + { subagentId: 'sub-1', reason: 'waiting' }, + 's1', + ); + + expect(events.filter((event) => event.type === 'taskCreated')).toEqual([]); + }); +}); diff --git a/apps/kimi-web/test/daemon-client.test.ts b/apps/kimi-web/test/daemon-client.test.ts index dbe9c79186..f9fb77a4fd 100644 --- a/apps/kimi-web/test/daemon-client.test.ts +++ b/apps/kimi-web/test/daemon-client.test.ts @@ -1,6 +1,7 @@ // apps/kimi-web/test/daemon-client.test.ts // DaemonKimiWebApi public REST adapter: session export binary/error contracts, -// getSessionGoal wire → app mapping, and raw stream-coordinate delivery. +// getSessionGoal wire → app mapping, raw stream-coordinate delivery, and +// roster-only projector reconciliation. // Wiring: real client/projector; fetch or WebSocket is stubbed at the network boundary. // Run: pnpm --filter @moonshot-ai/kimi-web exec vitest run test/daemon-client.test.ts @@ -176,6 +177,62 @@ describe('DaemonKimiWebApi.exportSession', () => { }); }); +describe('DaemonKimiWebApi.getSessionSnapshot', () => { + const wireSnapshot = { + as_of_seq: 1, + epoch: 'epoch-1', + session: { + id: 'session-1', + title: 'Session', + created_at: '2026-01-01T00:00:00.000Z', + updated_at: '2026-01-01T00:00:00.000Z', + busy: false, + archived: false, + metadata: { cwd: '/workspace' }, + agent_config: { model: 'model-1' }, + usage: { + input_tokens: 0, + output_tokens: 0, + cache_read_tokens: 0, + cache_creation_tokens: 0, + total_cost_usd: 0, + context_tokens: 0, + context_limit: 0, + turn_count: 0, + }, + permission_rules: [], + message_count: 0, + last_seq: 1, + }, + messages: { items: [], has_more: false }, + in_flight_turn: null, + pending_approvals: [], + pending_questions: [], + }; + + beforeEach(() => { + vi.stubGlobal('location', { search: '' }); + vi.stubGlobal('fetch', vi.fn()); + }); + + afterEach(() => { + vi.unstubAllGlobals(); + }); + + it('distinguishes an omitted legacy roster from an authoritative empty roster', async () => { + vi.mocked(fetch) + .mockResolvedValueOnce(envelope(wireSnapshot)) + .mockResolvedValueOnce(envelope({ ...wireSnapshot, subagents: [] })); + + const api = createApi(); + const legacy = await api.getSessionSnapshot('session-1'); + const current = await api.getSessionSnapshot('session-1'); + + expect(legacy.subagents).toBeUndefined(); + expect(current.subagents).toEqual([]); + }); +}); + describe('DaemonKimiWebApi.getSessionGoal', () => { beforeEach(() => { vi.stubGlobal('fetch', vi.fn()); @@ -296,6 +353,46 @@ describe('DaemonKimiWebApi.connectEvents', () => { }); }); + it('reconciles an authoritative roster without resetting the connection', () => { + FakeWebSocket.instances = []; + vi.stubGlobal('WebSocket', FakeWebSocket as unknown as typeof WebSocket); + const received: AppEvent[] = []; + connection = createApi().connectEvents({ + onEvent(event) { + received.push(event); + }, + onResync() {}, + onError() {}, + onConnectionChange() {}, + }); + const socket = FakeWebSocket.instances[0]!; + + socket.emit({ type: 'server_hello', payload: { protocol_version: 2 } }); + socket.emit({ + type: 'subagent.spawned', + seq: 1, + session_id: 'session-1', + timestamp: '2026-01-01T00:00:00.000Z', + payload: { + subagentId: 'sub-1', + description: 'Explore repo', + runInBackground: false, + }, + }); + received.length = 0; + + connection.reconcileSubagents('session-1', []); + socket.emit({ + type: 'assistant.delta', + seq: 2, + session_id: 'session-1', + timestamp: '2026-01-01T00:00:00.000Z', + payload: { agentId: 'sub-1', delta: 'late' }, + }); + + expect(received).toEqual([]); + }); + it('projects list-level work facts from the global session event', () => { FakeWebSocket.instances = []; vi.stubGlobal('WebSocket', FakeWebSocket as unknown as typeof WebSocket); diff --git a/apps/kimi-web/test/event-batcher.test.ts b/apps/kimi-web/test/event-batcher.test.ts index 0108d92419..7cd10ba9a1 100644 --- a/apps/kimi-web/test/event-batcher.test.ts +++ b/apps/kimi-web/test/event-batcher.test.ts @@ -662,6 +662,7 @@ describe('useKimiWebClient (resync integration)', () => { seedCalls += 1; if (seedCalls === 2) resolveSnapshotApplied(); }), + reconcileSubagents: vi.fn(), abort: vi.fn(), terminalAttach: vi.fn(), terminalInput: vi.fn(), @@ -746,9 +747,28 @@ describe('useKimiWebClient (resync integration)', () => { expect(assistantText()).toBe('seed'); expect(handlers).toBeDefined(); - const beforeResync = pendingDelta('before', 4, { seq: 11 }); + handlers!.onEvent( + { + type: 'taskCreated', + sessionId, + task: { + id: 'agent-stale', + sessionId, + kind: 'subagent', + description: 'Finished while disconnected', + status: 'running', + busy: true, + createdAt: '2026-01-01T00:00:00.000Z', + subagentPhase: 'working', + }, + }, + { sessionId, seq: 11 }, + ); + expect(client.activeAppTasks.value.map((task) => task.id)).toEqual(['agent-stale']); + + const beforeResync = pendingDelta('before', 4, { seq: 12 }); handlers!.onEvent(beforeResync.appEvent, beforeResync.meta); - handlers!.onResync(sessionId, 11, 'epoch-2'); + handlers!.onResync(sessionId, 12, 'epoch-2'); // onResync synchronously drains pre-resync text onto the old state. expect(assistantText()).toBe('seedbefore'); @@ -756,11 +776,12 @@ describe('useKimiWebClient (resync integration)', () => { // A frame can race the REST request. The second flush must consume it on // the old state before the authoritative snapshot replaces that state. - const duringSnapshot = pendingDelta('old', 10, { seq: 12 }); + const duringSnapshot = pendingDelta('old', 10, { seq: 13 }); handlers!.onEvent(duringSnapshot.appEvent, duringSnapshot.meta); resolveAuthoritativeSnapshot(authoritativeSnapshot); await snapshotApplied; expect(assistantText()).toBe('snapshot'); + expect(client.activeAppTasks.value).toEqual([]); const live = pendingDelta(' live', 8, { seq: 21 }); handlers!.onEvent(live.appEvent, live.meta); @@ -775,6 +796,645 @@ describe('useKimiWebClient (resync integration)', () => { vi.unstubAllGlobals(); } }); + + /** + * Shared harness for the turn-boundary subagent reconcile tests: a loaded + * session whose every further getSessionSnapshot call is handed to the test. + */ + async function setupReconcileHarness() { + vi.stubGlobal('WebSocket', vi.fn()); + // The client module is a singleton (eventConn, rawState); earlier tests in + // this file already imported it. Reset so each harness gets fresh state + // and its own connectEvents handlers. + vi.resetModules(); + + const sessionId = 'session-1'; + const session: AppSession = { + id: sessionId, + title: 'Session', + createdAt: '2026-01-01T00:00:00.000Z', + updatedAt: '2026-01-01T00:00:00.000Z', + status: 'running', + archived: false, + currentPromptId: 'prompt-1', + cwd: '/workspace', + model: 'model-1', + usage: { + inputTokens: 0, + outputTokens: 0, + cacheReadTokens: 0, + cacheCreationTokens: 0, + totalCostUsd: 0, + contextTokens: 0, + contextLimit: 0, + turnCount: 1, + }, + messageCount: 1, + lastSeq: 10, + workspaceId: 'workspace-1', + }; + const snapshot: AppSessionSnapshot = { + asOfSeq: 10, + epoch: 'epoch-1', + session, + messages: [ + { + id: 'message-1', + sessionId, + role: 'assistant', + content: [{ type: 'text', text: 'seed' }], + createdAt: '2026-01-01T00:00:00.000Z', + }, + ], + hasMoreMessages: false, + inFlightTurn: null, + subagents: [], + pendingApprovals: [], + pendingQuestions: [], + }; + + let handlers: KimiEventHandlers | undefined; + const reconcileResponses = Array.from({ length: 4 }, () => { + let resolveSnapshot!: (value: AppSessionSnapshot) => void; + const snapshotResponse = new Promise((resolve) => { + resolveSnapshot = resolve; + }); + let announceRequested!: () => void; + const requested = new Promise((resolve) => { + announceRequested = resolve; + }); + return { announceRequested, requested, resolveSnapshot, snapshotResponse }; + }); + let snapshotCalls = 0; + const getSessionSnapshot = vi.fn((_id: string) => { + snapshotCalls += 1; + if (snapshotCalls === 1) return Promise.resolve(snapshot); + const response = reconcileResponses[snapshotCalls - 2]; + if (response === undefined) throw new Error('Unexpected reconcile snapshot request'); + response.announceRequested(); + return response.snapshotResponse; + }); + const submitPrompt = vi.fn(async () => ({ + promptId: 'prompt-new', + userMessageId: 'message-new', + status: 'running' as const, + })); + let resolveSeeded!: () => void; + const seeded = new Promise((resolve) => { + resolveSeeded = resolve; + }); + let resolveRosterReconciled!: () => void; + const rosterReconciled = new Promise((resolve) => { + resolveRosterReconciled = resolve; + }); + let seedCalls = 0; + const connection: KimiEventConnection = { + subscribe: vi.fn(), + unsubscribe: vi.fn(), + bindNextPromptId: vi.fn(), + seedSnapshot: vi.fn(() => { + seedCalls += 1; + if (seedCalls === 2) resolveSeeded(); + }), + reconcileSubagents: vi.fn(() => { + resolveRosterReconciled(); + }), + abort: vi.fn(), + terminalAttach: vi.fn(), + terminalInput: vi.fn(), + terminalResize: vi.fn(), + terminalDetach: vi.fn(), + terminalClose: vi.fn(), + markSideChannelAgent: vi.fn(), + health: () => ({ connected: true, open: true, stale: false }), + reconnect: vi.fn(), + close: vi.fn(), + }; + const api: Partial = { + getAuth: vi.fn(async () => ({ + ready: true, + defaultModel: 'model-1', + managedProvider: null, + })), + getHealth: vi.fn(async () => ({ status: 'ok', uptimeSec: 1 })), + getMeta: vi.fn(async () => ({ + serverVersion: '0.0.0', + serverId: 'server-1', + startedAt: '2026-01-01T00:00:00.000Z', + capabilities: {}, + openInApps: [], + dangerousBypassAuth: false, + backend: 'v2', + })), + getConfig: vi.fn(async () => ({ providers: {}, defaultModel: 'model-1' })), + listModels: vi.fn(async () => []), + listProviders: vi.fn(async () => []), + listWorkspaces: vi.fn(async () => [ + { + id: 'workspace-1', + root: '/workspace', + name: 'Workspace', + sessionCount: 1, + }, + ]), + getFsHome: vi.fn(async () => ({ home: '/home/test', recentRoots: [] })), + listSessions: vi.fn(async () => ({ items: [session], hasMore: false })), + getSessionSnapshot, + submitPrompt, + startBtw: vi.fn(async () => ({ agentId: 'btw-1' })), + getSessionStatus: vi.fn(async () => ({ + model: 'model-1', + thinkingEffort: 'high', + permission: 'manual', + planMode: false, + swarmMode: false, + contextTokens: 0, + maxContextTokens: 0, + contextUsage: 0, + })), + getSessionGoal: vi.fn(async () => null), + getSessionWarnings: vi.fn(async () => []), + getGitStatus: vi.fn(async () => ({ + branch: '', + ahead: 0, + behind: 0, + entries: {}, + additions: 0, + deletions: 0, + pullRequest: null, + })), + listTasks: vi.fn(async () => []), + listSkills: vi.fn(async () => []), + listSkillsForWorkspace: vi.fn(async () => []), + getFileUrl: (fileId) => `file:${fileId}`, + connectEvents: vi.fn((nextHandlers) => { + handlers = nextHandlers; + return connection; + }), + }; + for (const key of Object.keys(clientApiMock)) delete clientApiMock[key]; + Object.assign(clientApiMock, api); + + const { useKimiWebClient } = await import('../src/composables/useKimiWebClient'); + const client = useKimiWebClient(); + await client.load(); + + const staleRow = { + id: 'agent-stale', + sessionId, + kind: 'subagent' as const, + description: 'Finished without a terminal event', + status: 'running' as const, + createdAt: '2026-01-01T00:00:00.000Z', + subagentPhase: 'working' as const, + }; + const authoritativeSnapshot: AppSessionSnapshot = { + ...snapshot, + asOfSeq: 20, + session: { ...session, lastSeq: 20, status: 'idle' as const }, + subagents: [ + { + ...staleRow, + status: 'completed', + subagentPhase: 'completed', + completedAt: '2026-01-01T00:01:00.000Z', + }, + ], + }; + + return { + sessionId, + client, + connection, + handlers: handlers!, + staleRow, + authoritativeSnapshot, + getSessionSnapshot, + submitPrompt, + reconcileRequested: reconcileResponses[0].requested, + resolveReconcileSnapshot: reconcileResponses[0].resolveSnapshot, + waitForReconcileRequest: (index: number) => reconcileResponses[index].requested, + resolveReconcileSnapshotAt: (index: number, value: AppSessionSnapshot) => { + reconcileResponses[index].resolveSnapshot(value); + }, + seeded, + rosterReconciled, + }; + } + + it('reconciles a stale foreground subagent row when the main turn ends', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 11 }, + ); + expect(h.client.activeAppTasks.value.map((task) => task.id)).toEqual(['agent-stale']); + + // Normal turn boundary: no reconnection, no resync — the turn-end hook + // alone must trigger the authoritative reconcile. + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await h.reconcileRequested; + + h.resolveReconcileSnapshot(h.authoritativeSnapshot); + await h.seeded; + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.status])).toEqual([ + ['agent-stale', 'completed'], + ]); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('reconciles a stale foreground subagent row on the lost-turn.ended fallback', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: true }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 12 }, + ); + expect(h.client.activeAppTasks.value.map((task) => task.id)).toEqual(['agent-stale']); + + // turn.ended never arrives; the quiet sessionWorkChanged fallback must + // request the same reconcile instead of only clearing working flags. + h.handlers.onEvent( + { type: 'sessionWorkChanged', sessionId: h.sessionId, busy: false, mainTurnActive: false }, + { sessionId: h.sessionId, seq: 13 }, + ); + await h.reconcileRequested; + + h.resolveReconcileSnapshot(h.authoritativeSnapshot); + await h.seeded; + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.status])).toEqual([ + ['agent-stale', 'completed'], + ]); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('applies a roster-fresh snapshot when background progress advances the session cursor', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await h.reconcileRequested; + + h.resolveReconcileSnapshot({ + ...h.authoritativeSnapshot, + asOfSeq: 11, + session: { ...h.authoritativeSnapshot.session, lastSeq: 11 }, + }); + await h.waitForReconcileRequest(1); + h.handlers.onEvent( + { + type: 'taskProgress', + sessionId: h.sessionId, + taskId: 'background-task', + outputChunk: 'still running', + stream: 'stdout', + }, + { sessionId: h.sessionId, seq: 20 }, + ); + h.resolveReconcileSnapshotAt(1, { + ...h.authoritativeSnapshot, + asOfSeq: 12, + session: { ...h.authoritativeSnapshot.session, lastSeq: 12 }, + }); + await h.rosterReconciled; + + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.status])).toEqual([ + ['agent-stale', 'completed'], + ]); + expect(h.getSessionSnapshot).toHaveBeenCalledTimes(3); + expect(h.connection.reconcileSubagents).toHaveBeenCalledWith( + h.sessionId, + h.authoritativeSnapshot.subagents, + ); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('does not make repeated subagent output projection stale the roster watermark', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await h.reconcileRequested; + + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 20 }, + ); + h.resolveReconcileSnapshot({ + ...h.authoritativeSnapshot, + asOfSeq: 12, + session: { ...h.authoritativeSnapshot.session, lastSeq: 12 }, + }); + await h.rosterReconciled; + + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.status])).toEqual([ + ['agent-stale', 'completed'], + ]); + expect(h.getSessionSnapshot).toHaveBeenCalledTimes(2); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('waits for a snapshot that covers a newer foreground lifecycle change', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await h.reconcileRequested; + + const suspendedRow = { + ...h.staleRow, + subagentPhase: 'suspended' as const, + suspendedReason: 'waiting', + }; + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: suspendedRow }, + { sessionId: h.sessionId, seq: 20 }, + ); + h.resolveReconcileSnapshot({ + ...h.authoritativeSnapshot, + asOfSeq: 12, + session: { ...h.authoritativeSnapshot.session, lastSeq: 12 }, + }); + await h.waitForReconcileRequest(1); + + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.subagentPhase])).toEqual([ + ['agent-stale', 'suspended'], + ]); + h.resolveReconcileSnapshotAt(1, { + ...h.authoritativeSnapshot, + asOfSeq: 20, + session: { ...h.authoritativeSnapshot.session, lastSeq: 20 }, + subagents: [suspendedRow], + }); + await h.seeded; + + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.subagentPhase])).toEqual([ + ['agent-stale', 'suspended'], + ]); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('queues only one immediate catch-up when the roster snapshot remains behind', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await h.reconcileRequested; + + const staleSnapshot = { + ...h.authoritativeSnapshot, + asOfSeq: 11, + session: { ...h.authoritativeSnapshot.session, lastSeq: 11 }, + }; + h.resolveReconcileSnapshot(staleSnapshot); + await h.waitForReconcileRequest(1); + h.resolveReconcileSnapshotAt(1, staleSnapshot); + await Promise.resolve(); + await Promise.resolve(); + + expect(h.getSessionSnapshot).toHaveBeenCalledTimes(3); + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.status])).toEqual([ + ['agent-stale', 'running'], + ]); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('falls back to the normal stale-snapshot retry when an older server omits the roster', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await h.reconcileRequested; + + h.resolveReconcileSnapshot({ + ...h.authoritativeSnapshot, + asOfSeq: 11, + session: { ...h.authoritativeSnapshot.session, lastSeq: 11 }, + subagents: undefined, + }); + await h.waitForReconcileRequest(1); + h.resolveReconcileSnapshotAt(1, { + ...h.authoritativeSnapshot, + asOfSeq: 12, + session: { ...h.authoritativeSnapshot.session, lastSeq: 12 }, + subagents: undefined, + }); + await h.seeded; + + expect(h.getSessionSnapshot).toHaveBeenCalledTimes(3); + expect(h.connection.seedSnapshot).toHaveBeenCalledTimes(2); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('completes a pending reconcile on a later refresh after stale roster snapshots', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await h.reconcileRequested; + + const staleSnapshot = { + ...h.authoritativeSnapshot, + asOfSeq: 11, + session: { ...h.authoritativeSnapshot.session, lastSeq: 11 }, + }; + h.resolveReconcileSnapshot(staleSnapshot); + await h.waitForReconcileRequest(1); + h.resolveReconcileSnapshotAt(1, staleSnapshot); + await Promise.resolve(); + await Promise.resolve(); + + const refreshed = h.client.selectSession(h.sessionId); + await h.waitForReconcileRequest(2); + h.resolveReconcileSnapshotAt(2, h.authoritativeSnapshot); + await refreshed; + await h.seeded; + + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.status])).toEqual([ + ['agent-stale', 'completed'], + ]); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('retries a pending reconcile after the initial snapshot request recovers', async () => { + const h = await setupReconcileHarness(); + vi.useFakeTimers(); + try { + h.getSessionSnapshot.mockRejectedValueOnce(new Error('offline')); + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await Promise.resolve(); + + expect(h.getSessionSnapshot).toHaveBeenCalledTimes(2); + await vi.advanceTimersByTimeAsync(1000); + await h.reconcileRequested; + h.resolveReconcileSnapshot(h.authoritativeSnapshot); + await h.seeded; + + expect(h.client.activeAppTasks.value.map((task) => [task.id, task.status])).toEqual([ + ['agent-stale', 'completed'], + ]); + } finally { + h.connection.close(); + vi.clearAllTimers(); + vi.useRealTimers(); + vi.unstubAllGlobals(); + } + }); + + it('does not reconcile a BTW side-channel row at the main turn boundary', async () => { + const h = await setupReconcileHarness(); + try { + await h.client.openSideChat(); + h.handlers.onEvent( + { + type: 'taskCreated', + sessionId: h.sessionId, + task: { ...h.staleRow, id: 'btw-1' }, + }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 12 }, + ); + await Promise.resolve(); + + expect(h.getSessionSnapshot).toHaveBeenCalledOnce(); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('does not request a second reconcile when a legacy quiet event follows turn end', async () => { + const h = await setupReconcileHarness(); + try { + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: true }, + { sessionId: h.sessionId, seq: 11 }, + ); + h.handlers.onEvent( + { type: 'taskCreated', sessionId: h.sessionId, task: h.staleRow }, + { sessionId: h.sessionId, seq: 12 }, + ); + h.handlers.onEvent( + { type: 'turnActiveChanged', sessionId: h.sessionId, active: false, reason: 'completed' }, + { sessionId: h.sessionId, seq: 13 }, + ); + await h.reconcileRequested; + h.handlers.onEvent( + { type: 'sessionWorkChanged', sessionId: h.sessionId, busy: false }, + { sessionId: h.sessionId, seq: 14 }, + ); + + h.resolveReconcileSnapshot(h.authoritativeSnapshot); + await h.seeded; + await Promise.resolve(); + await Promise.resolve(); + + expect(h.getSessionSnapshot).toHaveBeenCalledTimes(2); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); + + it('clears local waiting state when legacy quiet arrives without a turn-start marker', async () => { + const h = await setupReconcileHarness(); + try { + await h.client.sendPrompt('first prompt'); + expect(h.client.activity.value).toBe('running'); + + h.handlers.onEvent( + { type: 'sessionWorkChanged', sessionId: h.sessionId, busy: false }, + { sessionId: h.sessionId, seq: 11 }, + ); + + expect(h.client.inFlight.value).toBe(false); + expect(h.client.activity.value).toBe('idle'); + + await h.client.sendPrompt('second prompt'); + expect(h.submitPrompt).toHaveBeenCalledTimes(2); + } finally { + h.connection.close(); + vi.unstubAllGlobals(); + } + }); }); describe('isRenderEvent (queue classification)', () => { diff --git a/apps/kimi-web/test/lib-logic.test.ts b/apps/kimi-web/test/lib-logic.test.ts index 36f3f8eda7..56536fd803 100644 --- a/apps/kimi-web/test/lib-logic.test.ts +++ b/apps/kimi-web/test/lib-logic.test.ts @@ -9,7 +9,11 @@ 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 { keepLiveSubagents, mergeSnapshotSubagents } from '../src/lib/taskMerge'; +import { + isLiveForegroundSubagent, + keepLiveSubagents, + mergeSnapshotSubagents, +} from '../src/lib/taskMerge'; import { normalizeToolName, toolSummary } from '../src/lib/toolMeta'; import { collapsePrompt, humanizeCron } from '../src/lib/cronHumanize'; import { @@ -747,9 +751,61 @@ describe('mergeSnapshotSubagents', () => { expect(merged.map((t) => t.id)).toEqual(['a1', 'bash-1']); }); - it('returns the existing list untouched when the roster is empty', () => { + it('returns the existing list untouched when an older server omits the roster', () => { const existing = [subagent('a1')]; - expect(mergeSnapshotSubagents([], existing)).toBe(existing); + expect(mergeSnapshotSubagents(undefined, existing)).toBe(existing); + }); + + it('uses the authoritative roster terminal state for an existing live row', () => { + const existing = [subagent('a1', { status: 'running', subagentPhase: 'working' })]; + const roster = [subagent('a1', { status: 'completed', subagentPhase: 'completed' })]; + + expect(mergeSnapshotSubagents(roster, existing)[0]).toMatchObject({ + id: 'a1', + status: 'completed', + subagentPhase: 'completed', + }); + }); + + it('removes stale live foreground rows when an authoritative roster omits them', () => { + const existing = [ + subagent('a1', { status: 'running', subagentPhase: 'working' }), + subagent('a2', { status: 'running', subagentPhase: 'suspended' }), + subagent('a3', { status: 'running', runInBackground: true }), + subagent('a4', { status: 'running', backgroundTaskId: 'task-1' }), + subagent('a5', { status: 'completed', subagentPhase: 'completed' }), + ]; + const merged = mergeSnapshotSubagents([], existing); + + expect(merged.map((task) => task.id)).toEqual(['a3', 'a4', 'a5']); + }); + + it('keeps a roster-unowned side-channel row when the authoritative roster omits it', () => { + const sideChannel = subagent('btw-1', { + status: 'running', + subagentPhase: 'working', + }); + + expect(mergeSnapshotSubagents([], [sideChannel], 'btw-1')).toEqual([sideChannel]); + }); +}); + +describe('isLiveForegroundSubagent', () => { + const task = (overrides: Partial = {}): AppTask => ({ + id: 'agent-1', + sessionId: 's1', + kind: 'subagent', + description: 'Explore', + status: 'running', + createdAt: '2026-01-01T00:00:00.000Z', + ...overrides, + }); + + it('selects only running foreground rows for turn-end reconciliation', () => { + expect(isLiveForegroundSubagent(task())).toBe(true); + expect(isLiveForegroundSubagent(task({ runInBackground: true }))).toBe(false); + expect(isLiveForegroundSubagent(task({ backgroundTaskId: 'task-1' }))).toBe(false); + expect(isLiveForegroundSubagent(task({ status: 'completed' }))).toBe(false); }); }); diff --git a/apps/kimi-web/test/task-poller.test.ts b/apps/kimi-web/test/task-poller.test.ts index 0010d17595..29a7d43db4 100644 --- a/apps/kimi-web/test/task-poller.test.ts +++ b/apps/kimi-web/test/task-poller.test.ts @@ -1,7 +1,8 @@ // Scenario: terminal-output backfill for background tasks (useTaskPoller). // Responsibilities: folded background-subagent rows must receive the output -// fetched under their REST task id, and a transient getTask failure must not -// permanently suppress later backfills. +// fetched under their REST task id, a transient getTask failure must not +// permanently suppress later backfills, and successful list loads notify the +// consumer that delayed reconciliation may retry. // Wiring: the composable is real; daemon requests are stubbed. // Run: pnpm --filter @moonshot-ai/kimi-web exec vitest run test/task-poller.test.ts @@ -118,4 +119,26 @@ describe('useTaskPoller terminal-output backfill', () => { expect(apiMock.getTask).toHaveBeenCalledTimes(2); expect(state.tasksBySession['sess_1']?.[0]?.outputPreview).toBe('final result'); }); + + it('notifies the consumer after the task list loads successfully', async () => { + const state = createState([]); + const onTaskListLoaded = vi.fn(); + apiMock.listTasks.mockResolvedValue([]); + + const poller = useTaskPoller(state, computed(() => []), onTaskListLoaded); + await poller.loadTasksForSession('sess_1'); + + expect(onTaskListLoaded).toHaveBeenCalledWith('sess_1'); + }); + + it('does not notify the consumer when the task list request fails', async () => { + const state = createState([]); + const onTaskListLoaded = vi.fn(); + apiMock.listTasks.mockRejectedValue(new Error('offline')); + + const poller = useTaskPoller(state, computed(() => []), onTaskListLoaded); + await poller.loadTasksForSession('sess_1'); + + expect(onTaskListLoaded).not.toHaveBeenCalled(); + }); });