fix(stage-ui): restore streaming output; wrap streamText so finish/error truly gate completion (#714)

apps/stage-tamagotchi/src/renderer/components/ChatHistory.vue

@@ -3,7 +3,7 @@ import { MarkdownRenderer } from '@proj-airi/stage-ui/components'
 import { useChatStore } from '@proj-airi/stage-ui/stores/chat'
 import { useBroadcastChannel } from '@vueuse/core'
 import { storeToRefs } from 'pinia'
-import { nextTick, ref, watch } from 'vue'
+import { ref, watch } from 'vue'
 import { useI18n } from 'vue-i18n'
 
 const chatHistoryRef = ref<HTMLDivElement>()
@@ -31,25 +31,30 @@ watch(presentEvent, (ev) => {
   }
 })
 
+function scrollToBottom() {
+  requestAnimationFrame(() => {
+    requestAnimationFrame(() => {
+      if (!chatHistoryRef.value)
+        return
+
+      chatHistoryRef.value.scrollTop = chatHistoryRef.value.scrollHeight
+    })
+  })
+}
+
 onBeforeMessageComposed(async () => {
   // Scroll down to the new sent message
-  nextTick().then(() => {
-    if (!chatHistoryRef.value)
-      return
-
-    chatHistoryRef.value.scrollTop = chatHistoryRef.value.scrollHeight
-  })
+  await scrollToBottom()
 })
 
 onTokenLiteral(async () => {
   // Scroll down to the new responding message
-  nextTick().then(() => {
-    if (!chatHistoryRef.value)
-      return
-
-    chatHistoryRef.value.scrollTop = chatHistoryRef.value.scrollHeight
-  })
+  await scrollToBottom()
 })
+
+watch(sending, () => {
+  scrollToBottom()
+}, { flush: 'post' })
 </script>
 
 <template>

apps/stage-web/src/components/Widgets/ChatHistory.vue

@@ -2,7 +2,7 @@
 import { MarkdownRenderer } from '@proj-airi/stage-ui/components'
 import { useChatStore } from '@proj-airi/stage-ui/stores/chat'
 import { storeToRefs } from 'pinia'
-import { nextTick, ref } from 'vue'
+import { ref, watch } from 'vue'
 import { useI18n } from 'vue-i18n'
 
 const chatHistoryRef = ref<HTMLDivElement>()
@@ -12,25 +12,30 @@ const { messages, sending, streamingMessage } = storeToRefs(useChatStore())
 
 const { onBeforeMessageComposed, onTokenLiteral } = useChatStore()
 
+function scrollToBottom() {
+  requestAnimationFrame(() => {
+    requestAnimationFrame(() => {
+      if (!chatHistoryRef.value)
+        return
+
+      chatHistoryRef.value.scrollTop = chatHistoryRef.value.scrollHeight
+    })
+  })
+}
+
 onBeforeMessageComposed(async () => {
   // Scroll down to the new sent message
-  nextTick().then(() => {
-    if (!chatHistoryRef.value)
-      return
-
-    chatHistoryRef.value.scrollTop = chatHistoryRef.value.scrollHeight
-  })
+  await scrollToBottom()
 })
 
 onTokenLiteral(async () => {
   // Scroll down to the new responding message
-  nextTick().then(() => {
-    if (!chatHistoryRef.value)
-      return
-
-    chatHistoryRef.value.scrollTop = chatHistoryRef.value.scrollHeight
-  })
+  await scrollToBottom()
 })
+
+watch(sending, () => {
+  scrollToBottom()
+}, { flush: 'post' })
 </script>
 
 <template>

packages/stage-ui/src/stores/chat.ts

@@ -114,12 +114,11 @@ export const useChatStore = defineStore('chat', () => {
       attachments?: { type: 'image', data: string, mimeType: string }[]
     },
   ) {
+    if (!sendingMessage && !options.attachments?.length)
+      return
+    sending.value = true
+
     try {
-      sending.value = true
-
-      if (!sendingMessage && !options.attachments?.length)
-        return
-
       for (const hook of onBeforeMessageComposedHooks.value) {
         await hook(sendingMessage)
       }
@@ -209,53 +208,58 @@ export const useChatStore = defineStore('chat', () => {
 
       await stream(options.model, options.chatProvider, newMessages as Message[], {
         headers,
-        async onStreamEvent(event: StreamEvent) {
-          if (event.type === 'tool-call') {
-            toolCallQueue.enqueue({
-              type: 'tool-call',
-              toolCall: event,
-            })
-          }
-          else if (event.type === 'tool-result') {
-            toolCallQueue.enqueue({
-              type: 'tool-call-result',
-              id: event.toolCallId,
-              result: event.result,
-            })
-          }
-          else if (event.type === 'text-delta') {
-            fullText += event.text
-            await parser.consume(event.text)
-          }
-          else if (event.type === 'finish') {
-            // Finalize the parsing of the actual message content
-            await parser.end()
-
-            // Add the completed message to the history only if it has content
-            if (streamingMessage.value.slices.length > 0)
-              messages.value.push(toRaw(streamingMessage.value))
-
-            // Reset the streaming message for the next turn
-            streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [] }
-
-            // Instruct the TTS pipeline to flush by calling hooks directly
-            const flushSignal = `${TTS_FLUSH_INSTRUCTION}${TTS_FLUSH_INSTRUCTION}`
-            for (const hook of onTokenLiteralHooks.value)
-              await hook(flushSignal)
-
-            // Call the end-of-stream hooks
-            for (const hook of onStreamEndHooks.value)
-              await hook()
-
-            // Call the end-of-response hooks with the full text
-            for (const hook of onAssistantResponseEndHooks.value)
-              await hook(fullText)
-
-            // eslint-disable-next-line no-console
-            console.debug('LLM output:', fullText)
+        onStreamEvent: async (event: StreamEvent) => {
+          switch (event.type) {
+            case 'tool-call':
+              toolCallQueue.enqueue({
+                type: 'tool-call',
+                toolCall: event,
+              })
+              break
+            case 'tool-result':
+              toolCallQueue.enqueue({
+                type: 'tool-call-result',
+                id: event.toolCallId,
+                result: event.result,
+              })
+              break
+            case 'text-delta':
+              fullText += event.text
+              await parser.consume(event.text)
+              break
+            case 'finish':
+            // Do nothing, resolve
+              break
+            case 'error':
+              throw event.error ?? new Error('Stream error')
           }
         },
       })
+      // Finalize the parsing of the actual message content
+      await parser.end()
+
+      // Add the completed message to the history only if it has content
+      if (streamingMessage.value.slices.length > 0)
+        messages.value.push(toRaw(streamingMessage.value))
+
+      // Reset the streaming message for the next turn
+      streamingMessage.value = { role: 'assistant', content: '', slices: [], tool_results: [] }
+
+      // Instruct the TTS pipeline to flush by calling hooks directly
+      const flushSignal = `${TTS_FLUSH_INSTRUCTION}${TTS_FLUSH_INSTRUCTION}`
+      for (const hook of onTokenLiteralHooks.value)
+        await hook(flushSignal)
+
+      // Call the end-of-stream hooks
+      for (const hook of onStreamEndHooks.value)
+        await hook()
+
+      // Call the end-of-response hooks with the full text
+      for (const hook of onAssistantResponseEndHooks.value)
+        await hook(fullText)
+
+      // eslint-disable-next-line no-console
+      console.debug('LLM output:', fullText)
 
       for (const hook of onAfterSendHooks.value) {
         await hook(sendingMessage)

packages/stage-ui/src/stores/llm.ts

@@ -1,7 +1,6 @@
 import type { ChatProvider } from '@xsai-ext/shared-providers'
 import type { CommonContentPart, CompletionToolCall, Message } from '@xsai/shared-chat'
 
-import { readableStreamToAsyncIterator } from '@moeru/std'
 import { listModels } from '@xsai/model'
 import { XSAIError } from '@xsai/shared'
 import { streamText } from '@xsai/stream-text'
@@ -24,6 +23,19 @@ export interface StreamOptions {
   supportsTools?: boolean
 }
 
+// TODO: proper format for other error messages.
+function sanitizeMessages(messages: unknown[]): Message[] {
+  return messages.map((m: any) => {
+    if (m && m.role === 'error') {
+      return {
+        role: 'user',
+        content: `User encountered error: ${String(m.content ?? '')}`,
+      } as Message
+    }
+    return m as Message
+  })
+}
+
 function streamOptionsToolsCompatibilityOk(model: string, chatProvider: ChatProvider, _: Message[], options?: StreamOptions, toolsCompatibility: Map<string, boolean> = new Map()): boolean {
   return !!(options?.supportsTools || toolsCompatibility.get(`${chatProvider.chat(model).baseURL}-${model}`))
 }
@@ -31,33 +43,46 @@ function streamOptionsToolsCompatibilityOk(model: string, chatProvider: ChatProv
 async function streamFrom(model: string, chatProvider: ChatProvider, messages: Message[], options?: StreamOptions) {
   const headers = options?.headers
 
-  return await streamText({
-    ...chatProvider.chat(model),
-    maxSteps: 10,
-    // TODO: proper format for other error messages.
-    messages: messages.map(msg => ({ ...msg, content: (msg.role as string === 'error' ? `User encountered error: ${msg.content}` : msg.content), role: (msg.role as string === 'error' ? 'user' : msg.role) } as Message)),
-    headers,
-    // TODO: we need Automatic tools discovery
-    tools: streamOptionsToolsCompatibilityOk(model, chatProvider, messages, options)
-      ? [
-          ...await mcp(),
-          ...await debug(),
-        ]
-      : undefined,
-    onEvent(event) {
-      options?.onStreamEvent?.(event as StreamEvent)
-    },
+  const sanitized = sanitizeMessages(messages as unknown[])
+
+  return new Promise<void>(async (resolve, reject) => {
+    try {
+      await streamText({
+        ...chatProvider.chat(model),
+        maxSteps: 10,
+        messages: sanitized,
+        headers,
+        // TODO: we need Automatic tools discovery
+        tools: streamOptionsToolsCompatibilityOk(model, chatProvider, messages, options)
+          ? [
+              ...await mcp(),
+              ...await debug(),
+            ]
+          : undefined,
+        async onEvent(event) {
+          try {
+            await options?.onStreamEvent?.(event as StreamEvent)
+            if (event.type === 'finish')
+              resolve()
+            else if (event.type === 'error')
+              reject(event.error ?? new Error('Stream error'))
+          }
+          catch (err) {
+            reject(err)
+          }
+        },
+      })
+    }
+    catch (err) {
+      reject(err)
+    }
   })
 }
 
 export async function attemptForToolsCompatibilityDiscovery(model: string, chatProvider: ChatProvider, _: Message[], options?: Omit<StreamOptions, 'supportsTools'>): Promise<boolean> {
   async function attempt(enable: boolean) {
     try {
-      const res = await streamFrom(model, chatProvider, [{ role: 'user', content: 'Hello, world!' }], { ...options, supportsTools: enable })
-      for await (const _ of readableStreamToAsyncIterator(res.textStream)) {
-        // Drop
-      }
-
+      await streamFrom(model, chatProvider, [{ role: 'user', content: 'Hello, world!' }], { ...options, supportsTools: enable })
       return true
     }
     catch (err) {