fix(stage-ui): share streaming transcription sessions (#2260)

2026-08-14 00:48:06 +00:00 · 2026-08-12 00:36:26 +08:00
parent 9f34877bab
commit 788a509940
9 changed files with 203 additions and 72 deletions
@@ -48,13 +48,16 @@ const settingsAudioDeviceStore = useSettingsAudioDevice()
 const { stream, enabled } = storeToRefs(settingsAudioDeviceStore)
 const { startRecord, stopRecord, onStopRecord } = useAudioRecorder(stream)
 const hearingPipeline = useHearingSpeechInputPipeline()
-const { transcribeForRecording, transcribeForMediaStream, stopStreamingTranscription } = hearingPipeline
+const { removeStreamingTranscriptionConsumer, transcribeForRecording, transcribeForMediaStream, stopStreamingTranscription } = hearingPipeline
 const { supportsStreamInput } = storeToRefs(hearingPipeline)
 const providersStore = useProviderStore()
 const consciousnessStore = useConsciousnessStore()
 const { activeProvider: activeChatProvider, activeModel: activeChatModel } = storeToRefs(consciousnessStore)
 const chatStore = useChatStore()

+/** Identifies this page in the shared streaming transcription session. */
+const transcriptionConsumerId = 'stage-pocket:voice-input'
+
 const shouldUseStreamInput = computed(() => supportsStreamInput.value && !!stream.value)

 const {
@@ -104,6 +107,7 @@ async function handleSpeechStart() {
    // Use both callbacks to support incremental updates and final transcript replacement.
    // ChatArea uses only onSentenceEnd to avoid re-adding deleted text.
    await transcribeForMediaStream(stream.value, {
+      consumerId: transcriptionConsumerId,
      onSentenceEnd: (delta) => {
        const finalText = delta
        if (!finalText || !finalText.trim()) {
@@ -141,6 +145,7 @@ async function handleSpeechEnd() {

 function stopAudioInteraction() {
  try {
+    removeStreamingTranscriptionConsumer(transcriptionConsumerId)
    stopOnStopRecord?.()
    stopOnStopRecord = undefined
    // Stop any active streaming transcription sessions to prevent session leakage
@@ -325,11 +325,13 @@ const { nowSpeaking } = storeToRefs(useSpeakingStore())
 const hearingStore = useHearingStore()
 const { activeTranscriptionModel, activeTranscriptionProvider } = storeToRefs(hearingStore)
 const hearingPipeline = useHearingSpeechInputPipeline()
-const { transcribeForMediaStream, stopStreamingTranscription } = hearingPipeline
+const { removeStreamingTranscriptionConsumer, transcribeForMediaStream, stopStreamingTranscription } = hearingPipeline
 const { error: transcriptionError, supportsStreamInput } = storeToRefs(hearingPipeline)
 const chatStore = useChatStore()
 const chatSession = useChatSessionStore()
 const streamingTranscriptionUnavailable = ref(false)
+/** Identifies this page in the shared streaming transcription session. */
+const transcriptionConsumerId = 'stage-tamagotchi:voice-input'
 const shouldUseStreamInput = computed(() => supportsStreamInput.value && !!stream.value && !streamingTranscriptionUnavailable.value)
 const voiceTranscriptBuffer = createTranscriptBuffer({
  flushDelayMs: 1200,
@@ -588,17 +590,20 @@ async function startAudioInteractionConsumers() {
      return

    await transcribeForMediaStream(currentStream, {
+      consumerId: transcriptionConsumerId,
      onSentenceEnd: handleStreamingSentenceEnd,
      onSpeechEnd: handleStreamingSpeechEnd,
    })

    if (inspectVoiceInputStreamingRequestGate().skip) {
+      removeStreamingTranscriptionConsumer(transcriptionConsumerId)
      await stopStreamingTranscription(true)
      return
    }

    if (transcriptionError.value) {
      streamingTranscriptionUnavailable.value = true
+      removeStreamingTranscriptionConsumer(transcriptionConsumerId)
      await stopStreamingTranscription(true)
      console.warn('[Main Page] Streaming transcription unavailable; using recorder-backed fallback:', transcriptionError.value)
    }
@@ -616,6 +621,7 @@ async function stopAudioInteractionConsumers(options: StopAudioInteractionOption

  clearAssistantSpeechResumeTimer()
  voiceInputGeneration += 1
+  removeStreamingTranscriptionConsumer(transcriptionConsumerId)

  await Promise.all([
    stopStreamingTranscription(true),
@@ -45,13 +45,16 @@ const settingsAudioDeviceStore = useSettingsAudioDevice()
 const { stream, enabled } = storeToRefs(settingsAudioDeviceStore)
 const { startRecord, stopRecord, onStopRecord } = useAudioRecorder(stream)
 const hearingPipeline = useHearingSpeechInputPipeline()
-const { stopStreamingTranscription, transcribeForMediaStream, transcribeForRecording } = hearingPipeline
+const { removeStreamingTranscriptionConsumer, stopStreamingTranscription, transcribeForMediaStream, transcribeForRecording } = hearingPipeline
 const { supportsStreamInput } = storeToRefs(hearingPipeline)
 const providersStore = useProviderStore()
 const consciousnessStore = useConsciousnessStore()
 const { activeProvider: activeChatProvider, activeModel: activeChatModel } = storeToRefs(consciousnessStore)
 const chatStore = useChatStore()

+/** Identifies this page in the shared streaming transcription session. */
+const transcriptionConsumerId = 'stage-web:voice-input'
+
 const shouldUseStreamInput = computed(() => supportsStreamInput.value && !!stream.value)

 const {
@@ -91,6 +94,7 @@ async function startAudioInteraction() {

    if (shouldUseStreamInput.value && stream.value) {
      await transcribeForMediaStream(stream.value, {
+        consumerId: transcriptionConsumerId,
        onSentenceEnd: text => void sendVoiceInputTextToChat(text),
      })
      return
@@ -128,6 +132,7 @@ async function handleSpeechEnd() {

 function stopAudioInteraction() {
  try {
+    removeStreamingTranscriptionConsumer(transcriptionConsumerId)
    stopOnStopRecord?.()
    stopOnStopRecord = undefined
    void stopStreamingTranscription(true)
@@ -20,6 +20,7 @@ function createMockStore() {
 const mockTranscribedContent = 'test content'
 function createMockPipeline() {
  return {
+    removeStreamingTranscriptionConsumer: vi.fn(),
    transcribeForMediaStream: vi.fn().mockImplementation((_stream, options: { onSentenceEnd: (delta: string) => void }) => {
      options.onSentenceEnd(mockTranscribedContent)
    }),
@@ -330,6 +331,7 @@ describe('useTranscriptions', () => {
      await nextTick()
      expect(isListening.value).toBe(false)
      expect(mockHearingPipeline.stopStreamingTranscription).toHaveBeenCalledWith(true)
+      expect(mockHearingPipeline.removeStreamingTranscriptionConsumer).toHaveBeenCalledOnce()
    })

    it('should stop streaming on unmount', async () => {
@@ -351,6 +353,7 @@ describe('useTranscriptions', () => {
      app.unmount()
      await nextTick()
      expect(mockHearingPipeline.stopStreamingTranscription).toHaveBeenCalled()
+      expect(mockHearingPipeline.removeStreamingTranscriptionConsumer).toHaveBeenCalled()
    })
  })

@@ -5,7 +5,7 @@ import { useProviderStore } from '@proj-airi/stage-ui/stores/providers/provider'
 import { useSettingsAudioDevice } from '@proj-airi/stage-ui/stores/settings'
 import { until } from '@vueuse/core'
 import { storeToRefs } from 'pinia'
-import { nextTick, onScopeDispose, ref, toValue, watch } from 'vue'
+import { nextTick, onScopeDispose, ref, toValue, useId, watch } from 'vue'

 interface TranscriptionOptions {
  messageInputRef: Ref<string>
@@ -19,7 +19,7 @@ export function useTranscriptions(options: TranscriptionOptions) {
  const hearingStore = useHearingStore()
  const audioDeviceSettingsStore = useSettingsAudioDevice()
  const hearingPipeline = useHearingSpeechInputPipeline()
-  const { transcribeForMediaStream, stopStreamingTranscription } = hearingPipeline
+  const { removeStreamingTranscriptionConsumer, transcribeForMediaStream, stopStreamingTranscription } = hearingPipeline
  const { supportsStreamInput } = storeToRefs(hearingPipeline)
  const { configured: hearingConfigured, autoSendEnabled, autoSendDelay } = storeToRefs(hearingStore)
  const { enabled: hearingEnabled, stream } = storeToRefs(audioDeviceSettingsStore)
@@ -27,6 +27,7 @@ export function useTranscriptions(options: TranscriptionOptions) {
  const { askPermission, startStream } = audioDeviceSettingsStore

  const isListening = ref(false)
+  const transcriptionConsumerId = `interactive-area:${useId()}`

  // Auto-send logic
  let autoSendTimeout: ReturnType<typeof setTimeout> | undefined
@@ -58,6 +59,8 @@ export function useTranscriptions(options: TranscriptionOptions) {
  }

  const stopStreaming = async () => {
+    removeStreamingTranscriptionConsumer(transcriptionConsumerId)
+
    if (!isListening.value)
      return

@@ -176,6 +179,7 @@ export function useTranscriptions(options: TranscriptionOptions) {
    // Set listening state AFTER successful call
    try {
      await transcribeForMediaStream(stream.value, {
+        consumerId: transcriptionConsumerId,
        onSentenceEnd: (delta) => {
          if (delta && delta.trim()) {
            console.info('Received transcription delta:', delta, { source: 'useTranscriptions' })
@@ -46,6 +46,7 @@ const { startAnalyzer, stopAnalyzer, onAnalyzerUpdate, volumeLevel } = useAudioA
 const { audioContext } = storeToRefs(useAudioContext())
 const hearingSpeechInputPipeline = useHearingSpeechInputPipeline()
 const {
+  removeStreamingTranscriptionConsumer,
  transcribeForRecording,
  transcribeForMediaStream,
  stopStreamingTranscription,
@@ -55,6 +56,11 @@ const {
  error: transcriptionPipelineError,
 } = storeToRefs(hearingSpeechInputPipeline)

+/** Identifies monitoring callbacks in the shared streaming transcription session. */
+const monitoringTranscriptionConsumerId = 'hearing-settings:monitoring'
+/** Identifies test callbacks in the shared streaming transcription session. */
+const testTranscriptionConsumerId = 'hearing-settings:test'
+
 const animationFrame = ref<number>()

 const error = ref<string>('')
@@ -149,6 +155,7 @@ async function handleSpeechStart() {
    // Use both callbacks to support incremental updates and final transcript replacement.
    // ChatArea uses only onSentenceEnd to avoid re-adding deleted text.
    await transcribeForMediaStream(stream.value, {
+      consumerId: monitoringTranscriptionConsumerId,
      onSentenceEnd: (delta) => {
        transcriptions.value.push(delta)
      },
@@ -248,6 +255,7 @@ async function stopAudioMonitoring() {
    animationFrame.value = undefined
  }

+  removeStreamingTranscriptionConsumer(monitoringTranscriptionConsumerId)
  await stopStreamingTranscription(true, activeTranscriptionProvider.value)
  if (stream.value) { // Stop media stream
    stopStream()
@@ -397,6 +405,7 @@ async function startSTTTest() {
      console.info('Starting STT test with streaming input for provider:', activeTranscriptionProvider.value)

      await transcribeForMediaStream(stream.value, {
+        consumerId: testTranscriptionConsumerId,
        onSentenceEnd: (delta) => {
          if (delta && delta.trim()) {
            testStreamingText.value += `${delta} `
@@ -467,6 +476,7 @@ async function stopSTTTest() {
  isTestingSTT.value = false
  isTranscribing.value = false
  testStatusMessage.value = 'Stopped'
+  removeStreamingTranscriptionConsumer(testTranscriptionConsumerId)

  try {
    // Stop streaming transcription if active
@@ -3,6 +3,8 @@ import type { TranscriptionProviderWithExtraOptions } from '@xsai-ext/providers/
 import type { WithUnknown } from '@xsai/shared'
 import type { StreamTranscriptionResult, StreamTranscriptionOptions as XSAIStreamTranscriptionOptions } from '@xsai/stream-transcription'

+import type { StreamingTranscriptionCallbacks, StreamingTranscriptionConsumer } from './streaming-transcription-consumers'
+
 import { errorMessageFrom, tryCatch } from '@moeru/std'
 import { errorMessageFromValue, IOAttributes, IOEvents, IOSpanNames, IOSubsystems } from '@proj-airi/stage-shared'
 import { useLocalStorageManualReset } from '@proj-airi/stage-shared/composables'
@@ -22,6 +24,7 @@ import { streamTranscription } from '../../libs/providers/stream-transcription'
 import { useVAD } from '../ai/models/vad'
 import { useProviderConfigStore } from '../providers/config'
 import { useProviderStore } from '../providers/provider'
+import { StreamingTranscriptionConsumers } from './streaming-transcription-consumers'

 function errorMessage(err: unknown): string {
  const msg = errorMessageFromValue(err)
@@ -78,14 +81,6 @@ function transcriptionAnalyticsErrorCode(err: unknown): TranscriptionAnalyticsEr
  return message ? 'provider_error' : 'unknown'
 }

-function haveStreamingCallbacksChanged(
-  previous: { onSentenceEnd?: (delta: string) => void, onSpeechEnd?: (text: string) => void } | undefined,
-  next: { onSentenceEnd?: (delta: string) => void, onSpeechEnd?: (text: string) => void },
-): boolean {
-  return (next.onSentenceEnd !== undefined && next.onSentenceEnd !== previous?.onSentenceEnd)
-    || (next.onSpeechEnd !== undefined && next.onSpeechEnd !== previous?.onSpeechEnd)
-}
-
 export interface StreamTranscriptionFileInputOptions extends Omit<XSAIStreamTranscriptionOptions, 'file' | 'fileName'> {
  file: Blob
  fileName?: string
@@ -112,12 +107,10 @@ interface HearingTranscriptionInvokeOptions {
  providerOptions?: Record<string, unknown>
 }

-interface MediaStreamTranscriptionOptions {
+interface MediaStreamTranscriptionOptions extends StreamingTranscriptionConsumer {
  sampleRate?: number
  providerOptions?: Record<string, unknown>
  idleTimeoutMs?: number
-  onSentenceEnd?: (delta: string) => void
-  onSpeechEnd?: (text: string) => void
 }

 export const CONFIDENCE_THRESHOLD_DISABLED = -3
@@ -576,6 +569,11 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
  const { activeTranscriptionProvider, activeTranscriptionModel } = storeToRefs(hearingStore)
  const providersStore = useProviderStore()
  const providerStore = useProviderConfigStore()
+  const streamingConsumers = new StreamingTranscriptionConsumers()
+  const streamingCallbacks = {
+    onSentenceEnd: (delta: string) => streamingConsumers.emitSentenceEnd(delta),
+    onSpeechEnd: (text: string) => streamingConsumers.emitSpeechEnd(text),
+  }
  const {
    trackAudioDeviceUnavailable,
    trackVoiceInputCancelled,
@@ -590,19 +588,13 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
    result?: HearingTranscriptionResult & { recognition?: any }
    idleTimer?: ReturnType<typeof setTimeout>
    providerId?: string
-    callbacks?: {
-      onSentenceEnd?: (delta: string) => void
-      onSpeechEnd?: (text: string) => void
-    }
+    callbacks?: StreamingTranscriptionCallbacks
  }>()
  const streamingVadSession = shallowRef<{
    vad: Pick<ReturnType<typeof useVAD>, 'dispose'>
    lifecycle: ReturnType<typeof createVadStreamingSession>
    providerId: string
-    callbacks: {
-      onSentenceEnd?: (delta: string) => void
-      onSpeechEnd?: (text: string) => void
-    }
+    callbacks: StreamingTranscriptionCallbacks
    activeSegment?: {
      audioChunks: ArrayBuffer[]
      audioStreamController?: ReadableStreamDefaultController<ArrayBuffer>
@@ -611,6 +603,11 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech

  let asrSpan: Span | undefined

+  /** Removes callbacks owned by one streaming transcription consumer. */
+  function removeStreamingTranscriptionConsumer(consumerId: string) {
+    streamingConsumers.remove(consumerId)
+  }
+
  function startStreamingAsrSpan(providerId: string) {
    activeTurnSpan.value?.end()
    const turnSpan = startSpan(IOSpanNames.InteractionTurn)
@@ -882,7 +879,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech

  async function startVadRealtimeTranscription(
    providerId: string,
-    options: MediaStreamTranscriptionOptions | undefined,
+    options: MediaStreamTranscriptionOptions,
    vadSession: NonNullable<typeof streamingVadSession.value>,
  ) {
    const segment = vadSession.activeSegment
@@ -914,7 +911,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
      {
        providerOptions: {
          abortSignal: abortController.signal,
-          ...options?.providerOptions,
+          ...options.providerOptions,
        },
      },
    )
@@ -929,7 +926,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
  async function startVadStreamingTranscription(
    stream: MediaStream,
    providerId: string,
-    options: MediaStreamTranscriptionOptions | undefined,
+    options: MediaStreamTranscriptionOptions,
  ) {
    let vadSession!: NonNullable<typeof streamingVadSession.value>
    const vad = useVAD(vadWorkletUrl, {
@@ -958,10 +955,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
      vad,
      lifecycle,
      providerId,
-      callbacks: {
-        onSentenceEnd: options?.onSentenceEnd,
-        onSpeechEnd: options?.onSpeechEnd,
-      },
+      callbacks: streamingCallbacks,
    }
    streamingVadSession.value = vadSession

@@ -973,12 +967,12 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
    await vad.start(stream)
  }

-  async function transcribeForMediaStream(stream: MediaStream, options?: MediaStreamTranscriptionOptions) {
+  async function transcribeForMediaStream(stream: MediaStream, options: MediaStreamTranscriptionOptions) {
    console.info('[Hearing Pipeline] transcribeForMediaStream called', {
      supportsStreamInput: supportsStreamInput.value,
      hasStream: !!stream,
      providerId: activeTranscriptionProvider.value,
-      hasCallbacks: !!(options?.onSentenceEnd || options?.onSpeechEnd),
+      hasCallbacks: !!(options.onSentenceEnd || options.onSpeechEnd),
    })

    if (!supportsStreamInput.value) {
@@ -987,6 +981,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
    }

    error.value = undefined
+    let consumerRegistered = false

    try {
      const providerId = activeTranscriptionProvider.value
@@ -1013,36 +1008,22 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
          return
        }

+        streamingConsumers.register(options)
+        consumerRegistered = true
+
        // Check if session already exists and reuse it
        const existingSession = streamingSession.value
        if (existingSession && existingSession.providerId === 'browser-web-speech-api') {
-          const nextCallbacks = {
-            onSentenceEnd: options?.onSentenceEnd,
-            onSpeechEnd: options?.onSpeechEnd,
+          const idleTimeout = options.idleTimeoutMs ?? DEFAULT_STREAM_IDLE_TIMEOUT
+          if (existingSession.idleTimer) {
+            clearTimeout(existingSession.idleTimer)
+            existingSession.idleTimer = setTimeout(async () => {
+              await stopStreamingTranscription(false, existingSession.providerId)
+            }, idleTimeout)
          }
-          // For Web Speech API, if callbacks are provided and different, we need to restart
-          // because recognition instance callbacks are set once and can't be changed
-          const hasNewCallbacks = haveStreamingCallbacksChanged(existingSession.callbacks, nextCallbacks)

-          if (hasNewCallbacks) {
-            console.info('Web Speech API: New callbacks provided, restarting session to use them')
-            await stopStreamingTranscription(false, existingSession.providerId)
-            // Continue to create new session below
-            // Note: stopStreamingTranscription already clears streamingSession.value and waits for async cleanup
-          }
-          else {
-            // No new callbacks - just bump idle timer and reuse existing session
-            const idleTimeout = options?.idleTimeoutMs ?? DEFAULT_STREAM_IDLE_TIMEOUT
-            if (existingSession.idleTimer) {
-              clearTimeout(existingSession.idleTimer)
-              existingSession.idleTimer = setTimeout(async () => {
-                await stopStreamingTranscription(false, existingSession.providerId)
-              }, idleTimeout)
-            }
-
-            console.info('Web Speech API session already active, reusing existing session (no callback changes)')
-            return
-          }
+          console.info('Web Speech API session already active, reusing it with updated consumers')
+          return
        }

        startStreamingAsrSpan(providerId)
@@ -1095,7 +1076,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
            if (asrSpan)
              asrSpan.addEvent(IOEvents.ASRSentenceEnd, { [IOAttributes.ASRText]: delta })
            // Call the options callback
-            options?.onSentenceEnd?.(delta)
+            streamingCallbacks.onSentenceEnd(delta)
          },
          onSpeechEnd: (text) => {
            if (asrSpan) {
@@ -1104,7 +1085,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
              asrSpan = undefined
            }
            // Call the options callback
-            options?.onSpeechEnd?.(text)
+            streamingCallbacks.onSpeechEnd(text)
          },
        })

@@ -1119,10 +1100,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
          result: { ...result, mode: 'stream' as const, recognition: recognitionInstance },
          idleTimer,
          providerId,
-          callbacks: {
-            onSentenceEnd: options?.onSentenceEnd,
-            onSpeechEnd: options?.onSpeechEnd,
-          },
+          callbacks: streamingCallbacks,
        } as any // Type assertion needed because recognition is extra

        // Initial idle timer (only if enabled)
@@ -1154,19 +1132,17 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
        return
      }

+      streamingConsumers.register(options)
+      consumerRegistered = true
+
      const existingVadSession = streamingVadSession.value
      if (existingVadSession) {
-        const hasNewCallbacks = haveStreamingCallbacksChanged(existingVadSession.callbacks, {
-          onSentenceEnd: options?.onSentenceEnd,
-          onSpeechEnd: options?.onSpeechEnd,
-        })
-
-        if (hasNewCallbacks) {
-          console.info('[Hearing Pipeline] New callbacks provided, restarting VAD detection')
+        if (existingVadSession.providerId !== providerId) {
+          console.info('[Hearing Pipeline] Provider changed, restarting VAD detection')
          await stopStreamingTranscription(false, existingVadSession.providerId)
        }
        else {
-          console.info('[Hearing Pipeline] VAD detection already active, reusing it')
+          console.info('[Hearing Pipeline] VAD detection already active, reusing it with updated consumers')
          return
        }
      }
@@ -1174,6 +1150,9 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
      await startVadStreamingTranscription(stream, providerId, options)
    }
    catch (err) {
+      if (consumerRegistered)
+        streamingConsumers.remove(options.consumerId)
+
      endStreamingAsrSpan()

      if (isExpectedStreamStopError(err))
@@ -1256,6 +1235,7 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech

    transcribeForRecording,
    transcribeForMediaStream,
+    removeStreamingTranscriptionConsumer,
    stopStreamingTranscription,
    supportsStreamInput,
  }
@@ -0,0 +1,63 @@
+import { describe, expect, it, vi } from 'vitest'
+
+import { StreamingTranscriptionConsumers } from './streaming-transcription-consumers'
+
+describe('streaming transcription consumers', () => {
+  it('updates and removes consumers without restarting other callbacks', () => {
+    // ROOT CAUSE:
+    //
+    // The Hearing store kept one callback pair on the provider session. A new
+    // caller had to restart that session to receive results, which disconnected
+    // the existing caller.
+    //
+    // A consumer registry keeps the provider callbacks stable and routes each
+    // result to the current callback set for every owner.
+    const consumers = new StreamingTranscriptionConsumers()
+    const firstOriginal = vi.fn()
+    const firstUpdated = vi.fn()
+    const second = vi.fn()
+
+    consumers.register({ consumerId: 'first', onSentenceEnd: firstOriginal })
+    consumers.register({ consumerId: 'second', onSentenceEnd: second })
+    consumers.register({ consumerId: 'first', onSentenceEnd: firstUpdated })
+
+    consumers.emitSentenceEnd('hello')
+
+    expect(firstOriginal).not.toHaveBeenCalled()
+    expect(firstUpdated).toHaveBeenCalledOnce()
+    expect(firstUpdated).toHaveBeenCalledWith('hello')
+    expect(second).toHaveBeenCalledOnce()
+    expect(second).toHaveBeenCalledWith('hello')
+
+    consumers.remove('first')
+    consumers.emitSentenceEnd('world')
+
+    expect(firstUpdated).toHaveBeenCalledOnce()
+    expect(second).toHaveBeenCalledTimes(2)
+    expect(second).toHaveBeenLastCalledWith('world')
+  })
+
+  it('continues delivery when one consumer throws', () => {
+    const consumers = new StreamingTranscriptionConsumers()
+    const error = new Error('consumer failed')
+    const consoleError = vi.spyOn(console, 'error').mockImplementation(() => {})
+    const second = vi.fn()
+
+    consumers.register({
+      consumerId: 'first',
+      onSpeechEnd: () => { throw error },
+    })
+    consumers.register({ consumerId: 'second', onSpeechEnd: second })
+
+    consumers.emitSpeechEnd('complete')
+
+    expect(second).toHaveBeenCalledOnce()
+    expect(second).toHaveBeenCalledWith('complete')
+    expect(consoleError).toHaveBeenCalledWith(
+      '[Hearing Pipeline] Streaming consumer first onSpeechEnd failed:',
+      error,
+    )
+
+    consoleError.mockRestore()
+  })
+})
@@ -0,0 +1,55 @@
+/** Callbacks that receive results from one shared streaming transcription session. */
+export interface StreamingTranscriptionCallbacks {
+  onSentenceEnd?: (delta: string) => void
+  onSpeechEnd?: (text: string) => void
+}
+
+/** A consumer with a stable identity and its current callbacks. */
+export interface StreamingTranscriptionConsumer extends StreamingTranscriptionCallbacks {
+  /** Identifies the callback owner across registration updates and cleanup. */
+  consumerId: string
+}
+
+/**
+ * Routes one provider session to independent consumers.
+ *
+ * A consumer can replace its callbacks without restarting the provider. The
+ * registry isolates callback failures so one consumer cannot block another.
+ */
+export class StreamingTranscriptionConsumers {
+  private readonly consumers = new Map<string, StreamingTranscriptionCallbacks>()
+
+  /** Registers or replaces the callbacks for one consumer. */
+  register(consumer: StreamingTranscriptionConsumer) {
+    this.consumers.set(consumer.consumerId, {
+      onSentenceEnd: consumer.onSentenceEnd,
+      onSpeechEnd: consumer.onSpeechEnd,
+    })
+  }
+
+  /** Removes callbacks for one consumer. */
+  remove(consumerId: string) {
+    this.consumers.delete(consumerId)
+  }
+
+  /** Sends a completed sentence to all current consumers. */
+  emitSentenceEnd(delta: string) {
+    this.emit('onSentenceEnd', delta)
+  }
+
+  /** Sends completed speech text to all current consumers. */
+  emitSpeechEnd(text: string) {
+    this.emit('onSpeechEnd', text)
+  }
+
+  private emit(callbackName: keyof StreamingTranscriptionCallbacks, text: string) {
+    for (const [consumerId, callbacks] of this.consumers) {
+      try {
+        callbacks[callbackName]?.(text)
+      }
+      catch (cause) {
+        console.error(`[Hearing Pipeline] Streaming consumer ${consumerId} ${callbackName} failed:`, cause)
+      }
+    }
+  }
+}