From 449f28b1e5a86b25c37384105a9c6e0521dcdeed Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Andr=C3=A9=20Padez?= Date: Mon, 27 Jul 2026 00:00:11 +0000 Subject: [PATCH] =?UTF-8?q?chat:=20persistent=20Agent=20SDK=20session=20pe?= =?UTF-8?q?r=20chat=20=E2=80=94=20decouple=20worker=20from=20turn=20(Phase?= =?UTF-8?q?=201)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Root fix for orphaned background tasks: the platform drove Claude Code as a one-shot `claude -p` per turn (stdin ignored, process exits at turn end), so run_in_background/Monitor work — and its task_notification — had no live harness to return to. Now each chat session runs ONE long-lived Agent SDK query() with streaming input; turns are user messages pushed onto it, and the session stays warm between turns. - claude-manager: persistent `query({ prompt: AsyncIterable, options })` per sessionKey (bypassPermissions, --resume, mcp via extraArgs, CLAUDECODE stripped). Single consumer loop maps every SDK message → ChatEvent, incl. post-turn task_started / task_notification. interrupt() = stop-turn; abort() = kill-session; 30-min idle GC. - stream-parser: processMessage() (object-level, reused by the SDK loop) + task message handling. ChatEvent/ServerMessage gain task:started / task:notification. - API: the sidecar event subscription is now SESSION-scoped (no longer unsubscribes on 'result'), so background events after turn-end still reach the client. First turn opens the session; later turns push onto it. handleStop → interrupt (keeps session warm); disconnect/deleteSession → kill. - protocol/sidecar-registry/user-instance: claude:interrupt command + interruptClaude. - client: render task:started / task:notification in the transcript. Verified end-to-end through the real chat WS: a run_in_background task's completion arrives ~6s AFTER the turn's result; multi-turn on one warm session works. Co-Authored-By: Claude Opus 4.8 --- src/servers/api/chat/types.ts | 10 +- src/servers/api/chat/websocket.ts | 62 ++-- src/servers/channels/send-claude-code.ts | 13 +- src/servers/sidecar-registry.ts | 6 + src/servers/sidecar/claude/claude-manager.ts | 288 ++++++++++++------ src/servers/sidecar/claude/stream-parser.ts | 83 +++-- src/servers/sidecar/claude/user-instance.ts | 5 + src/servers/sidecar/protocol.ts | 2 + .../officerdev/src/apps/Chat/types.ts | 4 +- .../officerdev/src/hooks/useChat.ts | 11 + 10 files changed, 329 insertions(+), 155 deletions(-) diff --git a/src/servers/api/chat/types.ts b/src/servers/api/chat/types.ts index 6201291a..5cf9265c 100644 --- a/src/servers/api/chat/types.ts +++ b/src/servers/api/chat/types.ts @@ -126,7 +126,9 @@ export type ServerMessage = | { // Ack for a client 'disconnect': the session was torn down server-side. type: 'disconnected'; - }; + } + | { type: 'task:started'; taskId: string; description: string; taskType?: string } + | { type: 'task:notification'; taskId: string; status: 'completed' | 'failed' | 'stopped'; summary: string }; export type ChatEvent = | { type: 'text'; text: string } @@ -148,7 +150,11 @@ export type ChatEvent = cost: MessageCost; } | { type: 'error'; message: string } - | { type: 'stopped' }; + | { type: 'stopped' } + // Background-task lifecycle (run_in_background / Monitor), delivered in-stream by the persistent + // session — including AFTER the turn's `result`, which is the whole point of the persistent worker. + | { type: 'task:started'; taskId: string; description: string; taskType?: string } + | { type: 'task:notification'; taskId: string; status: 'completed' | 'failed' | 'stopped'; summary: string }; export type UserSession = { sessionId: string; diff --git a/src/servers/api/chat/websocket.ts b/src/servers/api/chat/websocket.ts index d6a17d7d..e6861c40 100644 --- a/src/servers/api/chat/websocket.ts +++ b/src/servers/api/chat/websocket.ts @@ -275,6 +275,18 @@ function createEventHandler(sessionId: string, model: string, cwd: string) { session.isGenerating = false; break; } + + case 'task:started': { + // Background task launched (run_in_background / Monitor). Independent of turn state. + sendToClient(ws, { type: 'task:started', taskId: event.taskId, description: event.description, taskType: event.taskType }); + break; + } + + case 'task:notification': { + // Background task finished — arrives AFTER the turn's 'result'; the whole point of the fix. + sendToClient(ws, { type: 'task:notification', taskId: event.taskId, status: event.status, summary: event.summary }); + break; + } } }; } @@ -372,21 +384,35 @@ async function handleClaudeCodeChat( const onEvent = createEventHandler(sessionId, model, cwd); try { - const handle = await sendClaudeCodeStreaming({ - userId, - email, - username, - prompt: effectivePrompt, - sessionKey: sessionId, - cwd, - model, - resumeSessionId: msg.resumeSessionId, - onEvent, - }); - - // Store sentinel so handleStop can kill it via sidecar - session.piProcess = sessionId as any; - session._claudeKill = handle.kill; + if (!session._claudeKill) { + // First turn of this session: open the persistent session + a SESSION-scoped event subscription + // (survives turn-end so background task:notifications keep flowing). handle.kill tears both down. + const handle = await sendClaudeCodeStreaming({ + userId, + email, + username, + prompt: effectivePrompt, + sessionKey: sessionId, + cwd, + model, + resumeSessionId: msg.resumeSessionId, + onEvent, + }); + session.piProcess = sessionId as any; + session._claudeKill = handle.kill; + } else { + // Session already live: push this turn onto the existing persistent session (no new subscription). + await sidecar.spawnClaudeStreaming({ + userId, + email, + username, + prompt: effectivePrompt, + sessionKey: sessionId, + cwd, + model, + resumeSessionId: msg.resumeSessionId, + }); + } } catch (err) { logger.error('Failed to start Claude Code streaming', { sessionId, error: String(err) }); sendToClient(ws, { type: 'error', message: 'Failed to start Claude Code' }); @@ -524,8 +550,10 @@ async function handleStop(ws: ServerWebSocket): Promise { if (session?.piProcess) { try { if (isClaudeModel(session.model)) { - sidecar.killClaude(sessionId, session.email); - logger.info('Killed Claude Code process via sidecar', { sessionId }); + // Interrupt the current turn but KEEP the persistent session alive (background tasks + the + // warm worker survive). Full teardown is 'disconnect' → deleteSession → _claudeKill. + void sidecar.interruptClaude(sessionId, session.email); + logger.info('Interrupted Claude Code turn via sidecar (session stays warm)', { sessionId }); } else { session._claudeKill?.(); // OpenCode: abort the turn via the stored handle logger.info('Aborted OpenCode turn', { sessionId }); diff --git a/src/servers/channels/send-claude-code.ts b/src/servers/channels/send-claude-code.ts index 98703746..639ca7c8 100644 --- a/src/servers/channels/send-claude-code.ts +++ b/src/servers/channels/send-claude-code.ts @@ -50,16 +50,11 @@ export async function sendClaudeCodeStreaming(params: ClaudeCodeStreamingParams) logger.info('Claude Code streaming exec (via sidecar)', { sessionKey: params.sessionKey }); - // Subscribe to events for this session + // Session-scoped subscription. The persistent session outlives the turn, so background task events + // (task:notification) arrive AFTER 'result' — do NOT unsubscribe on a terminal turn event; only on + // an explicit kill/teardown (the returned handle, called from deleteSession/disconnect). const unsub = sidecar.onClaudeEvent((sessionKey, event) => { - if (sessionKey === params.sessionKey) { - onEvent(event); - - // Unsubscribe when we get a terminal event - if (event.type === 'result' || event.type === 'error') { - unsub(); - } - } + if (sessionKey === params.sessionKey) onEvent(event); }); await sidecar.spawnClaudeStreaming(spawnParams); diff --git a/src/servers/sidecar-registry.ts b/src/servers/sidecar-registry.ts index 28cff231..75b41a2d 100644 --- a/src/servers/sidecar-registry.ts +++ b/src/servers/sidecar-registry.ts @@ -326,6 +326,12 @@ export function killClaude(sessionKey: string, email: string): void { if (sc) sendFireToSidecar(sc, { type: 'claude:kill', id: nextId(), sessionKey }); } +// Interrupt the current turn but keep the persistent session warm (the "stop" button). +export function interruptClaude(sessionKey: string, email: string): void { + const sc = findSidecarByName(`claude:${email}`); + if (sc) sendFireToSidecar(sc, { type: 'claude:interrupt', id: nextId(), sessionKey }); +} + export function clearClaudeSession(sessionKey: string, email?: string): void { if (email) { const sc = findSidecarByName(`claude:${email}`); diff --git a/src/servers/sidecar/claude/claude-manager.ts b/src/servers/sidecar/claude/claude-manager.ts index f512ba89..65edf57a 100644 --- a/src/servers/sidecar/claude/claude-manager.ts +++ b/src/servers/sidecar/claude/claude-manager.ts @@ -1,9 +1,9 @@ import { join } from 'node:path'; -import type { Subprocess } from 'bun'; +import { query, type Query } from '@anthropic-ai/claude-agent-sdk'; import type { ChatEvent } from '../../api/chat/types'; import type { ClaudeSpawnParams, ClaudeSpawnStreamingParams, ClaudeCodeResult } from '../protocol'; import { setClaudeSession, clearClaudeSession, getClaudeSession } from './state'; -import { parseStream } from './stream-parser'; +import { createParseState, processMessage } from './stream-parser'; const SEND_TIMEOUT_MS = 30 * 60 * 1000; @@ -15,8 +15,8 @@ const CLAUDE_BIN = '/usr/local/bin/claude'; const HOST_HOME = process.env.HOME!; const DATA_PATH = process.env.DATA_PATH ?? join(process.cwd(), 'data'); -// Active streaming processes -const activeProcs = new Map(); +// Tear a persistent session down after this long with no new turn (see PersistentSession below). +const IDLE_TIMEOUT_MS = 30 * 60 * 1000; // MCP config paths, set by user-instance at startup let mcpHostPath: string | undefined; // path on the host filesystem @@ -25,7 +25,7 @@ export function setMcpConfigPath(hostPath: string): void { mcpHostPath = hostPath; } -// ── Blocking send ── +// ── Blocking send (one-shot; used by the non-streaming 'claude:spawn' command) ── type ClaudeCodeOutput = { result: string; @@ -37,7 +37,7 @@ type ClaudeCodeOutput = { }; export async function spawnClaude(params: ClaudeSpawnParams): Promise { - const { prompt, sessionKey, email } = params; + const { prompt, sessionKey } = params; const existingSession = getClaudeSession(sessionKey); @@ -54,8 +54,6 @@ export async function spawnClaude(params: ClaudeSpawnParams): Promise void; + closeInput: () => void; + abort: AbortController; + emit: (event: ChatEvent) => void; + isGenerating: boolean; + idleTimer?: ReturnType; +}; + +const sessions = new Map(); + +/** A hand-rolled async iterable we can push turns onto and close on teardown. */ +function makeInputQueue() { + const buf: SdkUserMessage[] = []; + let wake: (() => void) | null = null; + let closed = false; + async function* gen(): AsyncGenerator { + while (true) { + if (buf.length) { + yield buf.shift()!; + continue; + } + if (closed) return; + await new Promise((r) => { + wake = r; + }); + } + } + return { + gen: gen(), + push(m: SdkUserMessage) { + buf.push(m); + wake?.(); + wake = null; + }, + close() { + closed = true; + wake?.(); + wake = null; + }, + }; +} + +function armIdle(session: PersistentSession): void { + if (session.idleTimer) clearTimeout(session.idleTimer); + session.idleTimer = setTimeout(() => { + killClaudeSession(session.sessionKey); + }, IDLE_TIMEOUT_MS); +} + +function createSession(params: ClaudeSpawnStreamingParams, onEvent: (event: ChatEvent) => void): PersistentSession { + const { sessionKey } = params; + const input = makeInputQueue(); + const abort = new AbortController(); + + const session: PersistentSession = { + sessionKey, + query: undefined as unknown as Query, + pushTurn: () => {}, + closeInput: () => input.close(), + abort, + emit: onEvent, + isGenerating: false, + }; + + // Strip the nested-session guard vars so the SDK can spawn `claude` (mirrors the old spawn env clean). + const { CLAUDECODE: _c, CLAUDE_CODE_ENTRYPOINT: _e, CLAUDE_CODE_SSE_PORT: _s, ...cleanEnv } = process.env; + + const resumeId = getClaudeSession(sessionKey) ?? params.resumeSessionId; + const subModel = params.model?.split('/')[1]; + + const q = query({ + prompt: input.gen as AsyncIterable, + options: { + cwd: params.cwd ?? HOST_HOME, + permissionMode: 'bypassPermissions', + allowDangerouslySkipPermissions: true, + includePartialMessages: true, + abortController: abort, + pathToClaudeCodeExecutable: CLAUDE_BIN, + settingSources: ['user', 'project', 'local'], + env: cleanEnv as Record, + stderr: (d: string) => { + if (d.trim()) console.error(`[claude:stream:${sessionKey}] ${d.slice(0, 300)}`); + }, + ...(subModel ? { model: subModel } : {}), + ...(resumeId ? { resume: resumeId } : {}), + ...(mcpHostPath ? { extraArgs: { 'mcp-config': mcpHostPath } } : {}), + }, + }); + + session.query = q; + session.pushTurn = (prompt: string) => { + if (session.idleTimer) clearTimeout(session.idleTimer); + session.isGenerating = true; + input.push({ type: 'user', message: { role: 'user', content: prompt }, parent_tool_use_id: null, session_id: sessionKey }); + }; + + sessions.set(sessionKey, session); + + // Single consumer loop for the session's whole life. Turn-end (`result`) and errors flip isGenerating + // and (re)arm the idle timer; the session process stays alive so later task_notifications still flow. + void (async () => { + const state = createParseState(); + const emit = (event: ChatEvent) => { + if (event.type === 'result' || event.type === 'error' || event.type === 'stopped') { + session.isGenerating = false; + armIdle(session); + } + session.emit(event); + }; + try { + for await (const msg of q as AsyncGenerator>) { + processMessage(msg, state, { + onEvent: emit, + onSessionId: (id: string) => setClaudeSession(sessionKey, id), + }); + } + } catch (err) { + if (!abort.signal.aborted) { + session.emit({ type: 'error', message: err instanceof Error ? err.message : String(err) }); + } + } finally { + if (session.idleTimer) clearTimeout(session.idleTimer); + sessions.delete(sessionKey); + } + })(); + + return session; +} + +/** Start a turn: create the persistent session if needed, then push the prompt as a user message. */ export async function spawnClaudeStreaming( params: ClaudeSpawnStreamingParams, onEvent: (event: ChatEvent) => void, ): Promise { - const { prompt, sessionKey, email } = params; - - const existingSession = getClaudeSession(sessionKey); - - const claudeArgs = [ - CLAUDE_BIN, - '-p', - prompt, - '--dangerously-skip-permissions', - '--output-format', - 'stream-json', - '--verbose', - '--include-partial-messages', - ]; - - const { CLAUDECODE: _, ...cleanEnv } = process.env; - const mcpConfig = mcpHostPath; - if (mcpConfig) claudeArgs.push('--mcp-config', mcpConfig); - - const subModel = params.model?.split('/')[1]; - if (subModel) claudeArgs.push('--model', subModel); - - // Resume: an in-memory mapping (subsequent turns of a live chat) takes precedence; otherwise a - // caller-supplied session uuid (reopening a session from the /chat list) resumes Claude's transcript. - const resumeId = existingSession ?? params.resumeSessionId; - if (resumeId) { - claudeArgs.push('--resume', resumeId); - if (!existingSession) setClaudeSession(sessionKey, resumeId); + let session = sessions.get(params.sessionKey); + if (session) { + session.emit = onEvent; // adopt the latest emitter (equivalent across turns; keeps events flowing) + } else { + session = createSession(params, onEvent); } + session.pushTurn(params.prompt); +} - const spawnCmd = claudeArgs; - // Honor a caller-provided working directory (e.g. /chat runs from a dedicated general_chat_sessions - // dir); fall back to the owner's host home. - const spawnCwd = params.cwd ?? HOST_HOME; - - const proc = Bun.spawn(spawnCmd, { - stdin: 'ignore', - stdout: 'pipe', - stderr: 'pipe', - cwd: spawnCwd, - env: cleanEnv as Record, - }); - - activeProcs.set(sessionKey, proc); - - const timeout = setTimeout(() => { - try { - proc.kill(); - } catch { - /* already dead */ - } - onEvent({ type: 'error', message: 'Claude Code timed out after 5 minutes' }); - }, SEND_TIMEOUT_MS); - - // Process NDJSON stream +/** Interrupt the current turn but KEEP the session alive (the "stop" button). */ +export async function interruptClaudeSession(sessionKey: string): Promise { + const session = sessions.get(sessionKey); + if (!session) return false; try { - const stdout = proc.stdout as ReadableStream; - const callbacks = { - onEvent, - onSessionId: (sessionId: string) => setClaudeSession(sessionKey, sessionId), - }; - - const state = await parseStream(stdout, callbacks); - - clearTimeout(timeout); - - if (!state.gotResult) { - const exitCode = await proc.exited; - const stderr = await new Response(proc.stderr as ReadableStream).text(); - if (state.textBuffer) { - onEvent({ type: 'text', text: state.textBuffer }); - } - if (exitCode !== 0) { - onEvent({ type: 'error', message: `Claude Code exited with code ${exitCode}: ${stderr.trim().slice(0, 200)}` }); - } else { - onEvent({ type: 'result', cost: { inputTokens: 0, outputTokens: 0, totalUSD: 0 } }); - } - } - } catch (err) { - clearTimeout(timeout); - onEvent({ type: 'error', message: String(err) }); - } finally { - activeProcs.delete(sessionKey); + await session.query.interrupt(); + session.isGenerating = false; + return true; + } catch { + return false; } } +/** Fully tear the session down (the "disconnect" action / idle GC): abort the query + close input. */ export function killClaudeSession(sessionKey: string): boolean { - const proc = activeProcs.get(sessionKey); - if (proc) { - try { - proc.kill(); - } catch { - /* already dead */ - } - activeProcs.delete(sessionKey); - return true; + const session = sessions.get(sessionKey); + if (!session) return false; + if (session.idleTimer) clearTimeout(session.idleTimer); + try { + session.abort.abort(); + } catch { + /* already aborted */ } - return false; + try { + session.closeInput(); + } catch { + /* noop */ + } + sessions.delete(sessionKey); + return true; } export function clearSession(sessionKey: string): void { @@ -230,5 +318,5 @@ export function clearSession(sessionKey: string): void { } export function getActiveSessionKeys(): string[] { - return Array.from(activeProcs.keys()); + return Array.from(sessions.keys()); } diff --git a/src/servers/sidecar/claude/stream-parser.ts b/src/servers/sidecar/claude/stream-parser.ts index cb43f59e..2790ff4a 100644 --- a/src/servers/sidecar/claude/stream-parser.ts +++ b/src/servers/sidecar/claude/stream-parser.ts @@ -118,6 +118,62 @@ function handleResult( callbacks.onEvent({ type: 'result', cost }); } +function handleSystem(msg: Record, callbacks: StreamParserCallbacks): void { + const subtype = msg.subtype as string | undefined; + if (subtype === 'init') { + const sessionId = msg.session_id as string | undefined; + if (sessionId) callbacks.onSessionId(sessionId); + } else if (subtype === 'task_started') { + callbacks.onEvent({ + type: 'task:started', + taskId: (msg.task_id as string) ?? '', + description: (msg.description as string) ?? '', + taskType: msg.task_type as string | undefined, + }); + } else if (subtype === 'task_notification') { + callbacks.onEvent({ + type: 'task:notification', + taskId: (msg.task_id as string) ?? '', + status: (msg.status as 'completed' | 'failed' | 'stopped') ?? 'completed', + summary: (msg.summary as string) ?? '', + }); + } +} + +/** + * Map a single already-parsed message object (a Claude Code stream-json message, or the equivalent + * Agent SDK message — same shapes) to ChatEvents. Returns false if the message type was ignored. + */ +export function processMessage( + msg: Record, + state: ParseState, + callbacks: StreamParserCallbacks, +): boolean { + const type = msg.type as string; + + switch (type) { + case 'stream_event': + handleStreamEvent(msg, state, callbacks.onEvent); + break; + case 'assistant': + handleAssistant(msg, state, callbacks.onEvent); + break; + case 'user': + handleUser(msg, callbacks.onEvent); + break; + case 'system': + handleSystem(msg, callbacks); + break; + case 'result': + handleResult(msg, state, callbacks); + break; + default: + return false; + } + + return true; +} + /** * Process a single NDJSON line from Claude Code's stream output. * Returns false if the line was skipped (empty or malformed), true otherwise. @@ -136,32 +192,7 @@ export function processLine( return false; } - const type = msg.type as string; - - switch (type) { - case 'stream_event': - handleStreamEvent(msg, state, callbacks.onEvent); - break; - case 'assistant': - handleAssistant(msg, state, callbacks.onEvent); - break; - case 'user': - handleUser(msg, callbacks.onEvent); - break; - case 'system': - if (msg.subtype === 'init') { - const sessionId = msg.session_id as string | undefined; - if (sessionId) callbacks.onSessionId(sessionId); - } - break; - case 'result': - handleResult(msg, state, callbacks); - break; - default: - return false; - } - - return true; + return processMessage(msg, state, callbacks); } /** diff --git a/src/servers/sidecar/claude/user-instance.ts b/src/servers/sidecar/claude/user-instance.ts index 0e5c0f91..59e9c5b3 100644 --- a/src/servers/sidecar/claude/user-instance.ts +++ b/src/servers/sidecar/claude/user-instance.ts @@ -137,6 +137,11 @@ async function handleCommand(cmd: SidecarCommand, reply: ReplyFn) { reply({ type: 'claude:killed', id: cmd.id }); break; + case 'claude:interrupt': + await claudeManager.interruptClaudeSession(cmd.sessionKey); + reply({ type: 'claude:interrupted', id: cmd.id }); + break; + case 'claude:clear-session': claudeManager.clearSession(cmd.sessionKey); reply({ type: 'claude:session-cleared', id: cmd.id }); diff --git a/src/servers/sidecar/protocol.ts b/src/servers/sidecar/protocol.ts index 880f8c2b..b8011117 100644 --- a/src/servers/sidecar/protocol.ts +++ b/src/servers/sidecar/protocol.ts @@ -15,6 +15,7 @@ export type SidecarCommand = | { type: 'claude:spawn'; id: string; params: ClaudeSpawnParams } | { type: 'claude:spawn-streaming'; id: string; params: ClaudeSpawnStreamingParams } | { type: 'claude:kill'; id: string; sessionKey: string } + | { type: 'claude:interrupt'; id: string; sessionKey: string } | { type: 'claude:clear-session'; id: string; sessionKey: string } // OpenCode — drive a turn via `opencode run … --format json` (tools re-anchored to cwd via --dir) | { type: 'opencode:run-streaming'; id: string; params: OpenCodeRunParams } @@ -38,6 +39,7 @@ export type SidecarEvent = | { type: 'claude:result'; id: string; result: ClaudeCodeResult } | { type: 'claude:error'; id: string; error: string } | { type: 'claude:killed'; id: string } + | { type: 'claude:interrupted'; id: string } | { type: 'claude:session-cleared'; id: string } // VNC | { type: 'vnc:started'; id: string; port: number; display: number } diff --git a/src/workspaces/officerdev/src/apps/Chat/types.ts b/src/workspaces/officerdev/src/apps/Chat/types.ts index 5ce9b6a4..bfd3e04f 100644 --- a/src/workspaces/officerdev/src/apps/Chat/types.ts +++ b/src/workspaces/officerdev/src/apps/Chat/types.ts @@ -50,7 +50,9 @@ export type ServerMessage = | { type: 'sync:messages'; sessionId: string; messages: Message[]; isGenerating: boolean; streamingText: string } | { type: 'error'; message: string; errorCode?: string } | { type: 'stopped' } - | { type: 'disconnected' }; + | { type: 'disconnected' } + | { type: 'task:started'; taskId: string; description: string; taskType?: string } + | { type: 'task:notification'; taskId: string; status: 'completed' | 'failed' | 'stopped'; summary: string }; export type Message = { id: string; diff --git a/src/workspaces/officerdev/src/hooks/useChat.ts b/src/workspaces/officerdev/src/hooks/useChat.ts index f81de143..577527dd 100644 --- a/src/workspaces/officerdev/src/hooks/useChat.ts +++ b/src/workspaces/officerdev/src/hooks/useChat.ts @@ -209,6 +209,17 @@ export function useChat(initialSessionId?: string, initialModel?: string | null, commitStreaming(); setIsGenerating(false); break; + + case 'task:started': + setMessages((prev) => [...prev, { role: 'assistant', id: crypto.randomUUID(), text: `⏳ Background task started — ${msg.description}` }]); + break; + + case 'task:notification': { + // The fix in action: a background task's completion arriving after the turn ended. + const icon = msg.status === 'completed' ? '✅' : msg.status === 'failed' ? '❌' : '⏹️'; + setMessages((prev) => [...prev, { role: 'assistant', id: crypto.randomUUID(), text: `${icon} Background task ${msg.status} — ${msg.summary}` }]); + break; + } } }