From 7c2700d09908e53ff8860c0403442da6ab51e970 Mon Sep 17 00:00:00 2001 From: NgoQuocViet2001 Date: Sun, 9 Aug 2026 00:49:31 +0700 Subject: [PATCH] fix(chat): abort abandoned response streams --- admin/inertia/components/chat/index.tsx | 37 ++++++++++++++--- admin/inertia/components/chat/stream_abort.ts | 21 ++++++++++ admin/tests/unit/chat_stream_abort.spec.ts | 41 +++++++++++++++++++ 3 files changed, 93 insertions(+), 6 deletions(-) create mode 100644 admin/inertia/components/chat/stream_abort.ts create mode 100644 admin/tests/unit/chat_stream_abort.spec.ts diff --git a/admin/inertia/components/chat/index.tsx b/admin/inertia/components/chat/index.tsx index 8abf607..635ba7b 100644 --- a/admin/inertia/components/chat/index.tsx +++ b/admin/inertia/components/chat/index.tsx @@ -14,6 +14,7 @@ import { DEFAULT_QUERY_REWRITE_MODEL } from '../../../constants/ollama' import { useSystemSetting } from '~/hooks/useSystemSetting' import Switch from '~/components/inputs/Switch' import InfoTooltip from '~/components/InfoTooltip' +import { abortActiveStream, clearStreamIfCurrent } from './stream_abort' interface ChatProps { enabled: boolean @@ -42,6 +43,19 @@ export default function Chat({ const [isMobileSidebarOpen, setIsMobileSidebarOpen] = useState(false) const streamAbortRef = useRef(null) + const abortCurrentStream = useCallback(() => { + abortActiveStream(streamAbortRef) + }, []) + + useEffect(() => () => abortCurrentStream(), [abortCurrentStream]) + + useEffect(() => { + if (!enabled) { + abortCurrentStream() + setIsStreamingResponse(false) + } + }, [enabled, abortCurrentStream]) + useEffect(() => { if (!isMobileSidebarOpen) return const closeOnEscape = (event: KeyboardEvent) => { @@ -146,6 +160,8 @@ export default function Chat({ const deleteAllSessionsMutation = useMutation({ mutationFn: () => api.deleteAllChatSessions(), onSuccess: () => { + abortCurrentStream() + setIsStreamingResponse(false) queryClient.invalidateQueries({ queryKey: ['chatSessions'] }) setActiveSessionId(null) setMessages([]) @@ -254,6 +270,8 @@ export default function Chat({ api.unloadChatModels(newModel).catch((err) => { console.warn('Failed to unload previous chat model:', err) }) + abortCurrentStream() + setIsStreamingResponse(false) setSelectedModel(newModel) setPendingModelSwitch(null) // Clear the active session and messages — the next user message will @@ -261,7 +279,7 @@ export default function Chat({ // which already calls api.createChatSession with `selectedModel`. setActiveSessionId(null) setMessages([]) - }, [pendingModelSwitch]) + }, [pendingModelSwitch, abortCurrentStream]) const handleCancelModelSwitch = useCallback(() => { setPendingModelSwitch(null) @@ -269,9 +287,11 @@ export default function Chat({ const handleNewChat = useCallback(() => { // Just clear the active session and messages - don't create a session yet + abortCurrentStream() + setIsStreamingResponse(false) setActiveSessionId(null) setMessages([]) - }, []) + }, [abortCurrentStream]) const handleClearHistory = useCallback(() => { openModal( @@ -297,6 +317,8 @@ export default function Chat({ async (sessionId: string) => { // Cancel any ongoing suggestions fetch queryClient.cancelQueries({ queryKey: ['chatSuggestions'] }) + abortCurrentStream() + setIsStreamingResponse(false) setActiveSessionId(sessionId) // Load messages for this session @@ -330,7 +352,7 @@ export default function Chat({ console.warn('Failed to unload non-target chat models on session switch:', err) }) }, - [installedModels, queryClient, selectedModel] + [installedModels, queryClient, selectedModel, abortCurrentStream] ) const handleSendMessage = useCallback( @@ -364,6 +386,8 @@ export default function Chat({ { role: 'user' as const, content }, ] + abortCurrentStream() + if (streamingEnabled !== false) { // Streaming path const abortController = new AbortController() @@ -457,8 +481,9 @@ export default function Chat({ }) } } finally { - setIsStreamingResponse(false) - streamAbortRef.current = null + if (clearStreamIfCurrent(streamAbortRef, abortController)) { + setIsStreamingResponse(false) + } } if (fullContent && sessionId) { @@ -482,7 +507,7 @@ export default function Chat({ }) } }, - [activeSessionId, messages, selectedModel, collectionFilter, chatMutation, queryClient, streamingEnabled, effectiveThinking] + [activeSessionId, messages, selectedModel, collectionFilter, chatMutation, queryClient, streamingEnabled, effectiveThinking, abortCurrentStream] ) return ( diff --git a/admin/inertia/components/chat/stream_abort.ts b/admin/inertia/components/chat/stream_abort.ts new file mode 100644 index 0000000..beb4020 --- /dev/null +++ b/admin/inertia/components/chat/stream_abort.ts @@ -0,0 +1,21 @@ +export interface StreamAbortRef { + current: AbortController | null +} + +export function abortActiveStream(ref: StreamAbortRef): boolean { + const controller = ref.current + if (!controller) return false + + // Clear ownership before aborting. The rejected stream settles asynchronously, + // so its finally block must not be allowed to clear a replacement controller. + ref.current = null + controller.abort() + return true +} + +export function clearStreamIfCurrent(ref: StreamAbortRef, controller: AbortController): boolean { + if (ref.current !== controller) return false + + ref.current = null + return true +} diff --git a/admin/tests/unit/chat_stream_abort.spec.ts b/admin/tests/unit/chat_stream_abort.spec.ts new file mode 100644 index 0000000..65e5e90 --- /dev/null +++ b/admin/tests/unit/chat_stream_abort.spec.ts @@ -0,0 +1,41 @@ +import * as assert from 'node:assert/strict' +import { test } from 'node:test' + +import { + abortActiveStream, + clearStreamIfCurrent, + type StreamAbortRef, +} from '../../inertia/components/chat/stream_abort.js' + +test('aborts and releases the active stream', () => { + const controller = new AbortController() + const ref: StreamAbortRef = { current: controller } + + assert.equal(abortActiveStream(ref), true) + assert.equal(controller.signal.aborted, true) + assert.equal(ref.current, null) +}) + +test('does nothing when no stream is active', () => { + const ref: StreamAbortRef = { current: null } + + assert.equal(abortActiveStream(ref), false) + assert.equal(ref.current, null) +}) + +test('clears the controller owned by the settling stream', () => { + const controller = new AbortController() + const ref: StreamAbortRef = { current: controller } + + assert.equal(clearStreamIfCurrent(ref, controller), true) + assert.equal(ref.current, null) +}) + +test('does not clear a replacement stream controller', () => { + const staleController = new AbortController() + const replacementController = new AbortController() + const ref: StreamAbortRef = { current: replacementController } + + assert.equal(clearStreamIfCurrent(ref, staleController), false) + assert.equal(ref.current, replacementController) +})