Merge 7c2700d099 into b6f45d4b24

admin/inertia/components/chat/index.tsx

@@ -13,6 +13,7 @@ import { IconMenu2, IconX } from '@tabler/icons-react'
 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
@@ -41,6 +42,19 @@ export default function Chat({
   const [isMobileSidebarOpen, setIsMobileSidebarOpen] = useState(false)
   const streamAbortRef = useRef<AbortController | null>(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) => {
@@ -168,6 +182,8 @@ export default function Chat({
   const deleteAllSessionsMutation = useMutation({
     mutationFn: () => api.deleteAllChatSessions(),
     onSuccess: () => {
+      abortCurrentStream()
+      setIsStreamingResponse(false)
       queryClient.invalidateQueries({ queryKey: ['chatSessions'] })
       setActiveSessionId(null)
       setMessages([])
@@ -276,6 +292,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
@@ -283,7 +301,7 @@ export default function Chat({
     // which already calls api.createChatSession with `selectedModel`.
     setActiveSessionId(null)
     setMessages([])
-  }, [pendingModelSwitch])
+  }, [pendingModelSwitch, abortCurrentStream])
 
   const handleCancelModelSwitch = useCallback(() => {
     setPendingModelSwitch(null)
@@ -291,9 +309,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(
@@ -319,6 +339,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
@@ -352,7 +374,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(
@@ -386,6 +408,8 @@ export default function Chat({
         { role: 'user' as const, content },
       ]
 
+      abortCurrentStream()
+
       if (streamingEnabled !== false) {
         // Streaming path
         const abortController = new AbortController()
@@ -479,8 +503,9 @@ export default function Chat({
             })
           }
         } finally {
-          setIsStreamingResponse(false)
-          streamAbortRef.current = null
+          if (clearStreamIfCurrent(streamAbortRef, abortController)) {
+            setIsStreamingResponse(false)
+          }
         }
 
         if (fullContent && sessionId) {
@@ -504,7 +529,7 @@ export default function Chat({
         })
       }
     },
-    [activeSessionId, messages, selectedModel, collectionFilter, chatMutation, queryClient, streamingEnabled, effectiveThinking]
+    [activeSessionId, messages, selectedModel, collectionFilter, chatMutation, queryClient, streamingEnabled, effectiveThinking, abortCurrentStream]
   )
 
   return (

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
+}

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)
+})