feat: stream session manager + always show Claude Code built-in models, bump v0.17.4
Stream Session Manager — 切换会话不中断流： - src/lib/stream-session-manager.ts（新建）: 全局单例管理所有 SSE 流， 使用 globalThis 模式（同 conversation-registry.ts）保证 HMR 不丢状态。 管理 AbortController、idle timeout (330s)、tool timeout 自动重试、 5 分钟 GC，所有 SSE 回调写入 snapshot 并 emit 事件给订阅者。 导出 startStream / stopStream / subscribe / getSnapshot / respondToPermission / clearSnapshot 等 API。 - src/types/index.ts: 新增 ToolUseInfo、ToolResultInfo（从 ChatView 提取）、 StreamPhase、SessionStreamSnapshot、StreamEvent、StreamEventListener 类型， 作为管理器与 ChatView 之间的数据契约。 - src/components/chat/ChatView.tsx（重构）: 删除 abortControllerRef 及 unmount abort effect（此前切换会话的根因）、整个 fetch/SSE 逻辑约 280 行、 accumulatedRef/toolUsesRef/toolResultsRef/beforeunload handler。 改为 subscribe() + getSnapshot() 订阅模式，unmount 仅 unsubscribe 不 abort，mount 时通过 getSnapshot 恢复进行中的流状态。 sendMessage / stopStreaming / handlePermissionResponse 全部委托管理器。 - src/hooks/usePanel.ts: PanelContextValue 新增 activeStreamingSessions: Set<string> 和 pendingApprovalSessionIds: Set<string>，支持多会话并发流指示。 - src/components/layout/AppShell.tsx: 监听 stream-session-event 窗口事件 更新 activeStreamingSessions / pendingApprovalSessionIds Set 状态； 将 beforeunload handler 从 ChatView 移至此处（全局感知所有流）。 - src/components/layout/ChatListPanel.tsx: 侧栏流指示器改用 activeStreamingSessions.has() / pendingApprovalSessionIds.has()， 支持同时显示多个会话的流/审批状态。 模型选择器始终显示 Claude Code 内置模型： - src/app/api/providers/models/route.ts: 移除 hasEnvKey 环境变量检测条件， 始终显示 "Claude Code" provider 组（Sonnet 4.6 / Opus 4.6 / Haiku 4.5）。 根因：Claude Code CLI 通过 `claude login` 存储凭证在 ~/.claude/ 下， SDK 子进程可以读取，但 Next.js 进程的 process.env 中没有对应变量， 导致之前的 hasEnvKey 检测失败时模型选择器不显示内置模型。 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
7418andClaude Opus 4.6 committed
1 parent
0c379e5607
commit
54515e1cec
9 files changed
+856
-437
No files matched your search
Generated
+2
-2
@@ -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",
|
||||
|
||||
+1
-1
@@ -1,6 +1,6 @@
|
||||
{
|
||||
"name": "codepilot",
|
||||
"version": "0.17.3",
|
||||
"version": "0.17.4",
|
||||
"private": true,
|
||||
"author": {
|
||||
"name": "op7418",
|
||||
|
||||
@@ -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;
|
||||
|
||||
|
||||
+122
-410
@@ -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<ToolUseInfo[]>([]);
|
||||
const [toolResults, setToolResults] = useState<ToolResultInfo[]>([]);
|
||||
const [statusText, setStatusText] = useState<string | undefined>();
|
||||
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<PermissionRequestEvent | null>(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<SessionStreamSnapshot | null>(
|
||||
() => 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<string[]>([]);
|
||||
// Ref for sendMessage to allow self-referencing in timeout auto-retry
|
||||
const sendMessageRef = useRef<(content: string, files?: FileAttachment[]) => Promise<void>>(undefined);
|
||||
|
||||
const handleModeChange = useCallback((newMode: string) => {
|
||||
setMode(newMode);
|
||||
@@ -74,60 +81,68 @@ export function ChatView({ sessionId, initialMessages = [], initialHasMore = fal
|
||||
setCurrentModel(model);
|
||||
}, []);
|
||||
|
||||
const abortControllerRef = useRef<AbortController | null>(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<ToolUseInfo[]>([]);
|
||||
const toolResultsRef = useRef<ToolResultInfo[]>([]);
|
||||
// Ref for sendMessage to allow self-referencing in timeout auto-retry without circular deps
|
||||
const sendMessageRef = useRef<(content: string, files?: FileAttachment[]) => Promise<void>>(undefined);
|
||||
// Pending image generation notices — flushed into the next user message so the LLM knows about generated images
|
||||
const pendingImageNoticesRef = useRef<string[]>([]);
|
||||
|
||||
// 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<string, unknown>) => {
|
||||
if (!pendingPermission) return;
|
||||
|
||||
const body: { permissionRequestId: string; decision: { behavior: 'allow'; updatedPermissions?: unknown[]; updatedInput?: Record<string, unknown> } | { 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<string, unknown>) => {
|
||||
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<Record<string, unknown>> = [];
|
||||
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
|
||||
|
||||
@@ -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<string>();
|
||||
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<Set<string>>(EMPTY_SET);
|
||||
const [pendingApprovalSessionIds, setPendingApprovalSessionIds] = useState<Set<string>>(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<string>();
|
||||
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<string | null>(null);
|
||||
const [previewViewMode, setPreviewViewMode] = useState<PreviewViewMode>("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();
|
||||
|
||||
@@ -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<ChatSession[]>([]);
|
||||
@@ -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;
|
||||
|
||||
|
||||
@@ -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<string>;
|
||||
/** All sessions with pending permission approval */
|
||||
pendingApprovalSessionIds: Set<string>;
|
||||
previewFile: string | null;
|
||||
setPreviewFile: (path: string | null) => void;
|
||||
previewViewMode: PreviewViewMode;
|
||||
|
||||
@@ -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<StreamEventListener>;
|
||||
idleCheckTimer: ReturnType<typeof setInterval> | null;
|
||||
lastEventTime: number;
|
||||
gcTimer: ReturnType<typeof setTimeout> | 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<string, ActiveStream> {
|
||||
if (!(globalThis as Record<string, unknown>)[GLOBAL_KEY]) {
|
||||
(globalThis as Record<string, unknown>)[GLOBAL_KEY] = new Map<string, ActiveStream>();
|
||||
}
|
||||
return (globalThis as Record<string, unknown>)[GLOBAL_KEY] as Map<string, ActiveStream>;
|
||||
}
|
||||
|
||||
// ==========================================
|
||||
// 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<void> {
|
||||
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<Record<string, unknown>> = [];
|
||||
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<string, unknown>,
|
||||
): Promise<void> {
|
||||
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,
|
||||
};
|
||||
}
|
||||
}
|
||||
@@ -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;
|
||||
|
||||
Reference in new issue
Block a user