Skip to content

赵月辉 / fastgpt-migrated

Go to a project
Commit 910010fe by Ryo Committed by GitHub
Options

fix: scope chat resume to current app (#6860) 

* fix: scope chat resume to current app

Abort stale resume streams when switching apps and only resume after the current app/chat init state is aligned.

Made-with: Cursor

* refactor: centralize resume stream cancellation

Ensure only one resume stream is active per client tab and discard queued resume messages after abort.

Made-with: Cursor

* fix: make sure controller unique

* chore: remove redundant effect
parent 9a6d774c
Showing with 88 additions and 28 deletions
projects/app/src/components/core/chat/ChatContainer/ChatBox/index.tsx
...@@ -154,7 +154,7 @@ const ChatBox = ({ ...@@ -154,7 +154,7 @@ const ChatBox = ({
const questionGuideController = useRef(new AbortController()); const questionGuideController = useRef(new AbortController());
const pluginController = useRef(new AbortController()); const pluginController = useRef(new AbortController());
const resumeController = useRef<AbortController>(); const resumeController = useRef<AbortController>();
const resumedChatIdRef = useRef<string>(); const resumedChatTargetRef = useRef<string>();
const [isLoading, setIsLoading] = useState(false); const [isLoading, setIsLoading] = useState(false);
const [feedbackId, setFeedbackId] = useState<string>(); const [feedbackId, setFeedbackId] = useState<string>();
...@@ -180,6 +180,8 @@ const ChatBox = ({ ...@@ -180,6 +180,8 @@ const ChatBox = ({
const appId = useContextSelector(WorkflowRuntimeContext, (v) => v.appId); const appId = useContextSelector(WorkflowRuntimeContext, (v) => v.appId);
const chatId = useContextSelector(WorkflowRuntimeContext, (v) => v.chatId); const chatId = useContextSelector(WorkflowRuntimeContext, (v) => v.chatId);
const activeAppIdRef = useRef<string | undefined>(appId);
activeAppIdRef.current = appId;
const activeChatIdRef = useRef<string | undefined>(chatId); const activeChatIdRef = useRef<string | undefined>(chatId);
activeChatIdRef.current = chatId; activeChatIdRef.current = chatId;
const outLinkAuthData = useContextSelector(WorkflowRuntimeContext, (v) => v.outLinkAuthData); const outLinkAuthData = useContextSelector(WorkflowRuntimeContext, (v) => v.outLinkAuthData);
...@@ -198,18 +200,21 @@ const ChatBox = ({ ...@@ -198,18 +200,21 @@ const ChatBox = ({
const syncSidebarChatGenerateStatus = useMemoizedFn( const syncSidebarChatGenerateStatus = useMemoizedFn(
( (
status: ChatGenerateStatusEnum, status: ChatGenerateStatusEnum,
options?: { hasBeenRead?: boolean; targetChatId?: string } options?: { hasBeenRead?: boolean; targetAppId?: string; targetChatId?: string }
) => { ) => {
const targetAppId = options?.targetAppId ?? appId;
if (targetAppId !== appId) return;
const targetChatId = options?.targetChatId ?? chatId; const targetChatId = options?.targetChatId ?? chatId;
if (!targetChatId) return; if (!targetChatId) return;
setHistories((prev) => { setHistories((prev) => {
const idx = prev.findIndex((h) => h.chatId === targetChatId); const idx = prev.findIndex((h) => h.chatId === targetChatId && h.appId === targetAppId);
if (idx === -1) { if (idx === -1) {
queueMicrotask(loadHistories); queueMicrotask(loadHistories);
return [ return [
{ {
chatId: targetChatId, chatId: targetChatId,
appId, appId: targetAppId,
title: chatBoxData.title || t('common:core.chat.New Chat'), title: chatBoxData.title || t('common:core.chat.New Chat'),
customTitle: '', customTitle: '',
top: false, top: false,
...@@ -221,7 +226,7 @@ const ChatBox = ({ ...@@ -221,7 +226,7 @@ const ChatBox = ({
]; ];
} }
return prev.map((h) => return prev.map((h) =>
h.chatId === targetChatId h.chatId === targetChatId && h.appId === targetAppId
? { ? {
...h, ...h,
chatGenerateStatus: status, chatGenerateStatus: status,
...@@ -662,16 +667,26 @@ const ChatBox = ({ ...@@ -662,16 +667,26 @@ const ChatBox = ({
}); });
}); });
const isActiveResumeTarget = useMemoizedFn(
({ appId, chatId }: { appId: string; chatId: string }) =>
activeAppIdRef.current === appId && activeChatIdRef.current === chatId
);
const getResumeUnavailablePlaceholderText = useMemoizedFn(() => const getResumeUnavailablePlaceholderText = useMemoizedFn(() =>
t('chat:resume_placeholder_generating') t('chat:resume_placeholder_generating')
); );
const upsertResumeAiPlaceholder = useMemoizedFn( const upsertResumeAiPlaceholder = useMemoizedFn(
(responseChatId: string, text = '', status: `${ChatStatusEnum}` = ChatStatusEnum.loading) => { (
responseChatId: string,
text = '',
status: `${ChatStatusEnum}` = ChatStatusEnum.loading,
options?: { resetExistingValue?: boolean }
) => {
setChatRecords((state) => { setChatRecords((state) => {
const lastItem = state[state.length - 1]; const lastItem = state[state.length - 1];
if (lastItem?.dataId === responseChatId && lastItem.obj === ChatRoleEnum.AI) { if (lastItem?.dataId === responseChatId && lastItem.obj === ChatRoleEnum.AI) {
if (!text) { if (!text && !options?.resetExistingValue) {
return state; return state;
} }
...@@ -687,6 +702,7 @@ const ChatBox = ({ ...@@ -687,6 +702,7 @@ const ChatBox = ({
} }
} }
], ],
responseData: options?.resetExistingValue ? [] : item.responseData,
status, status,
...(status === ChatStatusEnum.finish ? { time: new Date() } : {}) ...(status === ChatStatusEnum.finish ? { time: new Date() } : {})
} }
...@@ -833,7 +849,7 @@ const ChatBox = ({ ...@@ -833,7 +849,7 @@ const ChatBox = ({
} }
]; ];
resumedChatIdRef.current = chatId; resumedChatTargetRef.current = `${appId}:${chatId}`;
setChatBoxData((state) => setChatBoxData((state) =>
state.chatId === chatId state.chatId === chatId
...@@ -1255,8 +1271,12 @@ const ChatBox = ({ ...@@ -1255,8 +1271,12 @@ const ChatBox = ({
useEffect(() => { useEffect(() => {
setQuestionGuide([]); setQuestionGuide([]);
setValue('chatStarted', false); setValue('chatStarted', false);
resumedChatIdRef.current = undefined; resumedChatTargetRef.current = undefined;
abortRequest('leave'); abortRequest('leave');
return () => {
abortRequest('leave');
};
}, [chatId, appId, abortRequest, setValue]); }, [chatId, appId, abortRequest, setValue]);
useEffect(() => { useEffect(() => {
...@@ -1267,20 +1287,24 @@ const ChatBox = ({ ...@@ -1267,20 +1287,24 @@ const ChatBox = ({
!appId || !appId ||
!chatId || !chatId ||
isChatting || isChatting ||
chatBoxData.appId !== appId ||
chatBoxData.chatId !== chatId ||
chatBoxData.chatGenerateStatus !== ChatGenerateStatusEnum.generating || chatBoxData.chatGenerateStatus !== ChatGenerateStatusEnum.generating ||
resumedChatIdRef.current === chatId resumedChatTargetRef.current === `${appId}:${chatId}`
) { ) {
return; return;
} }
resumedChatIdRef.current = chatId; resumedChatTargetRef.current = `${appId}:${chatId}`;
const resumeForAppId = appId;
const resumeForChatId = chatId; const resumeForChatId = chatId;
const responseChatId = resumeTargetAiDataId ?? getNanoid(24); const responseChatId = resumeTargetAiDataId ?? getNanoid(24);
const controller = new AbortController(); const controller = new AbortController();
resumeController.current = controller; resumeController.current = controller;
scrollToBottom('auto'); scrollToBottom('auto');
let resumeFinalStatus = ChatGenerateStatusEnum.done; let resumeFinalStatus = ChatGenerateStatusEnum.done;
let hasPreparedResumeAiRecord = false;
(async () => { (async () => {
try { try {
...@@ -1290,7 +1314,7 @@ const ChatBox = ({ ...@@ -1290,7 +1314,7 @@ const ChatBox = ({
outLinkAuthData, outLinkAuthData,
controller, controller,
onResumeUnavailable: () => { onResumeUnavailable: () => {
if (resumeForChatId !== activeChatIdRef.current) return; if (!isActiveResumeTarget({ appId: resumeForAppId, chatId: resumeForChatId })) return;
resumeFinalStatus = ChatGenerateStatusEnum.generating; resumeFinalStatus = ChatGenerateStatusEnum.generating;
upsertResumeAiPlaceholder( upsertResumeAiPlaceholder(
responseChatId, responseChatId,
...@@ -1299,15 +1323,18 @@ const ChatBox = ({ ...@@ -1299,15 +1323,18 @@ const ChatBox = ({
); );
}, },
onmessage: (message) => { onmessage: (message) => {
if (resumeForChatId !== activeChatIdRef.current) return; if (!isActiveResumeTarget({ appId: resumeForAppId, chatId: resumeForChatId })) return;
if (shouldCreateResumeAiPlaceholder(message.event)) { if (shouldCreateResumeAiPlaceholder(message.event)) {
upsertResumeAiPlaceholder(responseChatId); upsertResumeAiPlaceholder(responseChatId, '', ChatStatusEnum.loading, {
resetExistingValue: !hasPreparedResumeAiRecord
});
hasPreparedResumeAiRecord = true;
} }
generatingMessage(message); generatingMessage(message);
} }
}); });
if (resumeForChatId !== activeChatIdRef.current) return; if (!isActiveResumeTarget({ appId: resumeForAppId, chatId: resumeForChatId })) return;
if (completedChat) { if (completedChat) {
resumeFinalStatus = completedChat.chatGenerateStatus; resumeFinalStatus = completedChat.chatGenerateStatus;
...@@ -1362,7 +1389,7 @@ const ChatBox = ({ ...@@ -1362,7 +1389,7 @@ const ChatBox = ({
}); });
} catch (error) { } catch (error) {
if (controller.signal.aborted) return; if (controller.signal.aborted) return;
if (resumeForChatId !== activeChatIdRef.current) return; if (!isActiveResumeTarget({ appId: resumeForAppId, chatId: resumeForChatId })) return;
const isStreamError = (error as ResumeStreamErrorType | undefined)?.isStreamError === true; const isStreamError = (error as ResumeStreamErrorType | undefined)?.isStreamError === true;
resumeFinalStatus = isStreamError resumeFinalStatus = isStreamError
...@@ -1407,8 +1434,13 @@ const ChatBox = ({ ...@@ -1407,8 +1434,13 @@ const ChatBox = ({
}); });
} }
} finally { } finally {
resumeController.current = undefined; if (resumeController.current === controller) {
const finishedInActiveChat = activeChatIdRef.current === resumeForChatId; resumeController.current = undefined;
}
const finishedInActiveChat = isActiveResumeTarget({
appId: resumeForAppId,
chatId: resumeForChatId
});
const leftWhileResuming = const leftWhileResuming =
controller.signal.aborted && isAbortByLeave(controller.signal.reason); controller.signal.aborted && isAbortByLeave(controller.signal.reason);
...@@ -1417,7 +1449,7 @@ const ChatBox = ({ ...@@ -1417,7 +1449,7 @@ const ChatBox = ({
} }
setChatBoxData((state) => setChatBoxData((state) =>
state.chatId === resumeForChatId state.appId === resumeForAppId && state.chatId === resumeForChatId
? { ? {
...state, ...state,
chatGenerateStatus: resumeFinalStatus, chatGenerateStatus: resumeFinalStatus,
...@@ -1428,19 +1460,21 @@ const ChatBox = ({ ...@@ -1428,19 +1460,21 @@ const ChatBox = ({
if (finishedInActiveChat) { if (finishedInActiveChat) {
void postMarkChatRead({ void postMarkChatRead({
appId, appId: resumeForAppId,
chatId: resumeForChatId, chatId: resumeForChatId,
...outLinkAuthData ...outLinkAuthData
}) })
.catch(() => {}) .catch(() => {})
.finally(() => { .finally(() => {
syncSidebarChatGenerateStatus(resumeFinalStatus, { syncSidebarChatGenerateStatus(resumeFinalStatus, {
targetAppId: resumeForAppId,
hasBeenRead: true, hasBeenRead: true,
targetChatId: resumeForChatId targetChatId: resumeForChatId
}); });
}); });
} else { } else {
syncSidebarChatGenerateStatus(resumeFinalStatus, { syncSidebarChatGenerateStatus(resumeFinalStatus, {
targetAppId: resumeForAppId,
hasBeenRead: false, hasBeenRead: false,
targetChatId: resumeForChatId targetChatId: resumeForChatId
}); });
...@@ -1454,9 +1488,12 @@ const ChatBox = ({ ...@@ -1454,9 +1488,12 @@ const ChatBox = ({
appId, appId,
chatId, chatId,
isChatting, isChatting,
chatBoxData.appId,
chatBoxData.chatId,
chatBoxData.chatGenerateStatus, chatBoxData.chatGenerateStatus,
generatingMessage, generatingMessage,
hasMeaningfulAiOutput, hasMeaningfulAiOutput,
isActiveResumeTarget,
getResumeUnavailablePlaceholderText, getResumeUnavailablePlaceholderText,
outLinkAuthData, outLinkAuthData,
resumeTargetAiDataId, resumeTargetAiDataId,
......
projects/app/src/pageComponents/chat/ChatWindow/AppChatWindow.tsx
...@@ -52,6 +52,8 @@ const AppChatWindow = () => { ...@@ -52,6 +52,8 @@ const AppChatWindow = () => {
const chatRecords = useContextSelector(ChatRecordContext, (v) => v.chatRecords); const chatRecords = useContextSelector(ChatRecordContext, (v) => v.chatRecords);
const totalRecordsCount = useContextSelector(ChatRecordContext, (v) => v.totalRecordsCount); const totalRecordsCount = useContextSelector(ChatRecordContext, (v) => v.totalRecordsCount);
const isCurrentChatReady = chatBoxData.appId === appId && chatBoxData.chatId === chatId;
const pane = useContextSelector(ChatPageContext, (v) => v.pane); const pane = useContextSelector(ChatPageContext, (v) => v.pane);
const chatSettings = useContextSelector(ChatPageContext, (v) => v.chatSettings); const chatSettings = useContextSelector(ChatPageContext, (v) => v.chatSettings);
const handlePaneChange = useContextSelector(ChatPageContext, (v) => v.handlePaneChange); const handlePaneChange = useContextSelector(ChatPageContext, (v) => v.handlePaneChange);
...@@ -192,7 +194,7 @@ const AppChatWindow = () => { ...@@ -192,7 +194,7 @@ const AppChatWindow = () => {
<ChatBox <ChatBox
appId={appId} appId={appId}
chatId={chatId} chatId={chatId}
isReady={!loading && !!appId} isReady={!loading && !!appId && isCurrentChatReady}
enableAutoResume enableAutoResume
feedbackType={'user'} feedbackType={'user'}
chatType={ChatTypeEnum.chat} chatType={ChatTypeEnum.chat}
......
projects/app/src/pageComponents/chat/ChatWindow/HomeChatWindow.tsx
...@@ -89,6 +89,8 @@ const HomeChatWindow = () => { ...@@ -89,6 +89,8 @@ const HomeChatWindow = () => {
const chatRecords = useContextSelector(ChatRecordContext, (v) => v.chatRecords); const chatRecords = useContextSelector(ChatRecordContext, (v) => v.chatRecords);
const totalRecordsCount = useContextSelector(ChatRecordContext, (v) => v.totalRecordsCount); const totalRecordsCount = useContextSelector(ChatRecordContext, (v) => v.totalRecordsCount);
const isCurrentChatReady = chatBoxData.appId === appId && chatBoxData.chatId === chatId;
const isQuickApp = useMemo( const isQuickApp = useMemo(
() => chatSettings?.quickAppList.some((app) => app._id === appId), () => chatSettings?.quickAppList.some((app) => app._id === appId),
[chatSettings?.quickAppList, appId] [chatSettings?.quickAppList, appId]
...@@ -463,7 +465,7 @@ const HomeChatWindow = () => { ...@@ -463,7 +465,7 @@ const HomeChatWindow = () => {
<ChatBox <ChatBox
appId={appId} appId={appId}
chatId={chatId} chatId={chatId}
isReady={!loading && !!appId} isReady={!loading && !!appId && isCurrentChatReady}
enableAutoResume enableAutoResume
feedbackType={'user'} feedbackType={'user'}
chatType={ChatTypeEnum.home} chatType={ChatTypeEnum.home}
......
projects/app/src/web/common/api/fetch.ts
...@@ -401,6 +401,11 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum ...@@ -401,6 +401,11 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum
} }
return resolve({ responseText, completedChat, resumeUnavailable }); return resolve({ responseText, completedChat, resumeUnavailable });
}; };
const onAbort = () => {
finished = true;
responseQueue = [];
return onfinish();
};
const onfailed = (err?: any) => { const onfailed = (err?: any) => {
finished = true; finished = true;
const message = getErrText(err, error ?? '响应过程出现异常~'); const message = getErrText(err, error ?? '响应过程出现异常~');
...@@ -424,8 +429,7 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum ...@@ -424,8 +429,7 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum
function animateResponseLoop() { function animateResponseLoop() {
if (signal.aborted) { if (signal.aborted) {
responseQueue.forEach(applyMessageItem); return onAbort();
return onfinish();
} }
if (responseQueue.length > 0) { if (responseQueue.length > 0) {
...@@ -448,6 +452,8 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum ...@@ -448,6 +452,8 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum
animateResponseLoop(); animateResponseLoop();
const enqueue = (data: ResponseQueueItemType) => { const enqueue = (data: ResponseQueueItemType) => {
if (signal.aborted) return;
if (resumePhase === StreamResumePhaseEnum.catchup) { if (resumePhase === StreamResumePhaseEnum.catchup) {
applyMessageItem(data); applyMessageItem(data);
return; return;
...@@ -485,6 +491,8 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum ...@@ -485,6 +491,8 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum
} }
}, },
onmessage: ({ event, data }) => { onmessage: ({ event, data }) => {
if (signal.aborted) return;
if (event === StreamResumePhaseEvent) { if (event === StreamResumePhaseEvent) {
if (data === StreamResumePhaseEnum.catchup || data === StreamResumePhaseEnum.live) { if (data === StreamResumePhaseEnum.catchup || data === StreamResumePhaseEnum.live) {
resumePhase = data; resumePhase = data;
...@@ -546,8 +554,7 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum ...@@ -546,8 +554,7 @@ function $resumefetch({ url, onmessage, onResumeUnavailable, controller }: Resum
clearTimeout(timer); clearTimeout(timer);
if (controller.signal.aborted) { if (controller.signal.aborted) {
finished = true; return onAbort();
return;
} }
onfailed(err); onfailed(err);
...@@ -601,7 +608,10 @@ type StreamResumeFetchParams = { ...@@ -601,7 +608,10 @@ type StreamResumeFetchParams = {
onResumeUnavailable?: (data: ResumeUnavailableType) => void; onResumeUnavailable?: (data: ResumeUnavailableType) => void;
controller: AbortController; controller: AbortController;
}; };
export function streamResumeFetch(params: StreamResumeFetchParams) {
let activeResumeController: AbortController | undefined;
export async function streamResumeFetch(params: StreamResumeFetchParams) {
const { appId, chatId, outLinkAuthData, onmessage, onResumeUnavailable, controller } = params; const { appId, chatId, outLinkAuthData, onmessage, onResumeUnavailable, controller } = params;
const query = new URLSearchParams({ appId, chatId }); const query = new URLSearchParams({ appId, chatId });
...@@ -612,7 +622,16 @@ export function streamResumeFetch(params: StreamResumeFetchParams) { ...@@ -612,7 +622,16 @@ export function streamResumeFetch(params: StreamResumeFetchParams) {
const url = `/api/core/chat/resume?${query}`; const url = `/api/core/chat/resume?${query}`;
return $resumefetch({ url, onmessage, onResumeUnavailable, controller }); if (activeResumeController && activeResumeController !== controller) {
activeResumeController.abort('replace');
}
activeResumeController = controller;
return $resumefetch({ url, onmessage, onResumeUnavailable, controller }).finally(() => {
if (activeResumeController === controller) {
activeResumeController = undefined;
}
});
} }
export const onOptimizePrompt = async ({ export const onOptimizePrompt = async ({
......
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or sign in to comment