perf: 优化 Agent 流式会话处理

- 合并流式通知与持久化写入，限制浏览器会话缓存

- 增量投影轮次事件并稳定关联异步工具任务

easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.test.ts

@@ -1,7 +1,7 @@
 // @vitest-environment happy-dom
 
 import { createPinia, setActivePinia } from 'pinia';
-import { beforeEach, describe, expect, it, vi } from 'vitest';
+import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest';
 
 import { useUserStore } from '@easyflow/stores';
 
@@ -23,6 +23,10 @@ describe('agentChatRuntimeManager', () => {
     vi.clearAllMocks();
   });
 
+  afterEach(() => {
+    vi.useRealTimers();
+  });
+
   it('replaces draft image URLs and isolates snapshots by account', async () => {
     let callbacks: any;
     vi.mocked(sendAgentChat).mockImplementation((_data, options) => {
@@ -95,4 +99,76 @@ describe('agentChatRuntimeManager', () => {
     userStore.setUserInfo(firstAccount);
     expect(agentChatRuntimeManager.getSnapshot('101')).toBeUndefined();
   });
+
+  it('coalesces streaming notifications and persists terminal state immediately', async () => {
+    vi.useFakeTimers();
+    let callbacks: any;
+    vi.mocked(sendAgentChat).mockImplementation((_data, options) => {
+      callbacks = options;
+      return Promise.resolve() as any;
+    });
+    const account = {
+      avatar: '',
+      id: 'stream-user',
+      loginName: 'stream-user',
+      nickname: '流式用户',
+      tenantId: 'tenant-1',
+    };
+    useUserStore().setUserInfo(account);
+
+    await agentChatRuntimeManager.start({
+      agentId: 'agent-1',
+      prompt: '流式测试',
+      sessionId: 'stream-session',
+    });
+    const listener = vi.fn();
+    const unsubscribe = agentChatRuntimeManager.subscribe(listener);
+    const storageSpy = vi.spyOn(sessionStorage, 'setItem');
+
+    for (const delta of ['A', 'B', 'C']) {
+      callbacks.onMessage({
+        data: JSON.stringify({
+          domain: 'LLM',
+          payload: { delta },
+          type: 'MESSAGE',
+        }),
+      });
+    }
+
+    expect(listener).not.toHaveBeenCalled();
+    expect(storageSpy).not.toHaveBeenCalled();
+    expect(
+      JSON.stringify(
+        agentChatRuntimeManager.getSnapshot('stream-session')?.items,
+      ),
+    ).toContain('ABC');
+
+    await vi.advanceTimersByTimeAsync(50);
+    expect(listener).toHaveBeenCalledTimes(1);
+    expect(storageSpy).not.toHaveBeenCalled();
+
+    await vi.advanceTimersByTimeAsync(250);
+    expect(storageSpy).toHaveBeenCalledTimes(2);
+
+    callbacks.onMessage({
+      data: JSON.stringify({
+        domain: 'LLM',
+        payload: { delta: 'D' },
+        type: 'MESSAGE',
+      }),
+    });
+    listener.mockClear();
+    storageSpy.mockClear();
+    callbacks.onFinished();
+
+    expect(listener).toHaveBeenCalledTimes(1);
+    expect(storageSpy).toHaveBeenCalledTimes(2);
+    expect(agentChatRuntimeManager.getSnapshot('stream-session')).toEqual(
+      expect.objectContaining({ completed: true, sending: false }),
+    );
+
+    unsubscribe();
+    storageSpy.mockRestore();
+    clearAgentChatBrowserCache(account);
+  });
 });

easyflow-ui-admin/app/src/views/ai/agent-chat/agentChatRuntimeManager.ts

@@ -64,10 +64,15 @@ interface StartOptions {
 }
 
 const STORAGE_VERSION = 2;
+const STREAM_NOTIFY_INTERVAL_MS = 50;
+const STREAM_PERSIST_INTERVAL_MS = 300;
+const MAX_RUNTIME_SESSIONS_PER_IDENTITY = 10;
 
 const sessions = new Map<string, RuntimeSessionState>();
 const listeners = new Set<() => void>();
 const latestSessionIds = new Map<string, string>();
+const persistTimers = new Map<string, ReturnType<typeof setTimeout>>();
+let notifyTimer: ReturnType<typeof setTimeout> | undefined;
 
 function clone<T>(value: T): T {
   const serialized = JSON.stringify(value);
@@ -108,6 +113,24 @@ function notify() {
   }
 }
 
+function notifyNow() {
+  if (notifyTimer) {
+    clearTimeout(notifyTimer);
+    notifyTimer = undefined;
+  }
+  notify();
+}
+
+function scheduleNotify() {
+  if (notifyTimer || listeners.size === 0) {
+    return;
+  }
+  notifyTimer = setTimeout(() => {
+    notifyTimer = undefined;
+    notify();
+  }, STREAM_NOTIFY_INTERVAL_MS);
+}
+
 function persistSession(state: RuntimeSessionState) {
   const storage = safeSessionStorage();
   if (!storage) {
@@ -118,7 +141,7 @@ function persistSession(state: RuntimeSessionState) {
     agentName: state.agentName,
     completed: state.completed,
     error: state.error,
-    items: clone(state.items),
+    items: state.items,
     prompt: state.prompt,
     roundId: state.roundId,
     sessionId: state.sessionId,
@@ -136,6 +159,82 @@ function persistSession(state: RuntimeSessionState) {
   }
 }
 
+function cancelPersistTimer(scopedSessionKey: string) {
+  const timer = persistTimers.get(scopedSessionKey);
+  if (!timer) {
+    return;
+  }
+  clearTimeout(timer);
+  persistTimers.delete(scopedSessionKey);
+}
+
+function removeSessionCache(
+  scopedSessionKey: string,
+  state: RuntimeSessionState,
+) {
+  cancelPersistTimer(scopedSessionKey);
+  sessions.delete(scopedSessionKey);
+  try {
+    safeSessionStorage()?.removeItem(
+      storageKey(state.identity, state.sessionId),
+    );
+  } catch {
+    // 本地缓存清理失败不影响服务端会话。
+  }
+}
+
+function pruneRuntimeSessions(identity: string, currentSessionId: string) {
+  const scopedSessions = [...sessions.entries()]
+    .filter(([, session]) => session.identity === identity)
+    .sort((first, second) => second[1].updatedAt - first[1].updatedAt);
+  if (scopedSessions.length <= MAX_RUNTIME_SESSIONS_PER_IDENTITY) {
+    return;
+  }
+  const protectedKeys = new Set(
+    scopedSessions
+      .filter(
+        ([, session]) =>
+          session.sending || session.sessionId === currentSessionId,
+      )
+      .map(([key]) => key),
+  );
+  let retainedCount = protectedKeys.size;
+  for (const [key, session] of scopedSessions) {
+    if (protectedKeys.has(key)) {
+      continue;
+    }
+    if (retainedCount < MAX_RUNTIME_SESSIONS_PER_IDENTITY) {
+      retainedCount += 1;
+      continue;
+    }
+    removeSessionCache(key, session);
+  }
+}
+
+function touchState(state: RuntimeSessionState) {
+  state.updatedAt = Date.now();
+  latestSessionIds.set(state.identity, state.sessionId);
+  sessions.set(sessionKey(state.identity, state.sessionId), state);
+}
+
+function scheduleStateUpdate(state: RuntimeSessionState) {
+  touchState(state);
+  const scopedSessionKey = sessionKey(state.identity, state.sessionId);
+  if (!persistTimers.has(scopedSessionKey)) {
+    persistTimers.set(
+      scopedSessionKey,
+      setTimeout(() => {
+        persistTimers.delete(scopedSessionKey);
+        const current = sessions.get(scopedSessionKey);
+        if (current) {
+          persistSession(current);
+        }
+      }, STREAM_PERSIST_INTERVAL_MS),
+    );
+  }
+  scheduleNotify();
+}
+
 function restoreSession(identity: string, sessionId: string) {
   if (!identity) {
     return undefined;
@@ -179,11 +278,11 @@ function restoreSession(identity: string, sessionId: string) {
 }
 
 function upsertState(state: RuntimeSessionState) {
-  state.updatedAt = Date.now();
-  latestSessionIds.set(state.identity, state.sessionId);
-  sessions.set(sessionKey(state.identity, state.sessionId), state);
+  touchState(state);
+  cancelPersistTimer(sessionKey(state.identity, state.sessionId));
   persistSession(state);
-  notify();
+  pruneRuntimeSessions(state.identity, state.sessionId);
+  notifyNow();
 }
 
 function runningSession(identity = identityScope()) {
@@ -272,11 +371,12 @@ function replaceAcceptedImages(
 onAgentChatCacheClear((identity) => {
   for (const [key, session] of sessions) {
     if (session.identity === identity) {
+      cancelPersistTimer(key);
       sessions.delete(key);
     }
   }
   latestSessionIds.delete(identity);
-  notify();
+  notifyNow();
 });
 
 export const agentChatRuntimeManager = {
@@ -383,7 +483,7 @@ export const agentChatRuntimeManager = {
             void options.onInputAccepted?.();
           }
           applyAgentSseEnvelope(current.items, envelope, { roundId });
-          upsertState(current);
+          scheduleStateUpdate(current);
         },
       },
     );

easyflow-ui-admin/app/src/views/ai/agents/components/AgentTryoutPanel.vue

@@ -10,7 +10,7 @@ import type {
   AgentToolBinding,
 } from '../types';
 
-import { computed, onMounted, ref, watch } from 'vue';
+import { computed, onBeforeUnmount, onMounted, ref, watch } from 'vue';
 
 import { ChatTimeline } from '@easyflow/common-ui';
 import { BrushCleaning } from '@easyflow/icons';
@@ -39,6 +39,7 @@ const emit = defineEmits<{ close: [] }>();
 const {
   loading,
   clearDraftSession,
+  dispose,
   markToolApproving,
   markToolRejected,
   copyMessageText,
@@ -80,6 +81,9 @@ async function activateComposer() {
 }
 
 onMounted(() => void activateComposer());
+onBeforeUnmount(() => {
+  dispose();
+});
 
 watch(
   () => [props.agent.id, props.agent.localId],

easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.test.ts

@@ -241,4 +241,158 @@ describe('useAgentTryoutRawRounds', () => {
       toolCallId: 'call-approval',
     });
   });
+
+  it('异步工作流轮询事件始终归并到首张审批卡', () => {
+    const store = useAgentTryoutRawRounds({
+      mode: 'draft',
+      sessionId: 'session-raw-async-tool',
+    });
+    const roundId = store.createRound('生成文档');
+    store.recordEvent(roundId, {
+      domain: 'TOOL',
+      payload: {
+        input: { user_input: '写一篇小作文' },
+        requestId: 'req-async',
+        resumeToken: 'resume-async',
+        toolCallId: 'submit-call-1',
+        toolName: '文档生成',
+      },
+      type: 'FORM_REQUEST',
+    });
+    store.recordEvent(roundId, {
+      domain: 'TOOL',
+      payload: {
+        asyncTool: true,
+        phase: 'submit',
+        sourceToolCallId: 'submit-call-1',
+        status: 'RUNNING',
+        taskId: 'task-1',
+        toolCallId: 'task-1',
+        toolName: '文档生成',
+      },
+      type: 'TOOL_RESULT',
+    });
+    for (const sourceToolCallId of ['observe-call-1', 'observe-call-2']) {
+      store.recordEvent(roundId, {
+        domain: 'TOOL',
+        payload: {
+          asyncTool: true,
+          input: { taskId: 'task-1' },
+          phase: 'observe',
+          sourceToolCallId,
+          status: 'RUNNING',
+          taskId: 'task-1',
+          toolCallId: 'task-1',
+          toolName: '文档生成',
+        },
+        type: 'TOOL_CALL',
+      });
+    }
+    store.recordEvent(roundId, {
+      domain: 'TOOL',
+      payload: {
+        asyncTool: true,
+        phase: 'result',
+        sourceToolCallId: 'result-call-1',
+        status: 'SUCCEEDED',
+        taskId: 'task-1',
+        toolCallId: 'task-1',
+        toolName: '文档生成',
+      },
+      type: 'TOOL_RESULT',
+    });
+
+    const tools = store
+      .buildTimelineItems()
+      .filter((item) => item.type === 'tool');
+
+    expect(tools).toHaveLength(1);
+    expect(tools[0]).toMatchObject({
+      mode: 'approval',
+      status: 'success',
+      taskId: 'task-1',
+      toolCallId: 'task-1',
+      toolName: '文档生成',
+    });
+  });
+
+  it('连续文本增量压缩后刷新内容保持一致', () => {
+    const sessionId = 'stream-compaction';
+    const store = useAgentTryoutRawRounds({
+      mode: 'draft',
+      sessionId,
+    });
+    const roundId = store.createRound('你好');
+    const liveItems = store.buildTimelineItems();
+    const assistantText = (
+      items: ReturnType<typeof store.buildTimelineItems>,
+    ) =>
+      items
+        .flatMap((item) =>
+          item.type === 'message' && item.role === 'assistant'
+            ? item.parts
+                .filter((part) => part.type === 'text')
+                .map((part) => part.content)
+            : [],
+        )
+        .join('');
+
+    for (const delta of ['你', '好', '，', '世界']) {
+      const event = store.recordEvent(roundId, {
+        domain: 'LLM',
+        payload: { delta },
+        type: 'MESSAGE',
+      });
+      if (!event) {
+        throw new Error('流式事件记录失败');
+      }
+      store.projectEvent(liveItems, roundId, event);
+    }
+
+    expect(store.currentVariant(roundId)?.runtimeEvents).toHaveLength(1);
+    expect(store.currentVariant(roundId)?.runtimeEvents[0]?.payload.delta).toBe(
+      '你好，世界',
+    );
+    expect(assistantText(liveItems)).toBe('你好，世界');
+
+    store.flush();
+    const restored = useAgentTryoutRawRounds({
+      mode: 'draft',
+      sessionId,
+    });
+    expect(assistantText(restored.buildTimelineItems())).toBe('你好，世界');
+  });
+
+  it('存在待持久化增量时结束事件仍立即落盘', () => {
+    const sessionId = 'terminal-persist';
+    const store = useAgentTryoutRawRounds({
+      mode: 'draft',
+      sessionId,
+    });
+    const roundId = store.createRound('结束测试');
+
+    store.recordEvent(roundId, {
+      domain: 'LLM',
+      payload: { reasoning: '思考中' },
+      type: 'THINKING',
+    });
+    store.recordEvent(roundId, {
+      domain: 'SYSTEM',
+      payload: {},
+      type: 'DONE',
+    });
+
+    const restored = useAgentTryoutRawRounds({
+      mode: 'draft',
+      sessionId,
+    });
+    expect(restored.currentVariant(roundId)?.status).toBe('completed');
+    expect(
+      restored
+        .currentVariant(roundId)
+        ?.runtimeEvents.some(
+          (event) => event.domain === 'SYSTEM' && event.type === 'DONE',
+        ),
+    ).toBe(true);
+  });
 });

easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutRawRounds.ts

@@ -45,7 +45,7 @@ const STORAGE_VERSION = 1;
 const MAX_ROUNDS = 50;
 const MAX_VARIANTS = 10;
 const STORAGE_PREFIX = 'easyflow:agent-tryout-raw-rounds';
-const PERSIST_DEBOUNCE_MS = 300;
+const PERSIST_DEBOUNCE_MS = 500;
 const memorySessions = new Map<string, AgentTryoutRawRound[]>();
 
 function createRoundId() {
@@ -275,6 +275,45 @@ function selectedVariant(round: AgentTryoutRawRound) {
   );
 }
 
+function streamingPayloadKey(event: AgentTryoutRuntimeEvent) {
+  if (event.domain !== 'LLM') {
+    return undefined;
+  }
+  let candidates: string[] = [];
+  if (event.type === 'MESSAGE') {
+    candidates = ['delta'];
+  } else if (event.type === 'THINKING') {
+    candidates = ['reasoning', 'delta', 'text'];
+  }
+  return candidates.find((key) => typeof event.payload[key] === 'string');
+}
+
+function appendRuntimeEvent(
+  variant: AgentTryoutRawVariant,
+  event: AgentTryoutRuntimeEvent,
+) {
+  const previous = variant.runtimeEvents[variant.runtimeEvents.length - 1];
+  const currentKey = streamingPayloadKey(event);
+  const previousKey = previous ? streamingPayloadKey(previous) : undefined;
+  if (
+    previous &&
+    previous.domain === event.domain &&
+    previous.type === event.type &&
+    currentKey &&
+    currentKey === previousKey
+  ) {
+    previous.payload = {
+      ...previous.payload,
+      ...event.payload,
+      [currentKey]:
+        asText(previous.payload[currentKey]) +
+        asText(event.payload[currentKey]),
+    };
+    return;
+  }
+  variant.runtimeEvents.push(event);
+}
+
 function visibleText(item: ChatTimelineMessageItem) {
   return item.parts
     .filter((part) => part.type === 'text')
@@ -529,6 +568,7 @@ function projectEventToTimeline(
       payload.toolDisplayName ?? rawToolName ?? '工具',
     );
     const asyncTool = payload.asyncTool === true;
+    const taskInput = asRecord(payload.input ?? payload.toolInput);
     ChatTimelineBuilder.upsertToolCall(items, {
       input: payload.input ?? payload.toolInput,
       output: asyncTool
@@ -549,6 +589,12 @@ function projectEventToTimeline(
         variantIndex,
         'knowledge-retrieval',
       ),
+      sourceToolCallId: asyncTool
+        ? asText(payload.sourceToolCallId ?? payload.source_tool_call_id)
+        : undefined,
+      taskId: asyncTool
+        ? asText(payload.taskId ?? taskInput.taskId ?? taskInput.task_id)
+        : undefined,
       toolCallId: asText(
         payload.toolCallId ??
           payload.taskId ??
@@ -650,11 +696,6 @@ export function useAgentTryoutRawRounds(options: {
   }
 
   function schedulePersist() {
-    const key = storageKey(options.mode, options.sessionId);
-    memorySessions.set(
-      key,
-      [...rounds.values()].map((item) => clone(item)),
-    );
     if (persistTimer) {
       return;
     }
@@ -737,7 +778,7 @@ export function useAgentTryoutRawRounds(options: {
       payload: event.payload || {},
       type: event.type.toUpperCase(),
     };
-    variant.runtimeEvents.push(runtimeEvent);
+    appendRuntimeEvent(variant, runtimeEvent);
     if (runtimeEvent.domain === 'SYSTEM' && runtimeEvent.type === 'DONE') {
       variant.status = 'completed';
       round.status = 'completed';
@@ -754,9 +795,23 @@ export function useAgentTryoutRawRounds(options: {
       runtimeEvent.domain === 'ERROR'
     ) {
       persistNow();
-      return;
+      return runtimeEvent;
     }
     schedulePersist();
+    return runtimeEvent;
+  }
+
+  function projectEvent(
+    items: ChatTimelineItem[],
+    roundId: string,
+    event: AgentTryoutRuntimeEvent,
+  ) {
+    const round = rounds.get(roundId);
+    const variant = round && selectedVariant(round);
+    if (!round || !variant) {
+      return;
+    }
+    projectEventToTimeline(items, event, round.roundId, variant.variantIndex);
   }
 
   function completeRound(roundId: string) {
@@ -849,6 +904,7 @@ export function useAgentTryoutRawRounds(options: {
     createRound,
     currentVariant,
     getPrompt,
+    projectEvent,
     recordEvent,
     regenerateRound,
     selectVariant,

easyflow-ui-admin/app/src/views/ai/agents/composables/useAgentTryoutStream.ts

@@ -114,6 +114,7 @@ export function useAgentTryoutStream() {
     const sessionChanged = activeSessionId !== sessionId;
     activeSessionId = sessionId;
     if (!rawRounds || sessionChanged) {
+      rawRounds?.flush();
       rawRounds = useAgentTryoutRawRounds({
         mode: 'draft',
         sessionId,
@@ -174,12 +175,18 @@ export function useAgentTryoutStream() {
     const payload = envelope.payload || {};
 
     if (activeRoundId) {
-      rawRounds?.recordEvent(activeRoundId, {
+      const runtimeEvent = rawRounds?.recordEvent(activeRoundId, {
         domain,
         payload,
         type,
       });
-      rebuildTimeline();
+      if (runtimeEvent) {
+        rawRounds?.projectEvent(
+          timelineItems.value,
+          activeRoundId,
+          runtimeEvent,
+        );
+      }
     }
 
     if (domain === 'LLM' && type === 'MESSAGE') {
@@ -340,12 +347,23 @@ export function useAgentTryoutStream() {
     finishStoppedRun();
   }
 
+  function dispose() {
+    if (loading.value) {
+      userStopped = true;
+      sseClient.abort();
+      finishStoppedRun();
+      return;
+    }
+    rawRounds?.flush();
+  }
+
   return {
     loading,
     clearDraftSession,
     markToolApproving,
     markToolRejected,
     copyMessageText,
+    dispose,
     regenerateDraft,
     selectVariant,
     syncDraftContext,