perf(web): 优化流式 delta 处理，避免每 token 全数组拷贝

用 ref 累加 delta chunks + requestAnimationFrame 批量 flush，将每秒数十次 O(n) 数组拷贝合并为一帧一次状态更新。 - useMessages.ts: 新增 streamingRef 累加器 + rAF 调度，所有 delta 只 push 到 ref，由 flushStreaming 批量落盘；用 rafScheduledRef boolean 追踪调度状态（避免 rAF handle 为 0 时与 null 检查冲突）； flushStreaming 基于 prev 实际状态判断创建/更新消息（批处理安全） - useChat.ts: 5 处 setMessages([]) 改为 clearMessages()，切话题/ 会话/通道时重置流式 ref，避免脏状态残留 - test/setup.ts: 同步化 rAF 并返回递增非零 handle，保持测试同步语义 - useChat.test.ts: 新增对抗性测试（批量 delta 累加、stream_end 后 assistant_response 替换含 reasoning）复杂度：O(n²·L) → O(n) 每帧，15 个测试全部通过
2026-08-06 22:45:58 +08:00 · 2026-08-06 22:45:58 +08:00 · f8c984aef4
commit f8c984aef4
parent 510520e08e
4 changed files with 196 additions and 35 deletions
--- a/web/src/hooks/chat/useMessages.ts
+++ b/web/src/hooks/chat/useMessages.ts
@ -2,6 +2,7 @@ import {
  useState,
  useCallback,
  useRef,
+  useEffect,
  type Dispatch,
  type SetStateAction,
  type MutableRefObject,
@ -63,6 +64,110 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn {
    });
  }, []);

+  // ---- 流式 delta 累加器 ----
+  // 用 ref 暂存 delta chunks，rAF 批量 flush 到 state，避免每个 token 都拷贝整个消息数组。
+  // 首 delta 同步写入消息壳，保证消息立即存在（assistant_response 可按 id 替换）。
+  const streamingRef = useRef<{
+    id: string;
+    contentChunks: string[];
+    reasoningChunks: string[];
+    index: number;
+    timestamp: number;
+    dirty: boolean;
+  } | null>(null);
+  const rafHandleRef = useRef<number | null>(null);
+  // 独立的调度标志：rAF handle 在某些环境可能为 0（与 null 检查冲突），
+  // 用 boolean 显式追踪是否已有 pending flush，语义更稳健。
+  const rafScheduledRef = useRef(false);
+
+  // flush pending 流式内容到 state。force=true 时即使 dirty=false 也强制 flush一次，
+  // 用于 finishStreaming 确保壳已创建（边界：首 delta 后立即 stream_end，dirty 可能已清）。
+  const flushStreaming = useCallback((force = false) => {
+    const stream = streamingRef.current;
+    if (!stream) return;
+    if (!force && !stream.dirty) return;
+    stream.dirty = false;
+    const content = stream.contentChunks.join('');
+    const reasoning =
+      stream.reasoningChunks.length > 0 ? stream.reasoningChunks.join('') : undefined;
+    setMessages((prev) => {
+      // 优先用缓存的 index 直接定位（O(1)），并校验 id 仍匹配
+      if (stream.index >= 0 && stream.index < prev.length && prev[stream.index].id === stream.id) {
+        const updated = prev.slice();
+        const existing = updated[stream.index];
+        updated[stream.index] = {
+          ...existing,
+          content,
+          reasoningContent: reasoning ?? existing.reasoningContent,
+        };
+        return updated;
+      }
+      // 回退：按 id 查找
+      const idx = prev.findIndex((m) => m.id === stream.id && m.type === 'message');
+      if (idx >= 0) {
+        stream.index = idx;
+        const updated = prev.slice();
+        const existing = updated[idx];
+        updated[idx] = {
+          ...existing,
+          content,
+          reasoningContent: reasoning ?? existing.reasoningContent,
+        };
+        return updated;
+      }
+      // 消息壳尚未存在：创建（基于 prev 实际状态判断，不依赖外部标志，批处理安全）
+      stream.index = prev.length;
+      return [
+        ...prev,
+        {
+          id: stream.id,
+          role: 'assistant' as const,
+          content,
+          timestamp: stream.timestamp,
+          type: 'message' as const,
+          reasoningContent: reasoning,
+        },
+      ];
+    });
+  }, []);
+
+  const clearStreaming = useCallback(() => {
+    if (rafScheduledRef.current) {
+      if (rafHandleRef.current !== null) {
+        cancelAnimationFrame(rafHandleRef.current);
+      }
+      rafHandleRef.current = null;
+      rafScheduledRef.current = false;
+    }
+    streamingRef.current = null;
+  }, []);
+
+  const scheduleFlush = useCallback(() => {
+    if (rafScheduledRef.current) return;
+    rafScheduledRef.current = true;
+    rafHandleRef.current = requestAnimationFrame(() => {
+      rafHandleRef.current = null;
+      rafScheduledRef.current = false;
+      flushStreaming(false);
+    });
+  }, [flushStreaming]);
+
+  const finishStreaming = useCallback(() => {
+    flushStreaming(true);
+    clearStreaming();
+  }, [flushStreaming, clearStreaming]);
+
+  // 卸载时取消未决的 rAF，避免在已卸载组件上触发状态更新
+  useEffect(() => {
+    return () => {
+      if (rafScheduledRef.current && rafHandleRef.current !== null) {
+        cancelAnimationFrame(rafHandleRef.current);
+      }
+      rafHandleRef.current = null;
+      rafScheduledRef.current = false;
+    };
+  }, []);
+
  const handleMessage = useCallback((content: string, attachments?: Attachment[]) => {
    setMessages((prev) => [
      ...prev,
@ -79,8 +184,9 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn {
  }, []);

  const clearMessages = useCallback(() => {
+    clearStreaming();
    setMessages([]);
-  }, []);
+  }, [clearStreaming]);

  const handleStop = useCallback((): Command => {
    return { type: 'stop_execution' };
@ -88,6 +194,10 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn {

  const handleMainViewMessage = useCallback(
    (message: WsOutbound): boolean => {
+      // 非流式消息到达前，先把 pending 的流式内容落盘，避免被后续消息覆盖或丢失
+      if (message.type !== 'stream_delta') {
+        finishStreaming();
+      }
      switch (message.type) {
        case 'task_started': {
          const msg = message as TaskStarted;
@ -131,32 +241,26 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn {
        case 'stream_delta': {
          const msg = message as StreamDelta;
          if (msg.topic_id && msg.topic_id !== selectedTopicRef.current) return true;
-          setMessages((prev) => {
-            const existingIdx = prev.findIndex((m) => m.id === msg.id && m.type === 'message');
-            if (existingIdx >= 0) {
-              const updated = [...prev];
-              const existing = updated[existingIdx];
-              updated[existingIdx] = {
-                ...existing,
-                content: existing.content + msg.delta,
-                reasoningContent: msg.reasoning_delta
-                  ? (existing.reasoningContent || '') + msg.reasoning_delta
-                  : existing.reasoningContent,
-              };
-              return updated;
-            }
-            return [
-              ...prev,
-              {
-                id: msg.id,
-                role: 'assistant' as const,
-                content: msg.delta,
-                timestamp: Math.floor(Date.now() / 1000),
-                type: 'message' as const,
-                reasoningContent: msg.reasoning_delta,
-              },
-            ];
-          });
+
+          // 所有 delta 都只累加到 ref，由 rAF 批量 flush 到 state。
+          // 首 delta 由 flushStreaming 在消息不存在时创建壳；后续 delta 走 index 直接更新。
+          let stream = streamingRef.current;
+          if (!stream || stream.id !== msg.id) {
+            stream = {
+              id: msg.id,
+              contentChunks: [],
+              reasoningChunks: [],
+              index: -1,
+              timestamp: Math.floor(Date.now() / 1000),
+              dirty: false,
+            };
+            streamingRef.current = stream;
+          }
+          if (msg.delta) stream.contentChunks.push(msg.delta);
+          if (msg.reasoning_delta) stream.reasoningChunks.push(msg.reasoning_delta);
+          stream.dirty = true;
+          scheduleFlush();
+
          if (msg.user_message_id) applyUserMessageId(msg.user_message_id);
          return true;
        }
@ -305,7 +409,14 @@ export function useMessages(options: UseMessagesOptions): UseMessagesReturn {
          return false;
      }
    },
-    [selectedTopicRef, topicsRef, bumpTopicRefreshTrigger, applyUserMessageId],
+    [
+      selectedTopicRef,
+      topicsRef,
+      bumpTopicRefreshTrigger,
+      applyUserMessageId,
+      finishStreaming,
+      scheduleFlush,
+    ],
  );

  return {
--- a/web/src/hooks/useChat.test.ts
+++ b/web/src/hooks/useChat.test.ts
@ -423,4 +423,46 @@ describe('useChat - handleServerMessage characterization', () => {
    act(() => result.current.handleServerMessage(otherTopicStream));
    expect(result.current.messages.find((m) => m.id === 'm-other')).toBeUndefined();
  });
+
+  it('14. multiple stream_delta in one act() batch accumulate correctly', () => {
+    const { result } = renderUseChat();
+    // 模拟同一同步批次内连续到达多个 delta（验证 chunks 累加 + index 跟踪）
+    act(() => {
+      result.current.handleServerMessage({ type: 'stream_delta', id: 'batch1', delta: 'A' });
+      result.current.handleServerMessage({ type: 'stream_delta', id: 'batch1', delta: 'B' });
+      result.current.handleServerMessage({ type: 'stream_delta', id: 'batch1', delta: 'C' });
+    });
+    expect(result.current.messages).toHaveLength(1);
+    expect(result.current.messages[0].id).toBe('batch1');
+    expect(result.current.messages[0].content).toBe('ABC');
+  });
+
+  it('15. stream_end then assistant_response replaces streamed content including reasoning', () => {
+    const { result } = renderUseChat();
+    act(() => {
+      result.current.handleServerMessage({
+        type: 'stream_delta',
+        id: 'r1',
+        delta: 'text',
+        reasoning_delta: 'think',
+      });
+    });
+    expect(result.current.messages[0].content).toBe('text');
+    expect(result.current.messages[0].reasoningContent).toBe('think');
+    // stream_end 触发 finishStreaming（flush pending + clear）
+    act(() => result.current.handleServerMessage({ type: 'stream_end', id: 'r1' }));
+    // assistant_response 替换整条消息
+    act(() =>
+      result.current.handleServerMessage({
+        type: 'assistant_response',
+        id: 'r1',
+        content: 'final text',
+        role: 'assistant',
+        reasoning_content: 'final reasoning',
+      }),
+    );
+    expect(result.current.messages).toHaveLength(1);
+    expect(result.current.messages[0].content).toBe('final text');
+    expect(result.current.messages[0].reasoningContent).toBe('final reasoning');
+  });
 });
--- a/web/src/hooks/useChat.ts
+++ b/web/src/hooks/useChat.ts
@ -169,8 +169,7 @@ export function useChat(): UseChatReturn {
        const prevSid = sessions.selectedSessionIdRef.current;
        const prevSessionExists =
          prevSid !== null && message.sessions.some((s) => s.session_id === prevSid);
-        const isReconnect =
-          topics.selectedTopicRef.current !== null && prevSessionExists;
+        const isReconnect = topics.selectedTopicRef.current !== null && prevSessionExists;
        sessions.setSessions(message.sessions);
        if (isReconnect) {
          // 原 session 仍在，保持选中
@ -184,7 +183,7 @@ export function useChat(): UseChatReturn {
          // 首次连接、切换通道、或原 session 已被删除：清空旧数据避免污染
          topics.setTopics([]);
          topics.setSelectedTopic(null);
-          messages.setMessages([]);
+          messages.clearMessages();
          sessions.setSelectedSessionId((prev) =>
            prev && message.sessions.some((s) => s.session_id === prev)
              ? prev
@ -204,7 +203,7 @@ export function useChat(): UseChatReturn {

      case 'topic_list': {
        const autoFocused = topics.handleTopicList(message);
-        if (autoFocused) messages.setMessages([]);
+        if (autoFocused) messages.clearMessages();
        messages.setIsLoading(false);
        return;
      }
@ -276,7 +275,7 @@ export function useChat(): UseChatReturn {
  // ---- selectTopic: 切换话题，清空消息和子智能体栈 ----
  const selectTopic = useCallback((topicId: string) => {
    topics.setSelectedTopic(topicId);
-    messages.setMessages([]);
+    messages.clearMessages();
    // ref + state 双写，消除竞态窗口（与 enter/exitSubAgentView 一致）
    subAgent.subAgentViewRef.current = null;
    subAgent.subAgentStackRef.current = [];
@ -292,7 +291,7 @@ export function useChat(): UseChatReturn {
      sessions.setSelectedSessionId(null);
      topics.setTopics([]);
      topics.setSelectedTopic(null);
-      messages.setMessages([]);
+      messages.clearMessages();
      subAgent.subAgentViewRef.current = null;
      subAgent.subAgentStackRef.current = [];
      subAgent.setSubAgentStack([]);
@ -308,7 +307,7 @@ export function useChat(): UseChatReturn {
      sessions.setSelectedSessionId(sessionId);
      topics.setTopics([]);
      topics.setSelectedTopic(null);
-      messages.setMessages([]);
+      messages.clearMessages();
      subAgent.subAgentViewRef.current = null;
      subAgent.subAgentStackRef.current = [];
      subAgent.setSubAgentStack([]);
--- a/web/src/test/setup.ts
+++ b/web/src/test/setup.ts
@ -1 +1,10 @@
 import '@testing-library/jest-dom/vitest';
+
+// 让 requestAnimationFrame 在测试中同步触发，使 rAF 批量更新的状态在 act() 内立即可见。
+// 返回递增的非零 handle，与浏览器真实 rAF 语义一致（0 是合法 ID 但会与 null 检查冲突）。
+let rafIdCounter = 1;
+globalThis.requestAnimationFrame = (cb: FrameRequestCallback): number => {
+  cb(performance.now());
+  return rafIdCounter++;
+};
+globalThis.cancelAnimationFrame = (): void => {};