From 23369586c0d7d8b123b01b9164c0fbae2275eef1 Mon Sep 17 00:00:00 2001 From: chanx <1243304602@qq.com> Date: Mon, 10 Aug 2026 17:29:24 +0800 Subject: [PATCH] fix(chat): preserve in-flight SSE stream when switching conversations (#18037) --- .../chat/chat-box/single-chat-box.tsx | 9 +- .../pages/next-chats/hooks/use-click-card.ts | 10 +- .../next-chats/hooks/use-send-chat-message.ts | 166 +++++++++++++++--- 3 files changed, 153 insertions(+), 32 deletions(-) diff --git a/web/src/pages/next-chats/chat/chat-box/single-chat-box.tsx b/web/src/pages/next-chats/chat/chat-box/single-chat-box.tsx index a598d7f3ea..9fa115ad9e 100644 --- a/web/src/pages/next-chats/chat/chat-box/single-chat-box.tsx +++ b/web/src/pages/next-chats/chat/chat-box/single-chat-box.tsx @@ -42,6 +42,7 @@ export function SingleChatBox({ handleUploadFile, removeFile, setDerivedMessages, + activeStreamsRef, } = useSendMessage(controller); const { data: userInfo } = useFetchUserInfo(); const { data: currentDialog } = useFetchChat(); @@ -56,6 +57,12 @@ export function SingleChatBox({ const showInternet = useShowInternet(); useEffect(() => { + // Don't let backend data overwrite local streaming state while SSE is + // in-flight for this conversation. The server hasn't persisted the latest + // answer yet, and applying conversation.messages would discard the + // in-progress answer. + if (activeStreamsRef.current.has(conversationId)) return; + const messages = conversation?.messages; if (Array.isArray(messages)) { setDerivedMessages((prevMessages) => { @@ -76,7 +83,7 @@ export function SingleChatBox({ })); }); } - }, [conversation?.messages, setDerivedMessages]); + }, [conversation?.messages, conversationId, setDerivedMessages, activeStreamsRef]); useEffect(() => { // Clear the message list after deleting the conversation. diff --git a/web/src/pages/next-chats/hooks/use-click-card.ts b/web/src/pages/next-chats/hooks/use-click-card.ts index fc5fe2ebfe..f5811d723c 100644 --- a/web/src/pages/next-chats/hooks/use-click-card.ts +++ b/web/src/pages/next-chats/hooks/use-click-card.ts @@ -15,9 +15,15 @@ export function useHandleClickConversationCard() { const handleConversationCardClick = useCallback( (conversationId: string, isNew: boolean) => { setConversationBoth(conversationId, isNew ? 'true' : ''); - stopOutputMessage(); + // Switch to a fresh controller WITHOUT aborting the previous one. + // Aborting the in-flight completion request can cancel server-side + // persistence of the just-asked question, so switching conversations + // right after asking would lose the question. Let the old SSE finish + // in the background; the unmounted SingleChatBox's setState becomes + // a no-op and the server keeps the message. + setController(new AbortController()); }, - [setConversationBoth, stopOutputMessage], + [setConversationBoth], ); return { controller, handleConversationCardClick, stopOutputMessage }; diff --git a/web/src/pages/next-chats/hooks/use-send-chat-message.ts b/web/src/pages/next-chats/hooks/use-send-chat-message.ts index 49895d4530..174ef8af65 100644 --- a/web/src/pages/next-chats/hooks/use-send-chat-message.ts +++ b/web/src/pages/next-chats/hooks/use-send-chat-message.ts @@ -7,16 +7,60 @@ import { useSendMessageWithSse, } from '@/hooks/logic-hooks'; import { useGetChatSearchParams } from '@/hooks/use-chat-request'; -import { IMessage } from '@/interfaces/database/chat'; +import { IAnswer, IMessage } from '@/interfaces/database/chat'; import api from '@/utils/api'; -import { trim } from 'lodash'; -import { useCallback, useEffect } from 'react'; +import { buildMessageUuid } from '@/utils/chat'; +import { omit, trim } from 'lodash'; +import { useCallback, useEffect, useRef } from 'react'; import { useParams } from 'react-router'; import { v4 as uuid } from 'uuid'; import { useCreateConversationBeforeSendMessage } from './use-chat-url'; import { useFindPrologueFromDialogList } from './use-select-conversation-list'; import { useUploadFile } from './use-upload-file'; +// Append an answer to a messages array (mirrors addNewestAnswer logic in +// useSelectDerivedMessages) for updating a background conversation's cache +// without touching the currently displayed derivedMessages. +function appendAnswerToMessages( + messages: IMessage[], + answer: IAnswer, +): IMessage[] { + return [ + ...(messages.slice(0, -1) ?? []), + { + role: MessageType.Assistant, + content: answer.answer, + reference: answer.reference, + id: buildMessageUuid({ id: answer.id, role: MessageType.Assistant }), + prompt: answer.prompt, + audio_binary: answer.audio_binary, + ...omit(answer, 'reference'), + }, + ]; +} + +// Append a user question + empty assistant placeholder to a messages array +// (mirrors addNewestQuestion logic) for updating a background conversation's +// cache when the user asks a question right before switching conversations. +function appendQuestionToMessages( + messages: IMessage[], + message: IMessage, +): IMessage[] { + return [ + ...messages, + { + ...message, + id: buildMessageUuid(message), + }, + { + role: MessageType.Assistant, + content: '', + conversationId: message.conversationId, + id: buildMessageUuid({ ...message, role: MessageType.Assistant }), + }, + ]; +} + export const useSelectNextMessages = () => { const { scrollRef, @@ -33,6 +77,26 @@ export const useSelectNextMessages = () => { const { id: dialogId } = useParams(); const prologue = useFindPrologueFromDialogList(); + // Per-conversation message cache so switching conversations preserves the + // local derivedMessages (including in-flight SSE answers) instead of wiping + // them. Keyed by conversationId. + const messagesCacheRef = useRef>({}); + // Tracks conversations with an in-flight SSE stream so the backend sync + // effect in SingleChatBox can avoid overwriting local streaming state. + const activeStreamsRef = useRef>(new Set()); + // Tracks which conversationId the current derivedMessages belongs to, so + // the cache-write effect doesn't mistakenly write the previous conversation's + // messages into the newly switched conversation's cache entry during the + // render where conversationId has changed but derivedMessages hasn't yet. + const messagesConversationIdRef = useRef(conversationId); + + // On conversation switch, restore cached messages (or start empty). + useEffect(() => { + messagesConversationIdRef.current = conversationId; + const cached = messagesCacheRef.current[conversationId]; + setDerivedMessages(cached ?? []); + }, [conversationId, setDerivedMessages]); + const addPrologue = useCallback(() => { if (dialogId !== '' && isNew === 'true') { const nextMessage = { @@ -50,6 +114,13 @@ export const useSelectNextMessages = () => { addPrologue(); }, [addPrologue]); + // Persist current derivedMessages back into the cache for the conversation + // they actually belong to (not necessarily the URL's current conversationId + // during a switch). + useEffect(() => { + messagesCacheRef.current[messagesConversationIdRef.current] = derivedMessages; + }, [derivedMessages]); + return { scrollRef, messageContainerRef, @@ -60,6 +131,8 @@ export const useSelectNextMessages = () => { removeMessageById, removeMessagesAfterCurrentMessage, setDerivedMessages, + messagesCacheRef, + activeStreamsRef, }; }; @@ -82,6 +155,8 @@ export const useSendMessage = (controller: AbortController) => { removeMessageById, removeMessagesAfterCurrentMessage, setDerivedMessages, + messagesCacheRef, + activeStreamsRef, } = useSelectNextMessages(); const sendMessage = useCallback( @@ -97,29 +172,34 @@ export const useSendMessage = (controller: AbortController) => { messages?: IMessage[]; } & NextMessageInputOnPressEnterParameter) => { const sessionId = currentConversationId ?? conversationId; - const res = await send( - api.completionUrl, - { - chat_id: chatId, - session_id: sessionId, - // An explicitly provided list is authoritative, even when empty - // (e.g. regenerating the first question must truncate history). - messages: [ - ...(Array.isArray(messages) ? messages : (derivedMessages ?? [])), - message, - ], - pass_all_history_messages: true, - reasoning: Number(enableThinking), - internet: enableInternet, - }, - controller, - ); + activeStreamsRef.current.add(sessionId); + try { + const res = await send( + api.completionUrl, + { + chat_id: chatId, + session_id: sessionId, + // An explicitly provided list is authoritative, even when empty + // (e.g. regenerating the first question must truncate history). + messages: [ + ...(Array.isArray(messages) ? messages : (derivedMessages ?? [])), + message, + ], + pass_all_history_messages: true, + reasoning: Number(enableThinking), + internet: enableInternet, + }, + controller, + ); - if (res && (res?.response.status !== 200 || res?.data?.code !== 0)) { - // cancel loading - setValue(message.content); - console.info('removeLatestMessage111'); - removeLatestMessage(); + if (res && (res?.response.status !== 200 || res?.data?.code !== 0)) { + // cancel loading + setValue(message.content); + console.info('removeLatestMessage111'); + removeLatestMessage(); + } + } finally { + activeStreamsRef.current.delete(sessionId); } }, [ @@ -130,6 +210,7 @@ export const useSendMessage = (controller: AbortController) => { setValue, send, controller, + activeStreamsRef, ], ); @@ -159,13 +240,23 @@ export const useSendMessage = (controller: AbortController) => { const id = uuid(); - addNewestQuestion({ + const questionMessage = { content: value, files: files, id, role: MessageType.User, conversationId: targetConversationId, - }); + }; + // The await on createConversationBeforeSendMessage above can span a + // conversation switch. Route the question to the conversation it was + // asked in, not whichever is currently displayed. + if (targetConversationId === conversationId) { + addNewestQuestion(questionMessage); + } else { + const cached = messagesCacheRef.current[targetConversationId] ?? []; + messagesCacheRef.current[targetConversationId] = + appendQuestionToMessages(cached, questionMessage); + } if (done) { setValue(''); @@ -209,15 +300,30 @@ export const useSendMessage = (controller: AbortController) => { setValue, sendMessage, messageContainerRef, + conversationId, + messagesCacheRef, ], ); useEffect(() => { // #1289 - if (answer.answer && conversationId && isNew !== 'true') { + if (!answer.answer || isNew === 'true') return; + // Route the answer to the conversation it belongs to. answer.conversationId + // comes from the request body's session_id (set in useSendMessageWithSse.send), + // NOT from the URL, so it stays correct even after switching conversations. + // This prevents a background stream's answer from leaking into the currently + // displayed conversation, and keeps the background conversation's cache + // updating so switching back shows the streamed answer. + const targetId = answer.conversationId; + if (!targetId) return; + if (targetId === conversationId) { addNewestAnswer(answer); + } else { + const cached = messagesCacheRef.current[targetId] ?? []; + messagesCacheRef.current[targetId] = + appendAnswerToMessages(cached, answer); } - }, [answer, addNewestAnswer, conversationId, isNew]); + }, [answer, addNewestAnswer, conversationId, isNew, messagesCacheRef]); return { handlePressEnter, @@ -234,5 +340,7 @@ export const useSendMessage = (controller: AbortController) => { isUploading, removeFile, setDerivedMessages, + messagesCacheRef, + activeStreamsRef, }; };