From dfe4d42b491f69f2a4a8b2f2ce9cc34e9c498a01 Mon Sep 17 00:00:00 2001 From: Senor-Liu <71912659+Senor-Liu@users.noreply.github.com> Date: Mon, 24 Aug 2026 13:53:46 +0800 Subject: [PATCH] =?UTF-8?q?merge:=20=E5=90=88=E5=B9=B6=E4=BB=A3=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../components/SuperAgentConversation.jsx | 5 +- .../sections/superAgent/container/index.jsx | 21 +- .../superAgent/hooks/useSuperAgentStream.js | 228 +++++++++++++++++- .../sections/superAgent/utils/bodyExport.js | 10 + 4 files changed, 255 insertions(+), 9 deletions(-) diff --git a/web/client/src/sections/superAgent/components/SuperAgentConversation.jsx b/web/client/src/sections/superAgent/components/SuperAgentConversation.jsx index 6155553..648ba69 100644 --- a/web/client/src/sections/superAgent/components/SuperAgentConversation.jsx +++ b/web/client/src/sections/superAgent/components/SuperAgentConversation.jsx @@ -641,11 +641,12 @@ const SuperAgentConversation = ({ item.runLog.entries.some((entry) => entry?.status === "running")); const shouldShowActiveLoading = !hasStructuredCard && - (showLoadingIndicator || (isStreamingAnswer && hasRunningTool)); + showLoadingIndicator && + (isStreamingAnswer || hasRunningTool); const shouldShowLoading = !hasStructuredCard && !item.content && - (Boolean(item.loadingText) || shouldShowActiveLoading); + ((Boolean(item.loadingText) && showLoadingIndicator) || shouldShowActiveLoading); const hasBubbleContent = item.type !== "content" && Boolean(item.title || hasMessageContent || shouldShowLoading); diff --git a/web/client/src/sections/superAgent/container/index.jsx b/web/client/src/sections/superAgent/container/index.jsx index 4914ac0..e8ef86e 100644 --- a/web/client/src/sections/superAgent/container/index.jsx +++ b/web/client/src/sections/superAgent/container/index.jsx @@ -553,9 +553,15 @@ const SuperAgent = () => { const commitProgressTask = useCallback((sessionId, progressData, payload = {}) => { const rawProgress = Number(progressData?.progress); + const progressStatus = String(progressData?.status || "").toLowerCase(); const sessionStatus = String( payload.status || payload.task_status || "" ).toLowerCase(); + const isCompletedProgress = [ + "completed", + "done", + "success", + ].includes(progressStatus); const hasTerminalSessionStatus = TERMINAL_SESSION_STATUSES.has(sessionStatus); const normalizedProgressData = { ...progressData, @@ -563,8 +569,9 @@ const SuperAgent = () => { session_status: payload.session_status || payload.sessionStatus || - progressData.session_status || - "", + (isCompletedProgress + ? "completed" + : progressData.session_status || ""), ...(hasTerminalSessionStatus ? { status: sessionStatus } : {}), progress: Number.isFinite(rawProgress) ? Math.round(rawProgress) : 0, }; @@ -849,9 +856,19 @@ const SuperAgent = () => { ? restoredSessionTask : null; //[正文任务状态只来自当前 useStream thread 和其业务事件] + const latestBusinessProgressTask = + String( + latestProgressTaskRef.current?.session_id || + latestProgressTaskRef.current?.sessionId || + "" + ) === String(activeSessionId || "") + ? latestProgressTaskRef.current + : null; const runtimeTaskSources = { ...(activeRestoredSessionTask || {}), ...(activeStreamTask || {}), + // 业务 custom 事件比 SDK 可能滞后的 values 快照更新更快，终态不能被旧 task 覆盖。 + ...(latestBusinessProgressTask || {}), }; const runtimeTaskSourceType = String( runtimeTaskSources?.task_type || runtimeTaskSources?.taskType || "" diff --git a/web/client/src/sections/superAgent/hooks/useSuperAgentStream.js b/web/client/src/sections/superAgent/hooks/useSuperAgentStream.js index e7749be..1f31090 100644 --- a/web/client/src/sections/superAgent/hooks/useSuperAgentStream.js +++ b/web/client/src/sections/superAgent/hooks/useSuperAgentStream.js @@ -388,6 +388,136 @@ const getEventPayload = (event) => event?.params?.data ?? event; +//[刷新回放]保留当前会话的消息快照，避免 hydrate 返回的旧快照覆盖 SSE 增量。 +const mergeStreamMessageSnapshots = (previousMessages = [], currentMessages = []) => { + const previous = Array.isArray(previousMessages) ? previousMessages : []; + const current = Array.isArray(currentMessages) ? currentMessages : []; + if (!previous.length) return current; + if (!current.length) return previous; + + const currentById = new Map( + current + .filter((message) => message?.id) + .map((message) => [String(message.id), message]) + ); + const merged = []; + const usedIds = new Set(); + + previous.forEach((message) => { + const messageId = String(message?.id || ""); + if (!messageId) return; + merged.push(currentById.get(messageId) || message); + usedIds.add(messageId); + }); + + current.forEach((message) => { + const messageId = String(message?.id || ""); + if (!messageId || usedIds.has(messageId)) return; + merged.push(message); + }); + + return merged; +}; + +//[刷新回放]直接消费 root messages 事件，保证网络层已有数据时页面同步更新。 +const applyMessageEventToState = (currentState, event) => { + const data = event?.params?.data || {}; + const eventType = String(data.event || ""); + const nextMessages = new Map(currentState.messages); + let activeMessageId = currentState.activeMessageId; + const messageId = String(data.id || activeMessageId || "").trim(); + if (!messageId) return currentState; + + const previousMessage = nextMessages.get(messageId) || { + id: messageId, + type: "ai", + role: "ai", + content: "", + tool_calls: [], + }; + const nextMessage = { + ...previousMessage, + tool_calls: Array.isArray(previousMessage.tool_calls) + ? [...previousMessage.tool_calls] + : [], + }; + + if (eventType === "message-start") { + activeMessageId = messageId; + nextMessage.role = data.role || nextMessage.role || "ai"; + nextMessage.type = nextMessage.role; + } + + if (eventType === "content-block-delta") { + const delta = data.delta || {}; + if (delta.type === "text-delta") { + nextMessage.content = `${String(nextMessage.content || "")}${String(delta.text || "")}`; + } + if (delta.type === "tool-call-delta") { + const toolCallIndex = Number(data.index || 0); + const previousToolCall = nextMessage.tool_calls[toolCallIndex] || {}; + const deltaArgs = delta.args; + const nextArgs = + previousToolCall.args && + typeof previousToolCall.args === "object" && + deltaArgs && + typeof deltaArgs === "object" + ? { ...previousToolCall.args, ...deltaArgs } + : deltaArgs ?? previousToolCall.args; + nextMessage.tool_calls[toolCallIndex] = { + ...previousToolCall, + args: nextArgs, + status: "running", + }; + } + } + + if (eventType === "content-block-finish") { + const content = data.content || {}; + if (content.type === "text") { + nextMessage.content = content.text || nextMessage.content || ""; + } + if (content.type === "tool_call") { + const toolCallIndex = Number(data.index || 0); + nextMessage.tool_calls[toolCallIndex] = { + ...(nextMessage.tool_calls[toolCallIndex] || {}), + ...content, + status: "success", + }; + } + } + + nextMessages.set(messageId, nextMessage); + return { + messages: nextMessages, + activeMessageId, + }; +}; + +const RUNTIME_STATE_REFRESH_EVENT_TYPES = new Set([ + "generation.completed", + "generation.failed", + "generation.cancelled", + "generation.error", +]); + +const shouldRefreshRuntimeState = ({ + method = "", + payload = {}, + isResuming = false, +} = {}) => { + if (method === "messages") return isResuming; + if (method === "custom") { + return RUNTIME_STATE_REFRESH_EVENT_TYPES.has(String(payload.type || "")); + } + if (method === "lifecycle") { + return ["completed", "failed", "aborted"].includes( + String(payload.event || "") + ); + } + return false; +}; + const getEventDedupeKey = (event) => { if (event?.event_id) return String(event.event_id); const payload = getEventPayload(event) || {}; @@ -431,12 +561,20 @@ const useSuperAgentStream = ({ const [runtimeCheckpointDeliveries, setRuntimeCheckpointDeliveries] = useState([]); const [runtimeInterrupts, setRuntimeInterrupts] = useState([]); const [dismissedInterruptIds, setDismissedInterruptIds] = useState(new Set()); + const runtimeStateRefreshTimerRef = useRef(null); + const runtimeStateRefreshPendingRef = useRef(false); const resolvedInterruptIdsRef = useRef(new Set()); const resolvedInterruptInteractionsRef = useRef(new Set()); const [currentInterruptId, setCurrentInterruptId] = useState(""); const [hydratedThreadId, setHydratedThreadId] = useState(""); const [transportStateRevision, setTransportStateRevision] = useState(0); const eventRequestControllersRef = useRef(new Map()); + const eventMessageStateRef = useRef({ + sessionId: "", + activeMessageId: "", + messages: new Map(), + }); + const [eventStreamMessages, setEventStreamMessages] = useState([]); const pendingSubmitThreadIdRef = useRef(""); const hydrationSessionRef = useRef({ sessionId: "", @@ -529,6 +667,26 @@ const useSuperAgentStream = ({ transport: eventTransport, }); + //[恢复中断后同步 thread state][避免 SSE 已收到但消息投影仍停留在旧快照] + const scheduleRuntimeStateRefresh = useCallback((delay = 120, force = false) => { + if (runtimeStateRefreshTimerRef.current != null) return; + if (!force && runtimeStateRefreshPendingRef.current) return; + if (!force) runtimeStateRefreshPendingRef.current = true; + + runtimeStateRefreshTimerRef.current = setTimeout(async () => { + runtimeStateRefreshTimerRef.current = null; + if (!String(eventTransport.threadId || "").trim()) { + runtimeStateRefreshPendingRef.current = false; + return; + } + try { + await eventTransport.getState().catch(() => undefined); + } finally { + runtimeStateRefreshPendingRef.current = false; + } + }, delay); + }, [eventTransport]); + const handleBusinessEvent = useCallback((event) => { const eventKey = getEventDedupeKey(event); if (handledEventKeysRef.current.has(eventKey)) return; @@ -585,6 +743,11 @@ const useSuperAgentStream = ({ useEffect(() => { const sessionKey = String(activeSessionId || "").trim(); return () => { + if (runtimeStateRefreshTimerRef.current != null) { + clearTimeout(runtimeStateRefreshTimerRef.current); + runtimeStateRefreshTimerRef.current = null; + } + runtimeStateRefreshPendingRef.current = false; if (sessionKey) { const sessionController = eventRequestControllersRef.current.get(sessionKey); sessionController?.abort(); @@ -616,11 +779,18 @@ const useSuperAgentStream = ({ setIsBusinessRunning(false); setRuntimeCheckpointDeliveries([]); setRuntimeInterrupts([]); + runtimeStateRefreshPendingRef.current = false; setDismissedInterruptIds(new Set()); resolvedInterruptIdsRef.current.clear(); resolvedInterruptInteractionsRef.current.clear(); setCurrentInterruptId(""); setHydratedThreadId(""); + eventMessageStateRef.current = { + sessionId: String(activeSessionId || ""), + activeMessageId: "", + messages: new Map(), + }; + setEventStreamMessages([]); }, [activeSessionId]); useEffect(() => { @@ -649,6 +819,27 @@ const useSuperAgentStream = ({ transportStateRevision, ]); + const handleMessageEvent = useCallback((event) => { + if (event?.method !== "messages") return; + const namespace = event?.params?.namespace; + if (Array.isArray(namespace) && namespace.length > 0) return; + + const sessionId = String(activeSessionIdRef.current || "").trim(); + if (!sessionId) return; + if (eventMessageStateRef.current.sessionId !== sessionId) { + eventMessageStateRef.current = { + sessionId, + activeMessageId: "", + messages: new Map(), + }; + } + + const nextState = applyMessageEventToState(eventMessageStateRef.current, event); + if (nextState === eventMessageStateRef.current) return; + eventMessageStateRef.current = nextState; + setEventStreamMessages([...nextState.messages.values()]); + }, []); + const checkpointDeliveries = useMemo(() => { const deliveries = new Map(); const hydratedDeliveries = [ @@ -707,6 +898,18 @@ const useSuperAgentStream = ({ // 切换会话后旧 Thread 的监听器可能在清理前再收到一个事件， // 不能把旧会话的事件写入新会话的流式消息状态。 if (activeSessionIdRef.current !== listenerSessionId) return; + handleMessageEvent(event); + const eventPayload = getEventPayload(event) || {}; + if (shouldRefreshRuntimeState({ + method: event?.method, + payload: eventPayload, + isResuming: interruptResumeRef.current.active, + })) { + const isTerminalEvent = + event?.method === "custom" || event?.method === "lifecycle"; + const refreshDelay = event?.method === "messages" ? 160 : 0; + scheduleRuntimeStateRefresh(refreshDelay, isTerminalEvent); + } if (event?.method === "input.requested") { const data = event.params?.data || {}; const interruptId = String(data.interrupt_id || "").trim(); @@ -786,7 +989,14 @@ const useSuperAgentStream = ({ if (!BUSINESS_CHANNELS.includes(event?.method)) return; handleBusinessEvent(event); }); - }, [activeSessionId, handleBusinessEvent, stream.threadId]); + }, [ + activeSessionId, + handleBusinessEvent, + handleMessageEvent, + scheduleRuntimeStateRefresh, + stream.threadId, + transportStateRevision, + ]); const prepareForSessionSwitch = useCallback(() => { // 先让旧 thread 的事件监听失效。activeSessionId 在 React 下一次渲染 @@ -883,6 +1093,7 @@ const useSuperAgentStream = ({ setIsLocalRunActive(true); localRunHasSeenLoadingRef.current = false; interruptResumeRef.current = { active: true, running: false }; + runtimeStateRefreshPendingRef.current = false; pendingInterruptRef.current = false; awaitingInterruptRef.current = false; setRunPhase("resuming"); @@ -904,6 +1115,7 @@ const useSuperAgentStream = ({ interruptId: options.interruptId, namespace: options.namespace, }).then((result) => { + scheduleRuntimeStateRefresh(100); if (interruptId) { resolvedInterruptIdsRef.current.add(interruptId); const interactionKey = getInterruptInteractionKey( @@ -940,7 +1152,7 @@ const useSuperAgentStream = ({ setIsRespondingInterrupt(false); throw error; }); - }, [stream]); + }, [scheduleRuntimeStateRefresh, stream]); const stopRun = useCallback(async () => { localRunActiveRef.current = false; @@ -1087,9 +1299,15 @@ const useSuperAgentStream = ({ !hydratedTailIds.every((messageId) => rawStreamMessageIds.has(messageId)))); const shouldPreferHydratedState = shouldPreferHydratedRuntimeState || shouldPreferHydratedMessages; - const streamMessages = shouldPreferHydratedMessages - ? hydratedMessages - : rawStreamMessages; + const projectedMessages = shouldPreferHydratedMessages + ? hydratedMessages + : rawStreamMessages; + const streamMessages = mergeStreamMessageSnapshots( + projectedMessages, + isThreadBound && eventMessageStateRef.current.sessionId === currentSessionId + ? eventStreamMessages + : [] + ); const streamToolCalls = isStreamProjectionReady ? stream.toolCalls : []; const streamValues = shouldPreferHydratedState ? hydratedState?.values || {} diff --git a/web/client/src/sections/superAgent/utils/bodyExport.js b/web/client/src/sections/superAgent/utils/bodyExport.js index c7a41cf..336d9fc 100644 --- a/web/client/src/sections/superAgent/utils/bodyExport.js +++ b/web/client/src/sections/superAgent/utils/bodyExport.js @@ -303,6 +303,16 @@ export const stripDuplicateLeadingBodyHeading = (content, headingTitle) => { const headingIdentity = getPlainHeadingText(headingTitle); if (!html.trim() || !headingIdentity) return html; + //[题注保护][题注名称可能与章节标题相同，不能按普通首段标题删除] + const leadingBlockMatch = html.match(/^\s*<(?:p|div|figcaption)\b[^>]*>/i); + const leadingBlockAttributes = leadingBlockMatch?.[0] || ""; + const isLeadingCaption = + /data-ai-caption\s*=\s*["']1["']/i.test(leadingBlockAttributes) || + /class\s*=\s*["'][^"']*\bai-caption\b[^"']*["']/i.test( + leadingBlockAttributes + ); + if (isLeadingCaption) return html; + const markdownHeadingMatch = html.match(/^\s{0,3}#{1,6}\s+(.+?)(?:\r?\n|$)/); if (markdownHeadingMatch) { const firstHeadingIdentity = getPlainHeadingText(markdownHeadingMatch[1]);