diff --git a/package-lock.json b/package-lock.json index 6cac693..db992a0 100644 --- a/package-lock.json +++ b/package-lock.json @@ -1,12 +1,12 @@ { "name": "codepilot", - "version": "0.17.3", + "version": "0.17.4", "lockfileVersion": 3, "requires": true, "packages": { "": { "name": "codepilot", - "version": "0.17.3", + "version": "0.17.4", "dependencies": { "@ai-sdk/anthropic": "^3.0.47", "@ai-sdk/google": "^3.0.31", diff --git a/package.json b/package.json index 1eaadab..1b42d08 100644 --- a/package.json +++ b/package.json @@ -1,6 +1,6 @@ { "name": "codepilot", - "version": "0.17.3", + "version": "0.17.4", "private": true, "author": { "name": "op7418", diff --git a/src/app/api/providers/models/route.ts b/src/app/api/providers/models/route.ts index d40eed1..03dc410 100644 --- a/src/app/api/providers/models/route.ts +++ b/src/app/api/providers/models/route.ts @@ -73,16 +73,15 @@ export async function GET() { const providers = getAllProviders(); const groups: ProviderModelGroup[] = []; - // Check for environment variables - const hasEnvKey = !!(process.env.ANTHROPIC_API_KEY || process.env.ANTHROPIC_AUTH_TOKEN); - if (hasEnvKey) { - groups.push({ - provider_id: 'env', - provider_name: 'Environment', - provider_type: 'anthropic', - models: DEFAULT_MODELS, - }); - } + // Always show the built-in Claude Code provider group. + // Claude Code CLI stores credentials in ~/.claude/ (via `claude login`), + // which the SDK subprocess can read — even without ANTHROPIC_API_KEY in env. + groups.push({ + provider_id: 'env', + provider_name: 'Claude Code', + provider_type: 'anthropic', + models: DEFAULT_MODELS, + }); // Provider types that are not LLMs (e.g. image generation) — skip in chat model selector const MEDIA_PROVIDER_TYPES = new Set(['gemini-image']); @@ -102,16 +101,6 @@ export async function GET() { }); } - // If no groups at all (no env, no providers), show default Anthropic group - if (groups.length === 0) { - groups.push({ - provider_id: 'env', - provider_name: 'Anthropic', - provider_type: 'anthropic', - models: DEFAULT_MODELS, - }); - } - // Determine default provider const defaultProviderId = getDefaultProviderId() || groups[0].provider_id; diff --git a/src/components/chat/ChatView.tsx b/src/components/chat/ChatView.tsx index 46f3a50..ecee208 100644 --- a/src/components/chat/ChatView.tsx +++ b/src/components/chat/ChatView.tsx @@ -1,24 +1,20 @@ 'use client'; import { useState, useCallback, useEffect, useRef } from 'react'; -import type { Message, MessagesResponse, PermissionRequestEvent, FileAttachment } from '@/types'; +import type { Message, MessagesResponse, FileAttachment, SessionStreamSnapshot } from '@/types'; import { MessageList } from './MessageList'; import { MessageInput } from './MessageInput'; import { usePanel } from '@/hooks/usePanel'; -import { consumeSSEStream } from '@/hooks/useSSEStream'; import { BatchExecutionDashboard, BatchContextSync } from './batch-image-gen'; import { setLastGeneratedImages, transferPendingToMessage } from '@/lib/image-ref-store'; - -interface ToolUseInfo { - id: string; - name: string; - input: unknown; -} - -interface ToolResultInfo { - tool_use_id: string; - content: string; -} +import { + startStream, + stopStream, + subscribe, + getSnapshot, + respondToPermission, + clearSnapshot, +} from '@/lib/stream-session-manager'; interface ChatViewProps { sessionId: string; @@ -35,18 +31,29 @@ export function ChatView({ sessionId, initialMessages = [], initialHasMore = fal const [hasMore, setHasMore] = useState(initialHasMore); const [loadingMore, setLoadingMore] = useState(false); const loadingMoreRef = useRef(false); - const [streamingContent, setStreamingContent] = useState(''); - const [isStreaming, setIsStreaming] = useState(false); - const [toolUses, setToolUses] = useState([]); - const [toolResults, setToolResults] = useState([]); - const [statusText, setStatusText] = useState(); const [mode, setMode] = useState(initialMode || 'code'); const [currentModel, setCurrentModel] = useState(modelName || (typeof window !== 'undefined' ? localStorage.getItem('codepilot:last-model') : null) || 'sonnet'); const [currentProviderId, setCurrentProviderId] = useState(providerId || (typeof window !== 'undefined' ? localStorage.getItem('codepilot:last-provider-id') : null) || ''); - const [pendingPermission, setPendingPermission] = useState(null); - const [permissionResolved, setPermissionResolved] = useState<'allow' | 'deny' | null>(null); - const [streamingToolOutput, setStreamingToolOutput] = useState(''); - const toolTimeoutRef = useRef<{ toolName: string; elapsedSeconds: number } | null>(null); + + // Stream snapshot from the manager — drives all streaming UI + const [streamSnapshot, setStreamSnapshot] = useState( + () => getSnapshot(sessionId) + ); + + // Derive rendering state from snapshot (backward-compatible with MessageList props) + const isStreaming = streamSnapshot?.phase === 'active'; + const streamingContent = streamSnapshot?.streamingContent ?? ''; + const toolUses = streamSnapshot?.toolUses ?? []; + const toolResults = streamSnapshot?.toolResults ?? []; + const streamingToolOutput = streamSnapshot?.streamingToolOutput ?? ''; + const statusText = streamSnapshot?.statusText; + const pendingPermission = streamSnapshot?.pendingPermission ?? null; + const permissionResolved = streamSnapshot?.permissionResolved ?? null; + + // Pending image generation notices — flushed into the next user message so the LLM knows about generated images + const pendingImageNoticesRef = useRef([]); + // Ref for sendMessage to allow self-referencing in timeout auto-retry + const sendMessageRef = useRef<(content: string, files?: FileAttachment[]) => Promise>(undefined); const handleModeChange = useCallback((newMode: string) => { setMode(newMode); @@ -74,60 +81,68 @@ export function ChatView({ sessionId, initialMessages = [], initialHasMore = fal setCurrentModel(model); }, []); - const abortControllerRef = useRef(null); - - // Abort active stream on unmount (e.g., session switch via key={id} remount). - // This triggers the existing server-side cleanup chain: - // fetch abort → request.signal 'abort' → route abortController.abort() - // → collectStreamResponse catch block saves partial message to DB - // → permission-registry abort handler auto-denies pending permissions + // Subscribe to stream-session-manager for this session. + // On unmount we only unsubscribe — we do NOT abort the stream. useEffect(() => { - return () => { - if (abortControllerRef.current) { - abortControllerRef.current.abort(); - abortControllerRef.current = null; + // Restore snapshot if stream is already active (e.g., user switched away and back) + const existing = getSnapshot(sessionId); + if (existing) { + setStreamSnapshot(existing); + if (existing.phase === 'active') { + setStreamingSessionId(sessionId); } - setStreamingSessionId(''); - setPendingApprovalSessionId(''); - }; - }, [setStreamingSessionId, setPendingApprovalSessionId]); - - // Warn before closing window/tab while streaming to prevent accidental data loss - useEffect(() => { - if (!isStreaming) return; - const handler = (e: BeforeUnloadEvent) => { - e.preventDefault(); - e.returnValue = ''; - }; - window.addEventListener('beforeunload', handler); - return () => window.removeEventListener('beforeunload', handler); - }, [isStreaming]); - - // Ref to keep accumulated streaming content in sync regardless of React batching - const accumulatedRef = useRef(''); - // Refs to track tool data reliably across closures (state reads can be stale) - const toolUsesRef = useRef([]); - const toolResultsRef = useRef([]); - // Ref for sendMessage to allow self-referencing in timeout auto-retry without circular deps - const sendMessageRef = useRef<(content: string, files?: FileAttachment[]) => Promise>(undefined); - // Pending image generation notices — flushed into the next user message so the LLM knows about generated images - const pendingImageNoticesRef = useRef([]); - - // Re-sync streaming content when the window regains visibility (Electron/browser tab switch) - useEffect(() => { - const handleVisibilityChange = () => { - if (document.visibilityState === 'visible' && accumulatedRef.current) { - setStreamingContent(accumulatedRef.current); + if (existing.pendingPermission && !existing.permissionResolved) { + setPendingApprovalSessionId(sessionId); } - }; - document.addEventListener('visibilitychange', handleVisibilityChange); - // Also handle Electron-specific focus events - window.addEventListener('focus', handleVisibilityChange); + } else { + setStreamSnapshot(null); + } + + const unsubscribe = subscribe(sessionId, (event) => { + setStreamSnapshot(event.snapshot); + + // Sync panel state + if (event.type === 'phase-changed') { + if (event.snapshot.phase === 'active') { + setStreamingSessionId(sessionId); + } else { + setStreamingSessionId(''); + setPendingApprovalSessionId(''); + } + } + if (event.type === 'permission-request') { + setPendingApprovalSessionId(sessionId); + } + if (event.type === 'completed') { + setStreamingSessionId(''); + setPendingApprovalSessionId(''); + + // Append the final assistant message to the messages list + const finalContent = event.snapshot.finalMessageContent; + if (finalContent) { + const assistantMessage: Message = { + id: 'temp-assistant-' + Date.now(), + session_id: sessionId, + role: 'assistant', + content: finalContent, + created_at: new Date().toISOString(), + token_usage: event.snapshot.tokenUsage ? JSON.stringify(event.snapshot.tokenUsage) : null, + }; + // Transfer pending reference images to this message ID + transferPendingToMessage(assistantMessage.id); + setMessages((prev) => [...prev, assistantMessage]); + } + + // Clear the snapshot from the manager since we've consumed it + clearSnapshot(sessionId); + } + }); + return () => { - document.removeEventListener('visibilitychange', handleVisibilityChange); - window.removeEventListener('focus', handleVisibilityChange); + unsubscribe(); + // Do NOT abort — stream continues in the manager }; - }, []); + }, [sessionId, setStreamingSessionId, setPendingApprovalSessionId]); const initializedRef = useRef(false); useEffect(() => { @@ -172,55 +187,21 @@ export function ChatView({ sessionId, initialMessages = [], initialHasMore = fal } }, [sessionId, messages, hasMore]); + // Stop streaming — delegates to manager const stopStreaming = useCallback(() => { - abortControllerRef.current?.abort(); - abortControllerRef.current = null; - }, []); + stopStream(sessionId); + }, [sessionId]); - const handlePermissionResponse = useCallback(async (decision: 'allow' | 'allow_session' | 'deny', updatedInput?: Record) => { - if (!pendingPermission) return; - - const body: { permissionRequestId: string; decision: { behavior: 'allow'; updatedPermissions?: unknown[]; updatedInput?: Record } | { behavior: 'deny'; message?: string } } = { - permissionRequestId: pendingPermission.permissionRequestId, - decision: decision === 'deny' - ? { behavior: 'deny', message: 'User denied permission' } - : { - behavior: 'allow', - ...(decision === 'allow_session' && pendingPermission.suggestions - ? { updatedPermissions: pendingPermission.suggestions } - : {}), - ...(updatedInput ? { updatedInput } : {}), - }, - }; - - setPermissionResolved(decision === 'deny' ? 'deny' : 'allow'); - setPendingApprovalSessionId(''); - - try { - await fetch('/api/chat/permission', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify(body), - }); - } catch { - // Best effort - the stream will handle timeout - } - - // Clear permission state after a short delay so user sees the feedback. - // Only clear if no new permission request has arrived in the meantime. - const answeredId = pendingPermission.permissionRequestId; - setTimeout(() => { - setPendingPermission((current) => { - if (current?.permissionRequestId === answeredId) { - // Same request — safe to clear both - setPermissionResolved(null); - return null; - } - return current; // A new request arrived — keep it - }); - }, 1000); - }, [pendingPermission, setPendingApprovalSessionId]); + // Permission response — delegates to manager + const handlePermissionResponse = useCallback( + async (decision: 'allow' | 'allow_session' | 'deny', updatedInput?: Record) => { + setPendingApprovalSessionId(''); + await respondToPermission(sessionId, decision, updatedInput); + }, + [sessionId, setPendingApprovalSessionId] + ); + // Send message — delegates stream management to the manager const sendMessage = useCallback( async (content: string, files?: FileAttachment[], systemPromptAppend?: string, displayOverride?: string) => { if (isStreaming) return; @@ -245,304 +226,35 @@ export function ChatView({ sessionId, initialMessages = [], initialHasMore = fal token_usage: null, }; setMessages((prev) => [...prev, userMessage]); - setIsStreaming(true); - setStreamingSessionId(sessionId); - setStreamingContent(''); - accumulatedRef.current = ''; - setToolUses([]); - setToolResults([]); - setStatusText(undefined); - const controller = new AbortController(); - abortControllerRef.current = controller; - - let accumulated = ''; - - // Stream idle timeout: abort if no SSE events arrive for this duration. - // Set slightly above the 5-minute permission timeout (300s) to avoid races. - const STREAM_IDLE_TIMEOUT_MS = 330_000; - let lastEventTime = Date.now(); - let isIdleTimeout = false; - const idleCheckTimer = setInterval(() => { - if (Date.now() - lastEventTime >= STREAM_IDLE_TIMEOUT_MS) { - clearInterval(idleCheckTimer); - isIdleTimeout = true; - controller.abort(); - } - }, 10_000); - const markActive = () => { lastEventTime = Date.now(); }; - - // Flush any pending image generation notices into the prompt so - // the LLM (especially in SDK resume mode) knows about previously generated images. - let effectiveContent = content; - if (pendingImageNoticesRef.current.length > 0) { - const notices = pendingImageNoticesRef.current.join('\n\n'); + // Flush pending image notices + const notices = pendingImageNoticesRef.current.length > 0 + ? [...pendingImageNoticesRef.current] + : undefined; + if (notices) { pendingImageNoticesRef.current = []; - effectiveContent = `${notices}\n\n---\n\n${content}`; } - try { - const response = await fetch('/api/chat', { - method: 'POST', - headers: { 'Content-Type': 'application/json' }, - body: JSON.stringify({ - session_id: sessionId, - content: effectiveContent, - mode, - model: currentModel, - provider_id: currentProviderId, - ...(files && files.length > 0 ? { files } : {}), - ...(systemPromptAppend ? { systemPromptAppend } : {}), - }), - signal: controller.signal, - }); - - if (!response.ok) { - const err = await response.json(); - throw new Error(err.error || 'Failed to send message'); - } - - const reader = response.body?.getReader(); - if (!reader) throw new Error('No response stream'); - - const result = await consumeSSEStream(reader, { - onText: (acc) => { - markActive(); - accumulated = acc; - accumulatedRef.current = acc; - setStreamingContent(acc); - }, - onToolUse: (tool) => { - markActive(); - setStreamingToolOutput(''); - setToolUses((prev) => { - if (prev.some((t) => t.id === tool.id)) return prev; - const next = [...prev, tool]; - toolUsesRef.current = next; - return next; - }); - }, - onToolResult: (res) => { - markActive(); - setStreamingToolOutput(''); - setToolResults((prev) => { - // Last-wins: if same tool_use_id exists, replace with newer (potentially more complete) result - const existingIdx = prev.findIndex(r => r.tool_use_id === res.tool_use_id); - if (existingIdx >= 0) { - const next = [...prev]; - next[existingIdx] = res; - toolResultsRef.current = next; - return next; - } - const next = [...prev, res]; - toolResultsRef.current = next; - return next; - }); - // Refresh file tree after each tool completes — file writes, - // deletions, and other FS operations are done via tools. - window.dispatchEvent(new Event('refresh-file-tree')); - }, - onToolOutput: (data) => { - markActive(); - setStreamingToolOutput((prev) => { - const next = prev + (prev ? '\n' : '') + data; - return next.length > 5000 ? next.slice(-5000) : next; - }); - }, - onToolProgress: (toolName, elapsed) => { - markActive(); - setStatusText(`Running ${toolName}... (${elapsed}s)`); - }, - onStatus: (text) => { - markActive(); - if (text?.startsWith('Connected (')) { - setStatusText(text); - setTimeout(() => setStatusText(undefined), 2000); - } else { - setStatusText(text); - } - }, - onResult: () => { - markActive(); - /* token usage captured by consumeSSEStream */ - }, - onPermissionRequest: (permData) => { - markActive(); - setPendingPermission(permData); - setPermissionResolved(null); - setPendingApprovalSessionId(sessionId); - }, - onToolTimeout: (toolName, elapsedSeconds) => { - markActive(); - toolTimeoutRef.current = { toolName, elapsedSeconds }; - }, - onModeChanged: (sdkMode) => { - markActive(); - // Map SDK permissionMode to UI mode - const uiMode = sdkMode === 'plan' ? 'plan' : 'code'; - handleModeChange(uiMode); - }, - onTaskUpdate: () => { - markActive(); - // Notify TaskList to refresh its data - window.dispatchEvent(new CustomEvent('tasks-updated')); - }, - onError: (acc) => { - markActive(); - accumulated = acc; - accumulatedRef.current = acc; - setStreamingContent(acc); - }, - }); - - accumulated = result.accumulated; - - // Build the assistant message content. - // When tools were used, serialize as a JSON content-blocks array - // (same format the backend API route stores), so MessageItem's - // parseToolBlocks() can render tool UI from history. - const finalToolUses = toolUsesRef.current; - const finalToolResults = toolResultsRef.current; - const hasTools = finalToolUses.length > 0 || finalToolResults.length > 0; - - let messageContent = accumulated.trim(); - if (hasTools && messageContent) { - const contentBlocks: Array> = []; - if (accumulated.trim()) { - contentBlocks.push({ type: 'text', text: accumulated.trim() }); - } - for (const tu of finalToolUses) { - contentBlocks.push({ type: 'tool_use', id: tu.id, name: tu.name, input: tu.input }); - const tr = finalToolResults.find(r => r.tool_use_id === tu.id); - if (tr) { - contentBlocks.push({ type: 'tool_result', tool_use_id: tr.tool_use_id, content: tr.content }); - } - } - messageContent = JSON.stringify(contentBlocks); - } - - // Add the assistant message to the list - if (messageContent) { - const assistantMessage: Message = { - id: 'temp-assistant-' + Date.now(), - session_id: sessionId, - role: 'assistant', - content: messageContent, - created_at: new Date().toISOString(), - token_usage: result.tokenUsage ? JSON.stringify(result.tokenUsage) : null, - }; - // Transfer pending reference images to this message ID so MessageItem can - // retrieve them. StreamingMessage uses __pending__ directly, but once the - // message transitions to MessageItem, it's keyed by message.id. - transferPendingToMessage(assistantMessage.id); - - setMessages((prev) => [...prev, assistantMessage]); - } - } catch (error) { - clearInterval(idleCheckTimer); - - if (error instanceof DOMException && error.name === 'AbortError') { - // Stream idle timeout — no SSE events for too long - if (isIdleTimeout) { - const idleSecs = Math.round(STREAM_IDLE_TIMEOUT_MS / 1000); - const errContent = accumulated.trim() - ? accumulated.trim() + `\n\n**Error:** Stream idle timeout — no response for ${idleSecs}s. The connection may have dropped.` - : `**Error:** Stream idle timeout — no response for ${idleSecs}s. The connection may have dropped.`; - const errMessage: Message = { - id: 'temp-error-' + Date.now(), - session_id: sessionId, - role: 'assistant', - content: errContent, - created_at: new Date().toISOString(), - token_usage: null, - }; - setMessages((prev) => [...prev, errMessage]); - } else { - const timeoutInfo = toolTimeoutRef.current; - if (timeoutInfo) { - // Tool execution timed out — save partial content and auto-retry - if (accumulated.trim()) { - const partialMessage: Message = { - id: 'temp-assistant-' + Date.now(), - session_id: sessionId, - role: 'assistant', - content: accumulated.trim() + `\n\n*(tool ${timeoutInfo.toolName} timed out after ${timeoutInfo.elapsedSeconds}s)*`, - created_at: new Date().toISOString(), - token_usage: null, - }; - setMessages((prev) => [...prev, partialMessage]); - } - // Clean up before auto-retry - toolTimeoutRef.current = null; - setIsStreaming(false); - setStreamingSessionId(''); - setStreamingContent(''); - accumulatedRef.current = ''; - toolUsesRef.current = []; - toolResultsRef.current = []; - setToolUses([]); - setToolResults([]); - setStreamingToolOutput(''); - setStatusText(undefined); - setPendingPermission(null); - setPermissionResolved(null); - setPendingApprovalSessionId(''); - abortControllerRef.current = null; - // Auto-retry: send a follow-up message telling the model to adjust strategy - setTimeout(() => { - sendMessageRef.current?.( - `The previous tool "${timeoutInfo.toolName}" timed out after ${timeoutInfo.elapsedSeconds} seconds. Please try a different approach to accomplish the task. Avoid repeating the same operation that got stuck.` - ); - }, 500); - return; // Skip the normal finally cleanup since we did it above - } - // User manually stopped generation — add partial content - if (accumulated.trim()) { - const partialMessage: Message = { - id: 'temp-assistant-' + Date.now(), - session_id: sessionId, - role: 'assistant', - content: accumulated.trim() + '\n\n*(generation stopped)*', - created_at: new Date().toISOString(), - token_usage: null, - }; - setMessages((prev) => [...prev, partialMessage]); - } - } - } else { - const errMsg = error instanceof Error ? error.message : 'Unknown error'; - const errorMessage: Message = { - id: 'temp-error-' + Date.now(), - session_id: sessionId, - role: 'assistant', - content: `**Error:** ${errMsg}`, - created_at: new Date().toISOString(), - token_usage: null, - }; - setMessages((prev) => [...prev, errorMessage]); - } - } finally { - clearInterval(idleCheckTimer); - toolTimeoutRef.current = null; - setIsStreaming(false); - setStreamingSessionId(''); - setStreamingContent(''); - accumulatedRef.current = ''; - toolUsesRef.current = []; - toolResultsRef.current = []; - setToolUses([]); - setToolResults([]); - setStreamingToolOutput(''); - setStatusText(undefined); - setPendingPermission(null); - setPermissionResolved(null); - setPendingApprovalSessionId(''); - abortControllerRef.current = null; - // Notify file tree to refresh after AI finishes - window.dispatchEvent(new CustomEvent('refresh-file-tree')); - } + // Delegate to stream session manager + startStream({ + sessionId, + content, + mode, + model: currentModel, + providerId: currentProviderId, + files, + systemPromptAppend, + pendingImageNotices: notices, + onModeChanged: (sdkMode) => { + const uiMode = sdkMode === 'plan' ? 'plan' : 'code'; + handleModeChange(uiMode); + }, + sendMessageFn: (retryContent: string, retryFiles?: FileAttachment[]) => { + sendMessageRef.current?.(retryContent, retryFiles); + }, + }); }, - [sessionId, isStreaming, setStreamingSessionId, setPendingApprovalSessionId, mode, currentModel, currentProviderId] + [sessionId, isStreaming, mode, currentModel, currentProviderId, handleModeChange] ); // Keep sendMessageRef in sync so timeout auto-retry can call it diff --git a/src/components/layout/AppShell.tsx b/src/components/layout/AppShell.tsx index 221a8de..8f42ce6 100644 --- a/src/components/layout/AppShell.tsx +++ b/src/components/layout/AppShell.tsx @@ -15,7 +15,9 @@ import { UpdateContext, type UpdateInfo } from "@/hooks/useUpdate"; import { ImageGenContext, useImageGenState } from "@/hooks/useImageGen"; import { BatchImageGenContext, useBatchImageGenState } from "@/hooks/useBatchImageGen"; import { ErrorBoundary } from "./ErrorBoundary"; +import { getActiveSessionIds, getSnapshot } from "@/lib/stream-session-manager"; +const EMPTY_SET = new Set(); const CHATLIST_MIN = 180; const CHATLIST_MAX = 400; const RIGHTPANEL_MIN = 200; @@ -93,6 +95,40 @@ export function AppShell({ children }: { children: React.ReactNode }) { const [streamingSessionId, setStreamingSessionId] = useState(""); const [pendingApprovalSessionId, setPendingApprovalSessionId] = useState(""); + // --- Multi-session stream tracking (driven by stream-session-manager) --- + const [activeStreamingSessions, setActiveStreamingSessions] = useState>(EMPTY_SET); + const [pendingApprovalSessionIds, setPendingApprovalSessionIds] = useState>(EMPTY_SET); + + // Listen for global stream events from stream-session-manager + useEffect(() => { + const handler = () => { + const activeIds = getActiveSessionIds(); + setActiveStreamingSessions(activeIds.length > 0 ? new Set(activeIds) : EMPTY_SET); + + const approvals = new Set(); + for (const sid of activeIds) { + const snap = getSnapshot(sid); + if (snap?.pendingPermission && !snap.permissionResolved) { + approvals.add(sid); + } + } + setPendingApprovalSessionIds(approvals.size > 0 ? approvals : EMPTY_SET); + }; + window.addEventListener('stream-session-event', handler); + return () => window.removeEventListener('stream-session-event', handler); + }, []); + + // Warn before closing window/tab while any session is streaming + useEffect(() => { + if (activeStreamingSessions.size === 0) return; + const handler = (e: BeforeUnloadEvent) => { + e.preventDefault(); + e.returnValue = ''; + }; + window.addEventListener('beforeunload', handler); + return () => window.removeEventListener('beforeunload', handler); + }, [activeStreamingSessions]); + // --- Doc Preview state --- const [previewFile, setPreviewFileRaw] = useState(null); const [previewViewMode, setPreviewViewMode] = useState("source"); @@ -348,12 +384,14 @@ export function AppShell({ children }: { children: React.ReactNode }) { setStreamingSessionId, pendingApprovalSessionId, setPendingApprovalSessionId, + activeStreamingSessions, + pendingApprovalSessionIds, previewFile, setPreviewFile, previewViewMode, setPreviewViewMode, }), - [panelOpen, setPanelOpen, panelContent, workingDirectory, sessionId, sessionTitle, streamingSessionId, pendingApprovalSessionId, previewFile, setPreviewFile, previewViewMode] + [panelOpen, setPanelOpen, panelContent, workingDirectory, sessionId, sessionTitle, streamingSessionId, pendingApprovalSessionId, activeStreamingSessions, pendingApprovalSessionIds, previewFile, setPreviewFile, previewViewMode] ); const imageGenValue = useImageGenState(); diff --git a/src/components/layout/ChatListPanel.tsx b/src/components/layout/ChatListPanel.tsx index 5ab7038..5a33777 100644 --- a/src/components/layout/ChatListPanel.tsx +++ b/src/components/layout/ChatListPanel.tsx @@ -118,7 +118,7 @@ const MODE_BADGE_CONFIG = { export function ChatListPanel({ open, width }: ChatListPanelProps) { const pathname = usePathname(); const router = useRouter(); - const { streamingSessionId, pendingApprovalSessionId, workingDirectory } = usePanel(); + const { streamingSessionId, pendingApprovalSessionId, activeStreamingSessions, pendingApprovalSessionIds, workingDirectory } = usePanel(); const { t } = useTranslation(); const { isElectron, openNativePicker } = useNativeFolderPicker(); const [sessions, setSessions] = useState([]); @@ -495,9 +495,9 @@ export function ChatListPanel({ open, width }: ChatListPanelProps) { const isHovered = hoveredSession === session.id; const isDeleting = deletingSession === session.id; const isSessionStreaming = - streamingSessionId === session.id; + activeStreamingSessions.has(session.id) || streamingSessionId === session.id; const needsApproval = - pendingApprovalSessionId === session.id; + pendingApprovalSessionIds.has(session.id) || pendingApprovalSessionId === session.id; const mode = session.mode || "code"; const badgeCfg = MODE_BADGE_CONFIG[mode as keyof typeof MODE_BADGE_CONFIG] || MODE_BADGE_CONFIG.code; diff --git a/src/hooks/usePanel.ts b/src/hooks/usePanel.ts index f614df3..366b8da 100644 --- a/src/hooks/usePanel.ts +++ b/src/hooks/usePanel.ts @@ -21,6 +21,10 @@ export interface PanelContextValue { setStreamingSessionId: (id: string) => void; pendingApprovalSessionId: string; setPendingApprovalSessionId: (id: string) => void; + /** All sessions with active streams (supports multi-session streaming) */ + activeStreamingSessions: Set; + /** All sessions with pending permission approval */ + pendingApprovalSessionIds: Set; previewFile: string | null; setPreviewFile: (path: string | null) => void; previewViewMode: PreviewViewMode; diff --git a/src/lib/stream-session-manager.ts b/src/lib/stream-session-manager.ts new file mode 100644 index 0000000..4296683 --- /dev/null +++ b/src/lib/stream-session-manager.ts @@ -0,0 +1,633 @@ +/** + * Stream Session Manager — client-side singleton that manages SSE streams + * independently of React component lifecycle. + * + * When a user switches sessions, the old ChatView unmounts but the stream + * continues running here. The new ChatView (or the same one re-mounted) + * subscribes to get the current snapshot. + * + * Uses globalThis pattern (same as conversation-registry.ts) to survive + * Next.js HMR without losing state. + */ + +import { consumeSSEStream } from '@/hooks/useSSEStream'; +import { transferPendingToMessage } from '@/lib/image-ref-store'; +import type { + ToolUseInfo, + ToolResultInfo, + SessionStreamSnapshot, + StreamEvent, + StreamEventListener, + TokenUsage, + PermissionRequestEvent, + FileAttachment, +} from '@/types'; + +// ========================================== +// Internal types +// ========================================== + +interface ActiveStream { + sessionId: string; + abortController: AbortController; + snapshot: SessionStreamSnapshot; + listeners: Set; + idleCheckTimer: ReturnType | null; + lastEventTime: number; + gcTimer: ReturnType | null; + // Mutable accumulators (snapshot gets new object refs on each emit) + accumulatedText: string; + toolUsesArray: ToolUseInfo[]; + toolResultsArray: ToolResultInfo[]; + toolOutputAccumulated: string; + toolTimeoutInfo: { toolName: string; elapsedSeconds: number } | null; + isIdleTimeout: boolean; + sendMessageFn: ((content: string, files?: FileAttachment[]) => void) | null; +} + +export interface StartStreamParams { + sessionId: string; + content: string; + mode: string; + model: string; + providerId: string; + files?: FileAttachment[]; + systemPromptAppend?: string; + pendingImageNotices?: string[]; + /** Called when SDK mode changes (e.g. plan → code) */ + onModeChanged?: (mode: string) => void; + /** Reference to the outer sendMessage so tool-timeout auto-retry works */ + sendMessageFn?: (content: string, files?: FileAttachment[]) => void; +} + +// ========================================== +// Singleton via globalThis +// ========================================== + +const GLOBAL_KEY = '__streamSessionManager__' as const; +const STREAM_IDLE_TIMEOUT_MS = 330_000; +const GC_DELAY_MS = 5 * 60 * 1000; // 5 minutes + +function getStreamsMap(): Map { + if (!(globalThis as Record)[GLOBAL_KEY]) { + (globalThis as Record)[GLOBAL_KEY] = new Map(); + } + return (globalThis as Record)[GLOBAL_KEY] as Map; +} + +// ========================================== +// Helpers +// ========================================== + +function buildSnapshot(stream: ActiveStream): SessionStreamSnapshot { + return { + sessionId: stream.sessionId, + phase: stream.snapshot.phase, + streamingContent: stream.accumulatedText, + toolUses: [...stream.toolUsesArray], + toolResults: [...stream.toolResultsArray], + streamingToolOutput: stream.toolOutputAccumulated, + statusText: stream.snapshot.statusText, + pendingPermission: stream.snapshot.pendingPermission, + permissionResolved: stream.snapshot.permissionResolved, + tokenUsage: stream.snapshot.tokenUsage, + startedAt: stream.snapshot.startedAt, + completedAt: stream.snapshot.completedAt, + error: stream.snapshot.error, + finalMessageContent: stream.snapshot.finalMessageContent, + }; +} + +function emit(stream: ActiveStream, type: StreamEvent['type']) { + const snapshot = buildSnapshot(stream); + stream.snapshot = snapshot; // store latest + const event: StreamEvent = { type, sessionId: stream.sessionId, snapshot }; + for (const listener of stream.listeners) { + try { listener(event); } catch { /* listener error */ } + } + // Also dispatch window event for AppShell + if (typeof window !== 'undefined') { + window.dispatchEvent(new CustomEvent('stream-session-event', { detail: event })); + } +} + +function scheduleGC(stream: ActiveStream) { + if (stream.gcTimer) clearTimeout(stream.gcTimer); + stream.gcTimer = setTimeout(() => { + const map = getStreamsMap(); + const current = map.get(stream.sessionId); + if (current === stream && current.snapshot.phase !== 'active') { + map.delete(stream.sessionId); + } + }, GC_DELAY_MS); +} + +function cleanupTimers(stream: ActiveStream) { + if (stream.idleCheckTimer) { + clearInterval(stream.idleCheckTimer); + stream.idleCheckTimer = null; + } +} + +// ========================================== +// Public API +// ========================================== + +export function startStream(params: StartStreamParams): void { + const map = getStreamsMap(); + const existing = map.get(params.sessionId); + + // If already streaming this session, abort old stream first + if (existing && existing.snapshot.phase === 'active') { + existing.abortController.abort(); + cleanupTimers(existing); + } + + const abortController = new AbortController(); + + const stream: ActiveStream = { + sessionId: params.sessionId, + abortController, + snapshot: { + sessionId: params.sessionId, + phase: 'active', + streamingContent: '', + toolUses: [], + toolResults: [], + streamingToolOutput: '', + statusText: undefined, + pendingPermission: null, + permissionResolved: null, + tokenUsage: null, + startedAt: Date.now(), + completedAt: null, + error: null, + finalMessageContent: null, + }, + listeners: existing?.listeners ?? new Set(), + idleCheckTimer: null, + lastEventTime: Date.now(), + gcTimer: null, + accumulatedText: '', + toolUsesArray: [], + toolResultsArray: [], + toolOutputAccumulated: '', + toolTimeoutInfo: null, + isIdleTimeout: false, + sendMessageFn: params.sendMessageFn ?? null, + }; + + map.set(params.sessionId, stream); + emit(stream, 'phase-changed'); + + // Run the stream in background (non-blocking) + runStream(stream, params).catch(() => {}); +} + +async function runStream(stream: ActiveStream, params: StartStreamParams): Promise { + const markActive = () => { stream.lastEventTime = Date.now(); }; + + // Idle timeout checker + stream.idleCheckTimer = setInterval(() => { + if (Date.now() - stream.lastEventTime >= STREAM_IDLE_TIMEOUT_MS) { + cleanupTimers(stream); + stream.isIdleTimeout = true; + stream.abortController.abort(); + } + }, 10_000); + + // Flush pending image notices + let effectiveContent = params.content; + if (params.pendingImageNotices && params.pendingImageNotices.length > 0) { + const notices = params.pendingImageNotices.join('\n\n'); + effectiveContent = `${notices}\n\n---\n\n${params.content}`; + } + + try { + const response = await fetch('/api/chat', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify({ + session_id: params.sessionId, + content: effectiveContent, + mode: params.mode, + model: params.model, + provider_id: params.providerId, + ...(params.files && params.files.length > 0 ? { files: params.files } : {}), + ...(params.systemPromptAppend ? { systemPromptAppend: params.systemPromptAppend } : {}), + }), + signal: stream.abortController.signal, + }); + + if (!response.ok) { + const err = await response.json(); + throw new Error(err.error || 'Failed to send message'); + } + + const reader = response.body?.getReader(); + if (!reader) throw new Error('No response stream'); + + const result = await consumeSSEStream(reader, { + onText: (acc) => { + markActive(); + stream.accumulatedText = acc; + emit(stream, 'snapshot-updated'); + }, + onToolUse: (tool) => { + markActive(); + stream.toolOutputAccumulated = ''; + if (!stream.toolUsesArray.some(t => t.id === tool.id)) { + stream.toolUsesArray = [...stream.toolUsesArray, tool]; + } + emit(stream, 'snapshot-updated'); + }, + onToolResult: (res) => { + markActive(); + stream.toolOutputAccumulated = ''; + const existingIdx = stream.toolResultsArray.findIndex(r => r.tool_use_id === res.tool_use_id); + if (existingIdx >= 0) { + const next = [...stream.toolResultsArray]; + next[existingIdx] = res; + stream.toolResultsArray = next; + } else { + stream.toolResultsArray = [...stream.toolResultsArray, res]; + } + emit(stream, 'snapshot-updated'); + // Refresh file tree after each tool completes + window.dispatchEvent(new Event('refresh-file-tree')); + }, + onToolOutput: (data) => { + markActive(); + const next = stream.toolOutputAccumulated + (stream.toolOutputAccumulated ? '\n' : '') + data; + stream.toolOutputAccumulated = next.length > 5000 ? next.slice(-5000) : next; + emit(stream, 'snapshot-updated'); + }, + onToolProgress: (toolName, elapsed) => { + markActive(); + stream.snapshot = { ...stream.snapshot, statusText: `Running ${toolName}... (${elapsed}s)` }; + emit(stream, 'snapshot-updated'); + }, + onStatus: (text) => { + markActive(); + if (text?.startsWith('Connected (')) { + stream.snapshot = { ...stream.snapshot, statusText: text }; + emit(stream, 'snapshot-updated'); + setTimeout(() => { + // Only clear if still the same status + if (stream.snapshot.statusText === text) { + stream.snapshot = { ...stream.snapshot, statusText: undefined }; + emit(stream, 'snapshot-updated'); + } + }, 2000); + } else { + stream.snapshot = { ...stream.snapshot, statusText: text }; + emit(stream, 'snapshot-updated'); + } + }, + onResult: (usage) => { + markActive(); + stream.snapshot = { ...stream.snapshot, tokenUsage: usage }; + }, + onPermissionRequest: (permData) => { + markActive(); + stream.snapshot = { + ...stream.snapshot, + pendingPermission: permData, + permissionResolved: null, + }; + emit(stream, 'permission-request'); + }, + onToolTimeout: (toolName, elapsedSeconds) => { + markActive(); + stream.toolTimeoutInfo = { toolName, elapsedSeconds }; + }, + onModeChanged: (sdkMode) => { + markActive(); + if (params.onModeChanged) { + params.onModeChanged(sdkMode); + } + }, + onTaskUpdate: () => { + markActive(); + window.dispatchEvent(new CustomEvent('tasks-updated')); + }, + onError: (acc) => { + markActive(); + stream.accumulatedText = acc; + emit(stream, 'snapshot-updated'); + }, + }); + + // Stream completed successfully — build final message content + const accumulated = result.accumulated; + const finalToolUses = stream.toolUsesArray; + const finalToolResults = stream.toolResultsArray; + const hasTools = finalToolUses.length > 0 || finalToolResults.length > 0; + + let messageContent = accumulated.trim(); + if (hasTools && messageContent) { + const contentBlocks: Array> = []; + if (accumulated.trim()) { + contentBlocks.push({ type: 'text', text: accumulated.trim() }); + } + for (const tu of finalToolUses) { + contentBlocks.push({ type: 'tool_use', id: tu.id, name: tu.name, input: tu.input }); + const tr = finalToolResults.find(r => r.tool_use_id === tu.id); + if (tr) { + contentBlocks.push({ type: 'tool_result', tool_use_id: tr.tool_use_id, content: tr.content }); + } + } + messageContent = JSON.stringify(contentBlocks); + } + + // Update snapshot with completion info + stream.snapshot = { + ...buildSnapshot(stream), + phase: 'completed', + completedAt: Date.now(), + tokenUsage: result.tokenUsage, + finalMessageContent: messageContent || null, + statusText: undefined, + pendingPermission: null, + permissionResolved: null, + }; + stream.accumulatedText = ''; + stream.toolUsesArray = []; + stream.toolResultsArray = []; + stream.toolOutputAccumulated = ''; + + cleanupTimers(stream); + emit(stream, 'completed'); + scheduleGC(stream); + + // Refresh file tree after completion + window.dispatchEvent(new CustomEvent('refresh-file-tree')); + + } catch (error) { + cleanupTimers(stream); + + if (error instanceof DOMException && error.name === 'AbortError') { + if (stream.isIdleTimeout) { + // Idle timeout + const idleSecs = Math.round(STREAM_IDLE_TIMEOUT_MS / 1000); + const errContent = stream.accumulatedText.trim() + ? stream.accumulatedText.trim() + `\n\n**Error:** Stream idle timeout — no response for ${idleSecs}s. The connection may have dropped.` + : `**Error:** Stream idle timeout — no response for ${idleSecs}s. The connection may have dropped.`; + + stream.snapshot = { + ...buildSnapshot(stream), + phase: 'error', + completedAt: Date.now(), + error: `Stream idle timeout (${idleSecs}s)`, + finalMessageContent: errContent, + statusText: undefined, + pendingPermission: null, + permissionResolved: null, + }; + stream.accumulatedText = ''; + stream.toolUsesArray = []; + stream.toolResultsArray = []; + stream.toolOutputAccumulated = ''; + emit(stream, 'completed'); + scheduleGC(stream); + } else if (stream.toolTimeoutInfo) { + // Tool timeout — auto-retry + const timeoutInfo = stream.toolTimeoutInfo; + const partialContent = stream.accumulatedText.trim() + ? stream.accumulatedText.trim() + `\n\n*(tool ${timeoutInfo.toolName} timed out after ${timeoutInfo.elapsedSeconds}s)*` + : null; + + stream.snapshot = { + ...buildSnapshot(stream), + phase: 'stopped', + completedAt: Date.now(), + finalMessageContent: partialContent, + statusText: undefined, + pendingPermission: null, + permissionResolved: null, + }; + stream.accumulatedText = ''; + stream.toolUsesArray = []; + stream.toolResultsArray = []; + stream.toolOutputAccumulated = ''; + stream.toolTimeoutInfo = null; + emit(stream, 'completed'); + scheduleGC(stream); + + // Auto-retry via sendMessageFn + if (stream.sendMessageFn) { + const fn = stream.sendMessageFn; + setTimeout(() => { + fn( + `The previous tool "${timeoutInfo.toolName}" timed out after ${timeoutInfo.elapsedSeconds} seconds. Please try a different approach to accomplish the task. Avoid repeating the same operation that got stuck.` + ); + }, 500); + } + } else { + // User manually stopped — add partial content with "(generation stopped)" + const partialContent = stream.accumulatedText.trim() + ? stream.accumulatedText.trim() + '\n\n*(generation stopped)*' + : null; + + stream.snapshot = { + ...buildSnapshot(stream), + phase: 'stopped', + completedAt: Date.now(), + finalMessageContent: partialContent, + statusText: undefined, + pendingPermission: null, + permissionResolved: null, + }; + stream.accumulatedText = ''; + stream.toolUsesArray = []; + stream.toolResultsArray = []; + stream.toolOutputAccumulated = ''; + emit(stream, 'completed'); + scheduleGC(stream); + } + } else { + // Non-abort error + const errMsg = error instanceof Error ? error.message : 'Unknown error'; + stream.snapshot = { + ...buildSnapshot(stream), + phase: 'error', + completedAt: Date.now(), + error: errMsg, + finalMessageContent: `**Error:** ${errMsg}`, + statusText: undefined, + pendingPermission: null, + permissionResolved: null, + }; + stream.accumulatedText = ''; + stream.toolUsesArray = []; + stream.toolResultsArray = []; + stream.toolOutputAccumulated = ''; + emit(stream, 'completed'); + scheduleGC(stream); + } + } +} + +// ========================================== +// Stop +// ========================================== + +export function stopStream(sessionId: string): void { + const stream = getStreamsMap().get(sessionId); + if (stream && stream.snapshot.phase === 'active') { + stream.abortController.abort(); + } +} + +// ========================================== +// Subscribe +// ========================================== + +export function subscribe(sessionId: string, listener: StreamEventListener): () => void { + const map = getStreamsMap(); + let stream = map.get(sessionId); + + if (!stream) { + // Create a placeholder entry to hold listeners even when no stream is active + stream = { + sessionId, + abortController: new AbortController(), + snapshot: { + sessionId, + phase: 'completed' as const, + streamingContent: '', + toolUses: [], + toolResults: [], + streamingToolOutput: '', + statusText: undefined, + pendingPermission: null, + permissionResolved: null, + tokenUsage: null, + startedAt: 0, + completedAt: null, + error: null, + finalMessageContent: null, + }, + listeners: new Set(), + idleCheckTimer: null, + lastEventTime: 0, + gcTimer: null, + accumulatedText: '', + toolUsesArray: [], + toolResultsArray: [], + toolOutputAccumulated: '', + toolTimeoutInfo: null, + isIdleTimeout: false, + sendMessageFn: null, + }; + map.set(sessionId, stream); + } + + stream.listeners.add(listener); + + return () => { + stream!.listeners.delete(listener); + }; +} + +// ========================================== +// Snapshot access +// ========================================== + +export function getSnapshot(sessionId: string): SessionStreamSnapshot | null { + const stream = getStreamsMap().get(sessionId); + if (!stream) return null; + // Don't return stale placeholder entries + if (stream.snapshot.startedAt === 0) return null; + return stream.snapshot; +} + +export function isStreamActive(sessionId: string): boolean { + const stream = getStreamsMap().get(sessionId); + return stream?.snapshot.phase === 'active' || false; +} + +export function getActiveSessionIds(): string[] { + const ids: string[] = []; + for (const [id, stream] of getStreamsMap()) { + if (stream.snapshot.phase === 'active') { + ids.push(id); + } + } + return ids; +} + +// ========================================== +// Permission response +// ========================================== + +export async function respondToPermission( + sessionId: string, + decision: 'allow' | 'allow_session' | 'deny', + updatedInput?: Record, +): Promise { + const stream = getStreamsMap().get(sessionId); + if (!stream || !stream.snapshot.pendingPermission) return; + + const perm = stream.snapshot.pendingPermission; + + const body = { + permissionRequestId: perm.permissionRequestId, + decision: decision === 'deny' + ? { behavior: 'deny' as const, message: 'User denied permission' } + : { + behavior: 'allow' as const, + ...(decision === 'allow_session' && perm.suggestions + ? { updatedPermissions: perm.suggestions } + : {}), + ...(updatedInput ? { updatedInput } : {}), + }, + }; + + // Update snapshot immediately + stream.snapshot = { + ...stream.snapshot, + permissionResolved: decision === 'deny' ? 'deny' : 'allow', + }; + emit(stream, 'snapshot-updated'); + + try { + await fetch('/api/chat/permission', { + method: 'POST', + headers: { 'Content-Type': 'application/json' }, + body: JSON.stringify(body), + }); + } catch { + // Best effort + } + + // Clear permission state after delay (only if no new request arrived) + const answeredId = perm.permissionRequestId; + setTimeout(() => { + if (stream.snapshot.pendingPermission?.permissionRequestId === answeredId) { + stream.snapshot = { + ...stream.snapshot, + pendingPermission: null, + permissionResolved: null, + }; + emit(stream, 'snapshot-updated'); + } + }, 1000); +} + +// ========================================== +// Cleanup +// ========================================== + +export function clearSnapshot(sessionId: string): void { + const stream = getStreamsMap().get(sessionId); + if (stream && stream.snapshot.phase !== 'active') { + if (stream.gcTimer) clearTimeout(stream.gcTimer); + // Keep the listeners entry but reset the snapshot + stream.snapshot = { + ...stream.snapshot, + startedAt: 0, + finalMessageContent: null, + }; + } +} diff --git a/src/types/index.ts b/src/types/index.ts index 746cd86..5634cd4 100644 --- a/src/types/index.ts +++ b/src/types/index.ts @@ -590,6 +590,49 @@ export interface MediaJobListResponse { jobs: MediaJob[]; } +// ========================================== +// Stream Session Manager Types +// ========================================== + +export interface ToolUseInfo { + id: string; + name: string; + input: unknown; +} + +export interface ToolResultInfo { + tool_use_id: string; + content: string; +} + +export type StreamPhase = 'active' | 'completed' | 'error' | 'stopped'; + +export interface SessionStreamSnapshot { + sessionId: string; + phase: StreamPhase; + streamingContent: string; + toolUses: ToolUseInfo[]; + toolResults: ToolResultInfo[]; + streamingToolOutput: string; + statusText: string | undefined; + pendingPermission: PermissionRequestEvent | null; + permissionResolved: 'allow' | 'deny' | null; + tokenUsage: TokenUsage | null; + startedAt: number; + completedAt: number | null; + error: string | null; + /** Final message content built at stream completion for ChatView to consume */ + finalMessageContent: string | null; +} + +export interface StreamEvent { + type: 'snapshot-updated' | 'phase-changed' | 'permission-request' | 'completed'; + sessionId: string; + snapshot: SessionStreamSnapshot; +} + +export type StreamEventListener = (event: StreamEvent) => void; + export interface ClaudeStreamOptions { prompt: string; sessionId: string;