fix(stage-ui): preserve streaming transcript corrections (#2261)

packages/stage-layouts/src/composables/use-transcriptions.test.ts

@@ -18,10 +18,16 @@ function createMockStore() {
 }
 
 const mockTranscribedContent = 'test content'
+interface MockStreamingCallbacks {
+  onSentenceEnd: (delta: string) => void
+  onSpeechEnd?: (text: string) => void
+  onTranscriptionUpdate?: (text: string) => void
+}
+
 function createMockPipeline() {
   return {
     removeStreamingTranscriptionConsumer: vi.fn(),
-    transcribeForMediaStream: vi.fn().mockImplementation((_stream, options: { onSentenceEnd: (delta: string) => void }) => {
+    transcribeForMediaStream: vi.fn().mockImplementation((_stream, options: MockStreamingCallbacks) => {
       options.onSentenceEnd(mockTranscribedContent)
     }),
     stopStreamingTranscription: vi.fn().mockResolvedValue(undefined),
@@ -264,6 +270,55 @@ describe('useTranscriptions', () => {
       expect(mockInput.value).toBe(`${prependText} ${mockTranscribedContent}`)
     })
 
+    it('replaces volatile snapshots when the provider corrects text', async () => {
+      // ROOT CAUSE:
+      //
+      // The input consumer only accepted final deltas. It had no operation for
+      // replacing a provider-owned draft when the provider corrected that text.
+      const mockInput = ref('prefix')
+      const observedInputs: string[] = []
+      mockHearingStore.configured.value = true
+      mockAudioDevice.stream.value = { id: 'stream-1' } as any
+      mockAudioDevice.enabled.value = true
+      mockHearingPipeline.supportsStreamInput.value = true
+      mockHearingPipeline.transcribeForMediaStream.mockImplementation((_stream, options: MockStreamingCallbacks) => {
+        options.onTranscriptionUpdate?.('今天天气很号')
+        observedInputs.push(mockInput.value)
+        options.onTranscriptionUpdate?.('今天天气很好')
+        observedInputs.push(mockInput.value)
+        options.onSentenceEnd('今天天气很好')
+      })
+
+      const { startStreamingTranscription }
+        = useTranscriptions({ ...createOptions(), messageInputRef: mockInput })
+
+      await startStreamingTranscription()
+
+      expect(observedInputs).toEqual(['prefix 今天天气很号', 'prefix 今天天气很好'])
+      expect(mockInput.value).toBe('prefix 今天天气很好')
+    })
+
+    it('preserves manual input changes during a volatile transcription', async () => {
+      const mockInput = ref('prefix')
+      mockHearingStore.configured.value = true
+      mockAudioDevice.stream.value = { id: 'stream-1' } as any
+      mockAudioDevice.enabled.value = true
+      mockHearingPipeline.supportsStreamInput.value = true
+      mockHearingPipeline.transcribeForMediaStream.mockImplementation((_stream, options: MockStreamingCallbacks) => {
+        options.onTranscriptionUpdate?.('provider draft')
+        mockInput.value = 'manual edit'
+        options.onTranscriptionUpdate?.('provider correction')
+        options.onSentenceEnd('provider final')
+      })
+
+      const { startStreamingTranscription }
+        = useTranscriptions({ ...createOptions(), messageInputRef: mockInput })
+
+      await startStreamingTranscription()
+
+      expect(mockInput.value).toBe('manual edit')
+    })
+
     it('should trigger auto-send after delay', async () => {
       const mockInput = ref('')
       const mockSendMessage = vi.fn()

packages/stage-layouts/src/composables/use-transcriptions.ts

@@ -1,5 +1,6 @@
 import type { MaybeRefOrGetter, Ref } from 'vue'
 
+import { useStreamingTranscriptionInput } from '@proj-airi/stage-ui/composables/use-streaming-transcription-input'
 import { useHearingSpeechInputPipeline, useHearingStore } from '@proj-airi/stage-ui/stores/modules/hearing'
 import { useProviderStore } from '@proj-airi/stage-ui/stores/providers/provider'
 import { useSettingsAudioDevice } from '@proj-airi/stage-ui/stores/settings'
@@ -28,6 +29,7 @@ export function useTranscriptions(options: TranscriptionOptions) {
 
   const isListening = ref(false)
   const transcriptionConsumerId = `interactive-area:${useId()}`
+  const streamingInput = useStreamingTranscriptionInput(messageInput)
 
   // Auto-send logic
   let autoSendTimeout: ReturnType<typeof setTimeout> | undefined
@@ -60,6 +62,7 @@ export function useTranscriptions(options: TranscriptionOptions) {
 
   const stopStreaming = async () => {
     removeStreamingTranscriptionConsumer(transcriptionConsumerId)
+    streamingInput.clear()
 
     if (!isListening.value)
       return
@@ -181,15 +184,13 @@ export function useTranscriptions(options: TranscriptionOptions) {
       await transcribeForMediaStream(stream.value, {
         consumerId: transcriptionConsumerId,
         onSentenceEnd: (delta) => {
-          if (delta && delta.trim()) {
-            console.info('Received transcription delta:', delta, { source: 'useTranscriptions' })
-            // Append transcribed text to message input
-            const currentText = messageInput.value.trim()
-            messageInput.value = currentText ? `${currentText} ${delta}` : delta
+          if (streamingInput.commit(delta)) {
+            console.info('Received final transcription:', delta, { source: 'useTranscriptions' })
             debouncedAutoSend()
           }
         },
-        // Omit onSpeechEnd to avoid re-adding user-deleted text; use sentence deltas only.
+        onSpeechEnd: streamingInput.clear,
+        onTranscriptionUpdate: streamingInput.replace,
       })
 
       // Only set listening to true if transcription started successfully
@@ -198,6 +199,7 @@ export function useTranscriptions(options: TranscriptionOptions) {
       console.info('Streaming transcription initiated successfully', { source: 'useTranscriptions' })
     }
     catch (err) {
+      streamingInput.clear()
       console.error('Transcription error:', err, { source: 'useTranscriptions' })
       isListening.value = false
       throw err

packages/stage-ui/src/composables/use-streaming-transcription-input.test.ts

@@ -0,0 +1,56 @@
+import { describe, expect, it } from 'vitest'
+import { shallowRef } from 'vue'
+
+import { useStreamingTranscriptionInput } from './use-streaming-transcription-input'
+
+describe('streaming transcription input', () => {
+  it('replaces provider corrections and clears only the owned suffix', () => {
+    const input = shallowRef('manual prefix')
+    const transcription = useStreamingTranscriptionInput(input)
+
+    transcription.replace('今天天气很号')
+    transcription.replace('今天天气很好')
+
+    expect(input.value).toBe('manual prefix 今天天气很好')
+
+    transcription.clear()
+
+    expect(input.value).toBe('manual prefix')
+  })
+
+  it('keeps manual edits after the user changes the volatile suffix', () => {
+    const input = shallowRef('manual prefix')
+    const transcription = useStreamingTranscriptionInput(input)
+
+    transcription.replace('provider draft')
+    input.value = 'user replacement'
+    transcription.replace('provider correction')
+    transcription.clear()
+
+    expect(input.value).toBe('user replacement')
+  })
+
+  it('commits the final correction as stable input', () => {
+    const input = shallowRef('manual prefix')
+    const transcription = useStreamingTranscriptionInput(input)
+
+    transcription.replace('provider draft')
+
+    expect(transcription.commit('provider final')).toBe(true)
+    expect(input.value).toBe('manual prefix provider final')
+
+    transcription.clear()
+
+    expect(input.value).toBe('manual prefix provider final')
+  })
+
+  it('removes the provider draft when the final transcript is empty', () => {
+    const input = shallowRef('manual prefix')
+    const transcription = useStreamingTranscriptionInput(input)
+
+    transcription.replace('provider draft')
+
+    expect(transcription.commit('')).toBe(false)
+    expect(input.value).toBe('manual prefix')
+  })
+})

packages/stage-ui/src/composables/use-streaming-transcription-input.ts

@@ -0,0 +1,88 @@
+import type { Ref } from 'vue'
+
+function joinInputAndTranscription(input: string, transcription: string) {
+  return [input.trimEnd(), transcription.trim()].filter(Boolean).join(' ')
+}
+
+/**
+ * Applies replaceable streaming transcription text to an editable input.
+ * Manual input changes detach the current provider-owned suffix.
+ */
+export function useStreamingTranscriptionInput(input: Ref<string>) {
+  let volatileTranscription = ''
+  let volatileTranscriptionDetached = false
+
+  function reset() {
+    volatileTranscription = ''
+    volatileTranscriptionDetached = false
+  }
+
+  function clear() {
+    if (volatileTranscription && input.value.endsWith(volatileTranscription))
+      input.value = input.value.slice(0, -volatileTranscription.length).trimEnd()
+
+    reset()
+  }
+
+  function replace(text: string) {
+    if (volatileTranscriptionDetached)
+      return
+
+    const nextTranscription = text.trim()
+    if (!nextTranscription) {
+      clear()
+      return
+    }
+
+    if (!volatileTranscription) {
+      volatileTranscription = nextTranscription
+      input.value = joinInputAndTranscription(input.value, nextTranscription)
+      return
+    }
+
+    if (!input.value.endsWith(volatileTranscription)) {
+      volatileTranscriptionDetached = true
+      volatileTranscription = ''
+      return
+    }
+
+    const stableInput = input.value.slice(0, -volatileTranscription.length)
+    volatileTranscription = nextTranscription
+    input.value = joinInputAndTranscription(stableInput, nextTranscription)
+  }
+
+  function commit(text: string) {
+    if (volatileTranscriptionDetached) {
+      reset()
+      return false
+    }
+
+    const finalTranscription = text.trim()
+    if (!finalTranscription) {
+      clear()
+      return false
+    }
+
+    if (!volatileTranscription) {
+      input.value = joinInputAndTranscription(input.value, finalTranscription)
+      return true
+    }
+
+    if (!input.value.endsWith(volatileTranscription)) {
+      reset()
+      return false
+    }
+
+    const stableInput = input.value.slice(0, -volatileTranscription.length)
+    input.value = joinInputAndTranscription(stableInput, finalTranscription)
+    reset()
+    return true
+  }
+
+  return {
+    clear,
+    commit,
+    replace,
+    reset,
+  }
+}

packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.test.ts

@@ -0,0 +1,135 @@
+import { afterEach, describe, expect, it, vi } from 'vitest'
+
+import { createAliyunNLSProvider } from './provider'
+
+class FakeWebSocket extends EventTarget {
+  static readonly CONNECTING = 0
+  static readonly OPEN = 1
+  static readonly CLOSING = 2
+  static readonly CLOSED = 3
+  static readonly instances: FakeWebSocket[] = []
+
+  binaryType: BinaryType = 'blob'
+  onclose: ((event: { code: number, reason: string }) => void) | null = null
+  onerror: ((event: Event) => void) | null = null
+  onmessage: ((event: MessageEvent<string>) => void) | null = null
+  onopen: ((event: Event) => void) | null = null
+  readyState = FakeWebSocket.CONNECTING
+  sent: unknown[] = []
+
+  constructor(readonly url: string | URL) {
+    super()
+    FakeWebSocket.instances.push(this)
+  }
+
+  close(code = 1000, reason = '') {
+    this.readyState = FakeWebSocket.CLOSED
+    this.onclose?.({ code, reason })
+  }
+
+  open() {
+    this.readyState = FakeWebSocket.OPEN
+    this.onopen?.(new Event('open'))
+  }
+
+  receive(payload: unknown) {
+    this.onmessage?.(new MessageEvent('message', { data: JSON.stringify(payload) }))
+  }
+
+  send(data: unknown) {
+    this.sent.push(data)
+  }
+}
+
+afterEach(() => {
+  FakeWebSocket.instances.length = 0
+  vi.unstubAllGlobals()
+})
+
+describe('aliyun NLS provider', () => {
+  it('forwards interim corrections and the final sentence as transcript snapshots', async () => {
+    // ROOT CAUSE:
+    //
+    // Aliyun emits each volatile hypothesis as `TranscriptionResultChanged`.
+    // The provider ignored these events and emitted only the final sentence.
+    vi.stubGlobal('WebSocket', FakeWebSocket)
+    vi.stubGlobal('fetch', vi.fn(async () => new Response(
+      JSON.stringify({
+        NlsRequestId: 'nls-request',
+        RequestId: 'request',
+        ErrMsg: '',
+        Token: {
+          ExpireTime: Math.floor(Date.now() / 1000) + 3600,
+          Id: 'token',
+          UserId: 'user',
+        },
+      }),
+      { headers: { 'Content-Type': 'application/json' } },
+    )))
+
+    const inputAudioStream = new ReadableStream<ArrayBuffer>()
+    const provider = createAliyunNLSProvider('access-key-id', 'access-key-secret', 'app-key')
+    const speech = provider.speech('aliyun-nls-v1', { inputAudioStream })
+    if (!speech.fetch)
+      throw new Error('The Aliyun provider did not create its streaming transport.')
+
+    const response = await speech.fetch(new URL('https://example.invalid/transcription'), {})
+    const responseText = response.text()
+
+    await vi.waitFor(() => expect(FakeWebSocket.instances).toHaveLength(1))
+    const socket = FakeWebSocket.instances[0]
+    if (!socket)
+      throw new Error('The Aliyun provider did not open its streaming socket.')
+    socket.open()
+    await vi.waitFor(() => expect(socket.sent).toHaveLength(1))
+    const startEvent = JSON.parse(socket.sent[0] as string)
+    expect(startEvent.payload.enable_intermediate_result).toBe(true)
+
+    socket.receive(serverEvent('SentenceBegin', { index: 1, time: 200 }))
+    socket.receive(serverEvent('TranscriptionResultChanged', {
+      index: 1,
+      result: '今天天气很号',
+      status: 20000000,
+      time: 1000,
+    }))
+    socket.receive(serverEvent('TranscriptionResultChanged', {
+      index: 1,
+      result: '今天天气很好',
+      status: 20000000,
+      time: 1200,
+    }))
+    socket.receive(serverEvent('SentenceEnd', {
+      begin_time: 200,
+      confidence: 0.95,
+      index: 1,
+      result: '今天天气很好',
+      stash_result: {
+        beginTime: 0,
+        currentTime: 1200,
+        sentenceId: 2,
+        text: '',
+      },
+      status: 20000000,
+      time: 1200,
+    }))
+    socket.receive(serverEvent('TranscriptionCompleted', undefined))
+
+    await expect(responseText).resolves.toContain('"type":"transcript.text.snapshot","text":"今天天气很号","isFinal":false')
+    await expect(responseText).resolves.toContain('"type":"transcript.text.snapshot","text":"今天天气很好","isFinal":true')
+  })
+})
+
+function serverEvent(name: string, payload: unknown) {
+  return {
+    header: {
+      appkey: 'app-key',
+      message_id: 'message-id',
+      name,
+      namespace: 'SpeechTranscriber',
+      status: 20000000,
+      status_message: 'SUCCESS',
+      task_id: 'task-id',
+    },
+    payload,
+  }
+}

packages/stage-ui/src/libs/providers/providers/aliyun-nls/provider.ts

@@ -1,5 +1,6 @@
 import type { SpeechProviderWithExtraOptions } from '@xsai-ext/providers/utils'
 
+import type { StreamTranscriptionSnapshot } from '../../stream-transcription'
 import type { EventStartTranscription, ServerEvent, ServerEvents } from './session'
 
 import { tryCatch } from '@moeru/std'
@@ -58,6 +59,7 @@ function createWaiter(timeoutMs: number, abortSignal?: AbortSignal) {
 }
 
 const DEFAULT_SESSION_OPTIONS: EventStartTranscription['payload'] = {
+  enable_intermediate_result: true,
   format: 'pcm',
   sample_rate: 16000,
 }
@@ -104,12 +106,16 @@ function toArrayBuffer(chunk: AudioChunk): ArrayBuffer {
 
 const sseEncoder = new TextEncoder()
 
-function encodeSSE(payload: { delta: string, type: 'transcript.text.delta' | 'transcript.text.done' }): Uint8Array {
+type AliyunTranscriptionSSEEvent
+  = | StreamTranscriptionSnapshot
+    | { delta: string, type: 'transcript.text.done' }
+
+function encodeSSE(payload: AliyunTranscriptionSSEEvent): Uint8Array {
   return sseEncoder.encode(`data: ${JSON.stringify(payload)}\n\n`)
 }
 
 interface InternalRealtimeOptions extends CreateAliyunStreamTranscriptionOptions {
-  onSentenceFinal?: (payload: ServerEvents['SentenceEnd']) => Promise<void> | void
+  onTranscriptSnapshot?: (snapshot: StreamTranscriptionSnapshot) => Promise<void> | void
   idleTimeoutMs?: number
   stopAckTimeoutMs?: number
 }
@@ -125,7 +131,7 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
     abortSignal,
     hooks,
     onSessionTerminated,
-    onSentenceFinal,
+    onTranscriptSnapshot,
     idleTimeoutMs = 8000,
     stopAckTimeoutMs = 2000,
   } = options
@@ -148,6 +154,30 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
   const stopWaiter = createWaiter(stopAckTimeoutMs, abortSignal)
   let stopping = false
   let cleanupPromise: Promise<void> | undefined
+  const sentenceStartMillisecondsByIndex = new Map<number, number>()
+  const sentenceTextByIndex = new Map<number, string>()
+
+  function transcriptSnapshot(
+    payload: Pick<ServerEvents['TranscriptionResultChanged'], 'index' | 'result' | 'time'>,
+    isFinal: boolean,
+  ): StreamTranscriptionSnapshot {
+    sentenceTextByIndex.set(payload.index, payload.result)
+    const orderedText = [...sentenceTextByIndex.entries()]
+      .sort(([leftIndex], [rightIndex]) => leftIndex - rightIndex)
+      .map(([, text]) => text.trim())
+      .filter(Boolean)
+      .join('\n')
+    const startMilliseconds = Math.min(...sentenceStartMillisecondsByIndex.values(), payload.time)
+
+    return {
+      type: 'transcript.text.snapshot',
+      text: orderedText,
+      isFinal,
+      locale: 'und',
+      startMilliseconds,
+      durationMilliseconds: Math.max(0, payload.time - startMilliseconds),
+    }
+  }
 
   async function requestStop(reason?: unknown) {
     if (stopping)
@@ -253,9 +283,22 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
           case 'TranscriptionStarted':
             onTranscriptionStarted()
             break
-          case 'SentenceEnd':
-            await onSentenceFinal?.(event.payload as ServerEvents['SentenceEnd'])
+          case 'SentenceBegin': {
+            const payload = event.payload as ServerEvents['SentenceBegin']
+            sentenceStartMillisecondsByIndex.set(payload.index, payload.time)
             break
+          }
+          case 'TranscriptionResultChanged': {
+            const payload = event.payload as ServerEvents['TranscriptionResultChanged']
+            await onTranscriptSnapshot?.(transcriptSnapshot(payload, false))
+            break
+          }
+          case 'SentenceEnd': {
+            const payload = event.payload as ServerEvents['SentenceEnd']
+            sentenceStartMillisecondsByIndex.set(payload.index, payload.begin_time)
+            await onTranscriptSnapshot?.(transcriptSnapshot(payload, true))
+            break
+          }
           case 'TranscriptionCompleted':
             stopWaiter.trigger()
             await cleanup(undefined, { sendStop: false, closeSocket: false })
@@ -342,12 +385,8 @@ export function createAliyunNLSProvider(
                     controller.error(error instanceof Error ? error : new Error(String(error)))
                   }
                 },
-                onSentenceFinal: async (payload) => {
-                  const text = payload.result ? `${payload.result}\n` : ''
-                  if (text)
-                    controller.enqueue(encodeSSE({ delta: text, type: 'transcript.text.delta' }))
-
-                  controller.enqueue(encodeSSE({ delta: '', type: 'transcript.text.done' }))
+                onTranscriptSnapshot: async (snapshot) => {
+                  controller.enqueue(encodeSSE(snapshot))
                 },
               }).then((handle) => {
                 sessionHandle = handle

packages/stage-ui/src/libs/providers/stream-transcription/index.test.ts

@@ -31,4 +31,36 @@ describe('streamTranscription', () => {
   it('rejects requests without an audio input', () => {
     expect(() => streamTranscription({})).toThrow('Audio stream or file is required')
   })
+
+  it('replaces volatile transcript snapshots instead of appending corrections', async () => {
+    // ROOT CAUSE:
+    //
+    // The adapter only accumulated `transcript.text.delta` events. Providers
+    // that emit complete volatile hypotheses could not replace incorrect text.
+    const encoder = new TextEncoder()
+    const responseBody = new ReadableStream<Uint8Array>({
+      start(controller) {
+        controller.enqueue(encoder.encode('data: {"type":"transcript.text.snapshot","text":"今天天气很号","isFinal":false,"locale":"zh-CN","startMilliseconds":0,"durationMilliseconds":1000}\n\n'))
+        controller.enqueue(encoder.encode('data: {"type":"transcript.text.snapshot","text":"今天天气很好","isFinal":true,"locale":"zh-CN","startMilliseconds":0,"durationMilliseconds":1200}\n\n'))
+        controller.close()
+      },
+    })
+    const audioStream = new ReadableStream<ArrayBuffer>({
+      start(controller) {
+        controller.close()
+      },
+    })
+
+    const result = streamTranscription({
+      baseURL: 'https://example.invalid/transcription',
+      fetch: async () => new Response(responseBody),
+      inputAudioStream: audioStream,
+    })
+    const updates = []
+    for await (const update of result.fullStream)
+      updates.push(update)
+
+    expect(updates).toHaveLength(2)
+    expect(await result.text).toBe('今天天气很好')
+  })
 })

packages/stage-ui/src/libs/providers/stream-transcription/index.ts

@@ -3,6 +3,23 @@ import type { StreamTranscriptionDelta, StreamTranscriptionResult } from '@xsai/
 
 type AudioChunk = ArrayBuffer | ArrayBufferView
 
+/** A complete transcript snapshot that replaces earlier volatile text. */
+export interface StreamTranscriptionSnapshot {
+  durationMilliseconds: number
+  isFinal: boolean
+  locale: string
+  startMilliseconds: number
+  text: string
+  type: 'transcript.text.snapshot'
+}
+
+export type AIRIStreamTranscriptionDelta = StreamTranscriptionDelta | StreamTranscriptionSnapshot
+
+/** xsAI stream result with AIRI's replaceable snapshot event. */
+export interface AIRIStreamTranscriptionResult extends Omit<StreamTranscriptionResult, 'fullStream'> {
+  fullStream: ReadableStream<AIRIStreamTranscriptionDelta>
+}
+
 /** Options for adapting an SSE transcription request to xsAI stream results. */
 export interface StreamTranscriptionOptions {
   abortSignal?: AbortSignal
@@ -33,7 +50,7 @@ function resolveAudioStream(options: StreamTranscriptionOptions): ReadableStream
   return stream as ReadableStream<AudioChunk>
 }
 
-function parseSSELine(line: string): StreamTranscriptionDelta | undefined {
+function parseSSELine(line: string): AIRIStreamTranscriptionDelta | undefined {
   if (!line || !line.startsWith('data:'))
     return undefined
 
@@ -42,14 +59,14 @@ function parseSSELine(line: string): StreamTranscriptionDelta | undefined {
   if (!data)
     return undefined
 
-  return JSON.parse(data) as StreamTranscriptionDelta
+  return JSON.parse(data) as AIRIStreamTranscriptionDelta
 }
 
 function createSSETransformer() {
   const decoder = new TextDecoder()
   let buffer = ''
 
-  return new TransformStream<Uint8Array, StreamTranscriptionDelta>({
+  return new TransformStream<Uint8Array, AIRIStreamTranscriptionDelta>({
     transform: (chunk, controller) => {
       buffer += decoder.decode(chunk, { stream: true })
       const lines = buffer.split('\n')
@@ -78,16 +95,16 @@ function createSSETransformer() {
  * The provider owns transport details. This adapter owns only request input,
  * SSE parsing, and the result streams consumed by Hearing.
  */
-export function streamTranscription(options: StreamTranscriptionOptions): StreamTranscriptionResult {
+export function streamTranscription(options: StreamTranscriptionOptions): AIRIStreamTranscriptionResult {
   const audioStream = resolveAudioStream(options)
   const fetcher = options.fetch ?? globalThis.fetch
   const deferredText = createDeferred<string>()
 
   let text = ''
   let textStreamCtrl: ReadableStreamDefaultController<string> | undefined
-  let fullStreamCtrl: ReadableStreamDefaultController<StreamTranscriptionDelta> | undefined
+  let fullStreamCtrl: ReadableStreamDefaultController<AIRIStreamTranscriptionDelta> | undefined
 
-  const fullStream = new ReadableStream<StreamTranscriptionDelta>({
+  const fullStream = new ReadableStream<AIRIStreamTranscriptionDelta>({
     start(controller) {
       fullStreamCtrl = controller
     },
@@ -119,13 +136,16 @@ export function streamTranscription(options: StreamTranscriptionOptions): Stream
 
       await response.body
         .pipeThrough(createSSETransformer())
-        .pipeTo(new WritableStream<StreamTranscriptionDelta>({
+        .pipeTo(new WritableStream<AIRIStreamTranscriptionDelta>({
           write: (chunk) => {
             fullStreamCtrl?.enqueue(chunk)
             if (chunk.type === 'transcript.text.delta') {
               text += chunk.delta
               textStreamCtrl?.enqueue(chunk.delta)
             }
+            else if (chunk.type === 'transcript.text.snapshot') {
+              text = chunk.text
+            }
           },
           close: () => {
             fullStreamCtrl?.close()

packages/stage-ui/src/stores/modules/hearing.ts

@@ -1,8 +1,9 @@
 import type { Span } from '@opentelemetry/api'
 import type { TranscriptionProviderWithExtraOptions } from '@xsai-ext/providers/utils'
 import type { WithUnknown } from '@xsai/shared'
-import type { StreamTranscriptionResult, StreamTranscriptionOptions as XSAIStreamTranscriptionOptions } from '@xsai/stream-transcription'
+import type { StreamTranscriptionOptions as XSAIStreamTranscriptionOptions } from '@xsai/stream-transcription'
 
+import type { AIRIStreamTranscriptionResult } from '../../libs/providers/stream-transcription'
 import type { StreamingTranscriptionCallbacks, StreamingTranscriptionConsumer } from './streaming-transcription-consumers'
 
 import { errorMessageFrom, tryCatch } from '@moeru/std'
@@ -90,11 +91,11 @@ export interface StreamTranscriptionStreamInputOptions extends Omit<XSAIStreamTr
   inputAudioStream: ReadableStream<ArrayBuffer>
 }
 
-export type StreamTranscription = (options: WithUnknown<StreamTranscriptionFileInputOptions | StreamTranscriptionStreamInputOptions>) => StreamTranscriptionResult
+export type StreamTranscription = (options: WithUnknown<StreamTranscriptionFileInputOptions | StreamTranscriptionStreamInputOptions>) => AIRIStreamTranscriptionResult
 
 type GenerateTranscriptionResponse = Awaited<ReturnType<typeof generateTranscription>>
 type HearingTranscriptionGenerateResult = GenerateTranscriptionResponse & { mode: 'generate' }
-type HearingTranscriptionStreamResult = StreamTranscriptionResult & { mode: 'stream' }
+type HearingTranscriptionStreamResult = AIRIStreamTranscriptionResult & { mode: 'stream' }
 export type HearingTranscriptionResult = HearingTranscriptionGenerateResult | HearingTranscriptionStreamResult
 
 type HearingTranscriptionInput = File | {
@@ -573,6 +574,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
   const streamingCallbacks = {
     onSentenceEnd: (delta: string) => streamingConsumers.emitSentenceEnd(delta),
     onSpeechEnd: (text: string) => streamingConsumers.emitSpeechEnd(text),
+    onTranscriptionUpdate: (text: string) => streamingConsumers.emitTranscriptionUpdate(text),
   }
   const {
     trackAudioDeviceUnavailable,
@@ -841,26 +843,34 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
     session: NonNullable<typeof streamingSession.value>,
     result: HearingTranscriptionResult,
   ) {
-    if (result.mode !== 'stream' || !result.textStream)
+    if (result.mode !== 'stream' || !result.fullStream)
       return
 
     const sessionSpan = asrSpan
     const sessionCallbacks = session.callbacks
     void (async () => {
       let fullText = ''
+      let latestSnapshotIsFinal = false
       try {
-        const reader = result.textStream.getReader()
+        const reader = result.fullStream.getReader()
 
         while (true) {
           const { done, value } = await reader.read()
           if (done)
             break
-          if (!value)
+          if (value.type === 'transcript.text.snapshot') {
+            latestSnapshotIsFinal = value.isFinal
+            fullText = value.text
+            sessionCallbacks?.onTranscriptionUpdate?.(fullText)
+            continue
+          }
+          if (value.type !== 'transcript.text.delta' || !value.delta)
             continue
 
-          fullText += value
-          sessionSpan?.addEvent(IOEvents.ASRSentenceEnd, { [IOAttributes.ASRText]: value })
-          sessionCallbacks?.onSentenceEnd?.(value)
+          fullText += value.delta
+          sessionCallbacks?.onTranscriptionUpdate?.(fullText)
+          sessionSpan?.addEvent(IOEvents.ASRSentenceEnd, { [IOAttributes.ASRText]: value.delta })
+          sessionCallbacks?.onSentenceEnd?.(value.delta)
         }
       }
       catch (err) {
@@ -868,6 +878,10 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
           console.error('Error reading text stream:', err)
       }
       finally {
+        if (latestSnapshotIsFinal && fullText.trim()) {
+          sessionSpan?.addEvent(IOEvents.ASRSentenceEnd, { [IOAttributes.ASRText]: fullText })
+          sessionCallbacks?.onSentenceEnd?.(fullText)
+        }
         sessionSpan?.setAttribute(IOAttributes.ASRText, fullText)
         sessionSpan?.end()
         if (asrSpan === sessionSpan)
@@ -972,7 +986,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
       supportsStreamInput: supportsStreamInput.value,
       hasStream: !!stream,
       providerId: activeTranscriptionProvider.value,
-      hasCallbacks: !!(options.onSentenceEnd || options.onSpeechEnd),
+      hasCallbacks: !!(options.onSentenceEnd || options.onSpeechEnd || options.onTranscriptionUpdate),
     })
 
     if (!supportsStreamInput.value) {

packages/stage-ui/src/stores/modules/streaming-transcription-consumers.test.ts

@@ -60,4 +60,17 @@ describe('streaming transcription consumers', () => {
 
     consoleError.mockRestore()
   })
+
+  it('routes complete transcript updates independently from final sentences', () => {
+    const consumers = new StreamingTranscriptionConsumers()
+    const onSentenceEnd = vi.fn()
+    const onTranscriptionUpdate = vi.fn()
+    consumers.register({ consumerId: 'input', onSentenceEnd, onTranscriptionUpdate })
+
+    consumers.emitTranscriptionUpdate('provider correction')
+
+    expect(onTranscriptionUpdate).toHaveBeenCalledOnce()
+    expect(onTranscriptionUpdate).toHaveBeenCalledWith('provider correction')
+    expect(onSentenceEnd).not.toHaveBeenCalled()
+  })
 })

packages/stage-ui/src/stores/modules/streaming-transcription-consumers.ts

@@ -2,6 +2,8 @@
 export interface StreamingTranscriptionCallbacks {
   onSentenceEnd?: (delta: string) => void
   onSpeechEnd?: (text: string) => void
+  /** Receives the complete current transcript after each provider update. */
+  onTranscriptionUpdate?: (text: string) => void
 }
 
 /** A consumer with a stable identity and its current callbacks. */
@@ -24,6 +26,7 @@ export class StreamingTranscriptionConsumers {
     this.consumers.set(consumer.consumerId, {
       onSentenceEnd: consumer.onSentenceEnd,
       onSpeechEnd: consumer.onSpeechEnd,
+      onTranscriptionUpdate: consumer.onTranscriptionUpdate,
     })
   }
 
@@ -42,6 +45,11 @@ export class StreamingTranscriptionConsumers {
     this.emit('onSpeechEnd', text)
   }
 
+  /** Sends the complete current transcript to all current consumers. */
+  emitTranscriptionUpdate(text: string) {
+    this.emit('onTranscriptionUpdate', text)
+  }
+
   private emit(callbackName: keyof StreamingTranscriptionCallbacks, text: string) {
     for (const [consumerId, callbacks] of this.consumers) {
       try {