diff --git a/apps/stage-pocket/src/pages/index.vue b/apps/stage-pocket/src/pages/index.vue index 264d54cdd..cc15465c6 100644 --- a/apps/stage-pocket/src/pages/index.vue +++ b/apps/stage-pocket/src/pages/index.vue @@ -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 diff --git a/apps/stage-tamagotchi/src/renderer/pages/index.vue b/apps/stage-tamagotchi/src/renderer/pages/index.vue index 0e73a6335..7118bc614 100644 --- a/apps/stage-tamagotchi/src/renderer/pages/index.vue +++ b/apps/stage-tamagotchi/src/renderer/pages/index.vue @@ -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), diff --git a/apps/stage-web/src/pages/index.vue b/apps/stage-web/src/pages/index.vue index 1fa2221fc..e7cb94b7b 100644 --- a/apps/stage-web/src/pages/index.vue +++ b/apps/stage-web/src/pages/index.vue @@ -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) diff --git a/packages/stage-layouts/src/composables/use-transcriptions.test.ts b/packages/stage-layouts/src/composables/use-transcriptions.test.ts index d4537145c..b1a692d04 100644 --- a/packages/stage-layouts/src/composables/use-transcriptions.test.ts +++ b/packages/stage-layouts/src/composables/use-transcriptions.test.ts @@ -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() }) }) diff --git a/packages/stage-layouts/src/composables/use-transcriptions.ts b/packages/stage-layouts/src/composables/use-transcriptions.ts index 8623501fa..cae576879 100644 --- a/packages/stage-layouts/src/composables/use-transcriptions.ts +++ b/packages/stage-layouts/src/composables/use-transcriptions.ts @@ -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 @@ -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 | 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' }) diff --git a/packages/stage-pages/src/pages/settings/modules/hearing.vue b/packages/stage-pages/src/pages/settings/modules/hearing.vue index 3bc7b892e..e22b65faf 100644 --- a/packages/stage-pages/src/pages/settings/modules/hearing.vue +++ b/packages/stage-pages/src/pages/settings/modules/hearing.vue @@ -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() const error = ref('') @@ -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 diff --git a/packages/stage-ui/src/stores/modules/hearing.ts b/packages/stage-ui/src/stores/modules/hearing.ts index 1c2c28bcc..8e4df1620 100644 --- a/packages/stage-ui/src/stores/modules/hearing.ts +++ b/packages/stage-ui/src/stores/modules/hearing.ts @@ -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 { file: Blob fileName?: string @@ -112,12 +107,10 @@ interface HearingTranscriptionInvokeOptions { providerOptions?: Record } -interface MediaStreamTranscriptionOptions { +interface MediaStreamTranscriptionOptions extends StreamingTranscriptionConsumer { sampleRate?: number providerOptions?: Record 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 providerId?: string - callbacks?: { - onSentenceEnd?: (delta: string) => void - onSpeechEnd?: (text: string) => void - } + callbacks?: StreamingTranscriptionCallbacks }>() const streamingVadSession = shallowRef<{ vad: Pick, 'dispose'> lifecycle: ReturnType providerId: string - callbacks: { - onSentenceEnd?: (delta: string) => void - onSpeechEnd?: (text: string) => void - } + callbacks: StreamingTranscriptionCallbacks activeSegment?: { audioChunks: ArrayBuffer[] audioStreamController?: ReadableStreamDefaultController @@ -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, ) { 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 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, } diff --git a/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.test.ts b/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.test.ts new file mode 100644 index 000000000..e0591ba41 --- /dev/null +++ b/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.test.ts @@ -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() + }) +}) diff --git a/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.ts b/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.ts new file mode 100644 index 000000000..faead8b21 --- /dev/null +++ b/packages/stage-ui/src/stores/modules/streaming-transcription-consumers.ts @@ -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() + + /** 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) + } + } + } +}