fix(stage-ui,stage-tamagotchi,stage-web): incorrect implementation of Aliyun NLS

2025-12-26 01:39:11 +08:00
parent bc6f2ac746
commit 1f4650a105
11 changed files with 576 additions and 60 deletions
@@ -103,11 +103,18 @@ watch([isOutsideFor250Ms, isAroundWindowBorderFor250Ms, isOutsideWindow, isTrans
 const settingsAudioDeviceStore = useSettingsAudioDevice()
 const { stream, enabled } = storeToRefs(settingsAudioDeviceStore)
 const { startRecord, stopRecord, onStopRecord } = useAudioRecorder(stream)
-const { transcribeForRecording } = useHearingSpeechInputPipeline()
+const hearingPipeline = useHearingSpeechInputPipeline()
+const {
+  transcribeForRecording,
+  transcribeForMediaStream,
+  stopStreamingTranscription,
+} = hearingPipeline
+const { supportsStreamInput } = storeToRefs(hearingPipeline)
 const providersStore = useProvidersStore()
 const consciousnessStore = useConsciousnessStore()
 const { activeProvider: activeChatProvider, activeModel: activeChatModel } = storeToRefs(consciousnessStore)
 const chatStore = useChatStore()
+const shouldUseStreamInput = computed(() => supportsStreamInput.value && !!stream.value)

 const {
  init: initVAD,
@@ -116,8 +123,12 @@ const {
  loaded: vadLoaded,
 } = useVAD(workletUrl, {
  threshold: ref(0.6),
-  onSpeechStart: () => startRecord(),
-  onSpeechEnd: () => stopRecord(),
+  onSpeechStart: () => {
+    void handleSpeechStart()
+  },
+  onSpeechEnd: () => {
+    void handleSpeechEnd()
+  },
 })

 let stopOnStopRecord: (() => void) | undefined
@@ -128,6 +139,49 @@ type CaptionChannelEvent
    | { type: 'caption-assistant', text: string }
 const { post: postCaption } = useBroadcastChannel<CaptionChannelEvent, CaptionChannelEvent>({ name: 'airi-caption-overlay' })

+async function handleSpeechStart() {
+  if (shouldUseStreamInput.value && stream.value) {
+    await transcribeForMediaStream(stream.value, {
+      onSentenceEnd: (delta) => {
+        const finalText = delta
+        if (!finalText || !finalText.trim()) {
+          return
+        }
+
+        postCaption({ type: 'caption-speaker', text: finalText })
+
+        void (async () => {
+          try {
+            const provider = await providersStore.getProviderInstance(activeChatProvider.value)
+            if (!provider || !activeChatModel.value)
+              return
+
+            await chatStore.send(finalText, { model: activeChatModel.value, chatProvider: provider as ChatProvider })
+          }
+          catch (err) {
+            console.error('Failed to send chat from voice:', err)
+          }
+        })()
+      },
+      onSpeechEnd: (text) => {
+        postCaption({ type: 'caption-speaker', text })
+      },
+    })
+    return
+  }
+
+  startRecord()
+}
+
+async function handleSpeechEnd() {
+  if (shouldUseStreamInput.value) {
+    // Keep streaming session alive; idle timer in pipeline will handle teardown.
+    return
+  }
+
+  stopRecord()
+}
+
 async function startAudioInteraction() {
  try {
    await initVAD()
@@ -136,6 +190,9 @@ async function startAudioInteraction() {

    // Hook once
    stopOnStopRecord = onStopRecord(async (recording) => {
+      if (shouldUseStreamInput.value)
+        return
+
      const text = await transcribeForRecording(recording)
      if (!text || !text.trim())
        return
@@ -164,6 +221,7 @@ function stopAudioInteraction() {
  try {
    stopOnStopRecord?.()
    stopOnStopRecord = undefined
+    void stopStreamingTranscription(true)
    disposeVAD()
  }
  catch {}
@@ -14,7 +14,7 @@ import { useProvidersStore } from '@proj-airi/stage-ui/stores/providers'
 import { useSettingsAudioDevice } from '@proj-airi/stage-ui/stores/settings'
 import { breakpointsTailwind, useBreakpoints, useMouse } from '@vueuse/core'
 import { storeToRefs } from 'pinia'
-import { onMounted, onUnmounted, ref, useTemplateRef, watch } from 'vue'
+import { computed, onMounted, onUnmounted, ref, useTemplateRef, watch } from 'vue'

 import Header from '../components/Layouts/Header.vue'
 import InteractiveArea from '../components/Layouts/InteractiveArea.vue'
@@ -47,12 +47,16 @@ onMounted(() => syncBackgroundTheme())
 const settingsAudioDeviceStore = useSettingsAudioDevice()
 const { stream, enabled } = storeToRefs(settingsAudioDeviceStore)
 const { startRecord, stopRecord, onStopRecord } = useAudioRecorder(stream)
-const { transcribeForRecording } = useHearingSpeechInputPipeline()
+const hearingPipeline = useHearingSpeechInputPipeline()
+const { transcribeForRecording, transcribeForMediaStream } = hearingPipeline
+const { supportsStreamInput } = storeToRefs(hearingPipeline)
 const providersStore = useProvidersStore()
 const consciousnessStore = useConsciousnessStore()
 const { activeProvider: activeChatProvider, activeModel: activeChatModel } = storeToRefs(consciousnessStore)
 const chatStore = useChatStore()

+const shouldUseStreamInput = computed(() => supportsStreamInput.value && !!stream.value)
+
 const {
  init: initVAD,
  dispose: disposeVAD,
@@ -60,8 +64,8 @@ const {
  loaded: vadLoaded,
 } = useVAD(workletUrl, {
  threshold: ref(0.6),
-  onSpeechStart: () => startRecord(),
-  onSpeechEnd: () => stopRecord(),
+  onSpeechStart: () => handleSpeechStart(),
+  onSpeechEnd: () => handleSpeechEnd(),
 })

 let stopOnStopRecord: (() => void) | undefined
@@ -95,6 +99,44 @@ async function startAudioInteraction() {
  }
 }

+async function handleSpeechStart() {
+  if (shouldUseStreamInput.value && stream.value) {
+    await transcribeForMediaStream(stream.value, {
+      onSentenceEnd: (delta) => {
+        const finalText = delta
+        if (!finalText || !finalText.trim()) {
+          return
+        }
+
+        void (async () => {
+          try {
+            const provider = await providersStore.getProviderInstance(activeChatProvider.value)
+            if (!provider || !activeChatModel.value)
+              return
+
+            await chatStore.send(finalText, { model: activeChatModel.value, chatProvider: provider as ChatProvider })
+          }
+          catch (err) {
+            console.error('Failed to send chat from voice:', err)
+          }
+        })()
+      },
+    })
+    return
+  }
+
+  startRecord()
+}
+
+async function handleSpeechEnd() {
+  if (shouldUseStreamInput.value) {
+    // Keep streaming session alive; idle timer in pipeline will handle teardown.
+    return
+  }
+
+  stopRecord()
+}
+
 function stopAudioInteraction() {
  try {
    stopOnStopRecord?.()
@@ -34,7 +34,14 @@ const { audioInputs, selectedAudioInput, stream } = storeToRefs(useSettingsAudio
 const { startRecord, stopRecord, onStopRecord } = useAudioRecorder(stream)
 const { startAnalyzer, stopAnalyzer, onAnalyzerUpdate, volumeLevel } = useAudioAnalyzer()
 const { audioContext } = storeToRefs(useAudioContext())
-const { transcribeForRecording } = useHearingSpeechInputPipeline()
+const {
+  transcribeForRecording,
+  transcribeForMediaStream,
+  stopStreamingTranscription,
+} = useHearingSpeechInputPipeline()
+const {
+  supportsStreamInput,
+} = storeToRefs(useHearingSpeechInputPipeline())

 const animationFrame = ref<number>()

@@ -54,6 +61,33 @@ const audioURLs = computed(() => {

 const useVADThreshold = ref(0.6) // 0.1 - 0.9
 const useVADModel = ref(true) // Toggle between VAD and volume-based detection
+const shouldUseStreamInput = computed(() => supportsStreamInput.value && !!stream.value)
+
+async function handleSpeechStart() {
+  if (shouldUseStreamInput.value && stream.value) {
+    await transcribeForMediaStream(stream.value, {
+      onSentenceEnd: (delta) => {
+        transcriptions.value.push(delta)
+      },
+      onSpeechEnd: (text) => {
+        transcriptions.value = [text]
+      },
+    })
+    return
+  }
+
+  startRecord()
+}
+
+async function handleSpeechEnd() {
+  if (shouldUseStreamInput.value) {
+    // For streaming providers, keep the session alive; idle timer will handle teardown.
+    return
+  }
+
+  stopRecord()
+}
+
 const {
  init: initVAD,
  dispose: disposeVAD,
@@ -66,8 +100,12 @@ const {
  loading: loadingVAD,
 } = useVAD(workletUrl, {
  threshold: useVADThreshold,
-  onSpeechStart: () => startRecord(),
-  onSpeechEnd: () => stopRecord(),
+  onSpeechStart: () => {
+    void handleSpeechStart()
+  },
+  onSpeechEnd: () => {
+    void handleSpeechEnd()
+  },
 })

 const isSpeechVolume = ref(false) // Volume-based speaking detection
@@ -122,6 +160,8 @@ async function stopAudioMonitoring() {
    cancelAnimationFrame(animationFrame.value)
    animationFrame.value = undefined
  }
+
+  await stopStreamingTranscription(true, activeTranscriptionProvider.value)
  if (stream.value) { // Stop media stream
    stopStream()
  }
@@ -174,6 +214,9 @@ function updateCustomModelName(value: string) {
 }

 onStopRecord(async (recording) => {
+  if (shouldUseStreamInput.value)
+    return
+
  if (recording && recording.size > 0)
    audios.value.push(recording)

@@ -365,10 +408,10 @@ onUnmounted(() => {
        </Button>

        <div>
-          <div v-for="(audio, index) in audioURLs" :key="index" class="mb-2">
-            <audio :src="audio" controls class="w-full" />
-            <div v-if="transcriptions[index]" class="mt-2 text-sm text-neutral-500 dark:text-neutral-400">
-              {{ transcriptions[index] }}
+          <div v-for="(transcription, index) in transcriptions" :key="index" class="mb-2">
+            <audio v-if="audioURLs[index]" :src="audioURLs[index]" controls class="w-full" />
+            <div v-if="transcription" class="mt-2 text-sm text-neutral-500 dark:text-neutral-400">
+              {{ transcription }}
            </div>
          </div>
        </div>
@@ -5,7 +5,14 @@ export function useAudioDevice() {
  const devices = useDevicesList({ constraints: { audio: true }, requestPermissions: true })
  const audioInputs = computed(() => devices.audioInputs.value)
  const selectedAudioInput = ref<string>(devices.audioInputs.value[0]?.deviceId || '')
-  const deviceConstraints = computed<MediaStreamConstraints>(() => ({ audio: { deviceId: { exact: selectedAudioInput.value }, autoGainControl: true, echoCancellation: true, noiseSuppression: true } }))
+  const deviceConstraints = computed<MediaStreamConstraints>(() => ({
+    audio: {
+      ...(selectedAudioInput.value ? { deviceId: { exact: selectedAudioInput.value } } : {}),
+      autoGainControl: true,
+      echoCancellation: true,
+      noiseSuppression: true,
+    },
+  }))
  const { stream, stop: stopStream, start: startStream } = useUserMedia({ constraints: deviceConstraints, enabled: false, autoSwitch: true })

  watch(audioInputs, () => {
@@ -2,9 +2,12 @@ import type { TranscriptionProviderWithExtraOptions } from '@xsai-ext/shared-pro
 import type { WithUnknown } from '@xsai/shared'
 import type { StreamTranscriptionResult, StreamTranscriptionOptions as XSAIStreamTranscriptionOptions } from '@xsai/stream-transcription'

+import { tryCatch } from '@moeru/std'
 import { generateTranscription } from '@xsai/generate-transcription'
 import { defineStore, storeToRefs } from 'pinia'
-import { computed, ref } from 'vue'
+import { computed, ref, shallowRef } from 'vue'
+
+import vadWorkletUrl from '../../workers/vad/process.worklet?worker&url'

 import { createResettableLocalStorage, createResettableRef } from '../../utils/resettable'
 import { useProvidersStore } from '../providers'
@@ -195,6 +198,243 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
  const hearingStore = useHearingStore()
  const { activeTranscriptionProvider, activeTranscriptionModel } = storeToRefs(hearingStore)
  const providersStore = useProvidersStore()
+  const streamingSession = shallowRef<{
+    audioContext: AudioContext
+    workletNode: AudioWorkletNode
+    mediaStreamSource: MediaStreamAudioSourceNode
+    audioStreamController?: ReadableStreamDefaultController<ArrayBuffer>
+    abortController: AbortController
+    result?: HearingTranscriptionResult
+    idleTimer?: ReturnType<typeof setTimeout>
+    providerId?: string
+  }>()
+
+  const supportsStreamInput = computed(() => {
+    return providersStore.getTranscriptionFeatures(activeTranscriptionProvider.value).supportsStreamInput
+  })
+
+  const DEFAULT_SAMPLE_RATE = 16000
+  const DEFAULT_STREAM_IDLE_TIMEOUT = 15000
+
+  function float32ToInt16(buffer: Float32Array) {
+    const output = new Int16Array(buffer.length)
+    for (let i = 0; i < buffer.length; i++) {
+      const value = Math.max(-1, Math.min(1, buffer[i]))
+      output[i] = value < 0 ? value * 0x8000 : value * 0x7FFF
+    }
+
+    return output
+  }
+
+  async function createAudioStreamFromMediaStream(stream: MediaStream, sampleRate = DEFAULT_SAMPLE_RATE, onActivity?: () => void) {
+    const audioContext = new AudioContext({ sampleRate, latencyHint: 'interactive' })
+    await audioContext.audioWorklet.addModule(vadWorkletUrl)
+    const workletNode = new AudioWorkletNode(audioContext, 'vad-audio-worklet-processor')
+
+    let audioStreamController: ReadableStreamDefaultController<ArrayBuffer> | undefined
+    const audioStream = new ReadableStream<ArrayBuffer>({
+      start(controller) {
+        audioStreamController = controller
+      },
+      cancel: () => {
+        audioStreamController = undefined
+      },
+    })
+
+    workletNode.port.onmessage = ({ data }: MessageEvent<{ buffer?: Float32Array }>) => {
+      const buffer = data?.buffer
+      if (!buffer || !audioStreamController)
+        return
+
+      const pcm16 = float32ToInt16(buffer)
+      // Clone buffer to avoid retaining underlying ArrayBuffer references
+      audioStreamController.enqueue(pcm16.buffer.slice(0))
+      onActivity?.()
+    }
+
+    const mediaStreamSource = audioContext.createMediaStreamSource(stream)
+    mediaStreamSource.connect(workletNode)
+
+    // Sink to avoid feedback/echo
+    const silentGain = audioContext.createGain()
+    silentGain.gain.value = 0
+    workletNode.connect(silentGain)
+    silentGain.connect(audioContext.destination)
+
+    return {
+      audioContext,
+      workletNode,
+      mediaStreamSource,
+      audioStream,
+      get controller() {
+        return audioStreamController
+      },
+    }
+  }
+
+  async function stopStreamingTranscription(abort?: boolean, disposeProviderId?: string) {
+    const session = streamingSession.value
+    if (!session)
+      return
+
+    try {
+      const reason = new DOMException(abort ? 'Aborted' : 'Stopped', 'AbortError')
+      // Ensure provider transports (e.g., Aliyun NLS) are signaled to stop over websocket.
+      if (!session.abortController.signal.aborted) {
+        session.abortController.abort(reason)
+      }
+
+      if (abort)
+        session.audioStreamController?.error(reason)
+      else
+        session.audioStreamController?.close()
+    }
+    catch {}
+
+    await tryCatch(() => {
+      session.mediaStreamSource.disconnect()
+      session.workletNode.port.onmessage = null
+      session.workletNode.disconnect()
+    })
+    await tryCatch(() => session.audioContext.close())
+
+    if (session.idleTimer)
+      clearTimeout(session.idleTimer)
+
+    streamingSession.value = undefined
+
+    if (session.result?.mode === 'stream') {
+      try {
+        const text = await session.result.text
+
+        if (disposeProviderId) {
+          await providersStore.disposeProviderInstance(disposeProviderId)
+        }
+
+        return text
+      }
+      catch (err) {
+        error.value = err instanceof Error ? err.message : String(err)
+        console.error('Error generating transcription:', error.value)
+      }
+    }
+
+    const text = session.result?.text
+    if (disposeProviderId)
+      await providersStore.disposeProviderInstance(disposeProviderId)
+
+    return text
+  }
+
+  async function transcribeForMediaStream(stream: MediaStream, options?: {
+    sampleRate?: number
+    providerOptions?: Record<string, unknown>
+    idleTimeoutMs?: number
+    onSentenceEnd?: (delta: string) => void
+    onSpeechEnd?: (text: string) => void
+  }) {
+    if (!supportsStreamInput.value)
+      return
+
+    try {
+      const providerId = activeTranscriptionProvider.value
+      const provider = await providersStore.getProviderInstance<TranscriptionProviderWithExtraOptions<string, any>>(providerId)
+      if (!provider) {
+        throw new Error('Failed to initialize speech provider')
+      }
+
+      const idleTimeout = options?.idleTimeoutMs ?? DEFAULT_STREAM_IDLE_TIMEOUT
+
+      // If a session already exists, just bump the idle timer and reuse the websocket/audio graph.
+      const existingSession = streamingSession.value
+      if (existingSession) {
+        if (existingSession.idleTimer) {
+          clearTimeout(existingSession.idleTimer)
+          existingSession.idleTimer = setTimeout(async () => {
+            await stopStreamingTranscription(false, existingSession.providerId)
+          }, idleTimeout)
+        }
+        return
+      }
+
+      const abortController = new AbortController()
+      let idleTimer: ReturnType<typeof setTimeout> | undefined
+      const bumpIdle = () => {
+        if (idleTimer)
+          clearTimeout(idleTimer)
+        idleTimer = setTimeout(async () => {
+          await stopStreamingTranscription(false, providerId)
+        }, idleTimeout)
+      }
+
+      const session = await createAudioStreamFromMediaStream(
+        stream,
+        options?.sampleRate ?? DEFAULT_SAMPLE_RATE,
+        () => bumpIdle(),
+      )
+
+      if (session.audioContext.state === 'suspended')
+        await session.audioContext.resume()
+
+      bumpIdle()
+
+      const model = activeTranscriptionModel.value
+      const result = await hearingStore.transcription(
+        providerId,
+        provider,
+        model,
+        { inputAudioStream: session.audioStream },
+        undefined,
+        {
+          providerOptions: {
+            abortSignal: abortController.signal,
+            ...options?.providerOptions,
+          },
+        },
+      )
+
+      streamingSession.value = {
+        audioContext: session.audioContext,
+        workletNode: session.workletNode,
+        mediaStreamSource: session.mediaStreamSource,
+        audioStreamController: session.controller,
+        abortController,
+        result,
+        idleTimer,
+        providerId,
+      }
+
+      // Stream out text deltas to caller without tearing down the session.
+      if (result.mode === 'stream' && result.textStream) {
+        void (async () => {
+          let fullText = ''
+          try {
+            const reader = result.textStream.getReader()
+
+            while (true) {
+              const { done, value } = await reader.read()
+              if (done)
+                break
+              if (value) {
+                fullText += value
+                options?.onSentenceEnd?.(value)
+              }
+            }
+          }
+          catch (err) {
+            console.error('Error reading text stream:', err)
+          }
+          finally {
+            options?.onSpeechEnd?.(fullText)
+          }
+        })()
+      }
+    }
+    catch (err) {
+      error.value = err instanceof Error ? err.message : String(err)
+      console.error('Error generating transcription:', error.value)
+    }
+  }

  async function transcribeForRecording(recording: Blob | null | undefined) {
    if (!recording)
@@ -229,5 +469,8 @@ export const useHearingSpeechInputPipeline = defineStore('modules:hearing:speech
    error,

    transcribeForRecording,
+    transcribeForMediaStream,
+    stopStreamingTranscription,
+    supportsStreamInput,
  }
 })
@@ -218,6 +218,7 @@ function createAnthropic(apiKey: string, baseURL: string = 'https://api.anthropi
 export const useProvidersStore = defineStore('providers', () => {
  const providerCredentials = useLocalStorage<Record<string, Record<string, unknown>>>('settings/credentials/providers', {})
  const addedProviders = useLocalStorage<Record<string, boolean>>('settings/providers/added', {})
+  const providerInstanceCache = ref<Record<string, unknown>>({})
  const { t } = useI18n()
  const baseUrlValidator = computed(() => (baseUrl: unknown) => {
    let msg = ''
@@ -904,9 +905,17 @@ export const useProvidersStore = defineStore('providers', () => {
        const provider = createAliyunNlsStreamProvider(accessKeyId, accessKeySecret, appKey, { region: resolvedRegion })

        return {
-          transcription(model: string, extraOptions?: AliyunRealtimeSpeechExtraOptions) {
-            return provider.speech(model, extraOptions)
-          },
+          transcription: (model: string, extraOptions?: AliyunRealtimeSpeechExtraOptions) => provider.speech(model, {
+            ...extraOptions,
+            sessionOptions: {
+              format: 'pcm',
+              sample_rate: 16000,
+              enable_punctuation_prediction: true,
+              enable_intermediate_result: true,
+              enable_words: true,
+              ...extraOptions?.sessionOptions,
+            },
+          }),
        } as TranscriptionProviderWithExtraOptions<string, AliyunRealtimeSpeechExtraOptions>
      },
      capabilities: {
@@ -1875,17 +1884,18 @@ export const useProvidersStore = defineStore('providers', () => {
    },
  }

+  // const validatedCredentials = ref<Record<string, string>>({})
+  const providerRuntimeState = ref<Record<string, ProviderRuntimeState>>({})
+
  const configuredProviders = computed(() => {
    const result: Record<string, boolean> = {}
    for (const [key, state] of Object.entries(providerRuntimeState.value)) {
      result[key] = state.isConfigured
    }
+
    return result
  })

-  // const validatedCredentials = ref<Record<string, string>>({})
-  const providerRuntimeState = ref<Record<string, ProviderRuntimeState>>({})
-
  function markProviderAdded(providerId: string) {
    addedProviders.value[providerId] = true
  }
@@ -2107,6 +2117,9 @@ export const useProvidersStore = defineStore('providers', () => {
    )

    for (const providerId of changedProviders) {
+      // Since credentials changed, dispose the cached instance so new creds take effect.
+      void disposeProviderInstance(providerId)
+
      // If the provider is configured and has the capability, refetch its models
      if (providerRuntimeState.value[providerId]?.isConfigured && providerMetadata[providerId]?.capabilities.listModels) {
        fetchModelsForProvider(providerId)
@@ -2160,6 +2173,10 @@ export const useProvidersStore = defineStore('providers', () => {
  | TranscriptionProvider
  | TranscriptionProviderWithExtraOptions,
  >(providerId: string): Promise<R> {
+    const cached = providerInstanceCache.value[providerId] as R | undefined
+    if (cached)
+      return cached
+
    const config = providerCredentials.value[providerId]
    if (!config)
      throw new Error(`Provider credentials for ${providerId} not found`)
@@ -2169,7 +2186,9 @@ export const useProvidersStore = defineStore('providers', () => {
      throw new Error(`Provider metadata for ${providerId} not found`)

    try {
-      return await metadata.createProvider(config) as R
+      const instance = await metadata.createProvider(config) as R
+      providerInstanceCache.value[providerId] = instance
+      return instance
    }
    catch (error) {
      console.error(`Error creating provider instance for ${providerId}:`, error)
@@ -2177,6 +2196,14 @@ export const useProvidersStore = defineStore('providers', () => {
    }
  }

+  async function disposeProviderInstance(providerId: string) {
+    const instance = providerInstanceCache.value[providerId] as { dispose?: () => Promise<void> | void } | undefined
+    if (instance?.dispose)
+      await instance.dispose()
+
+    delete providerInstanceCache.value[providerId]
+  }
+
  const availableProvidersMetadata = computedAsync<ProviderMetadata[]>(async () => {
    const providers: ProviderMetadata[] = []

@@ -2273,6 +2300,7 @@ export const useProvidersStore = defineStore('providers', () => {
    allAvailableModels,
    loadModelsForConfiguredProviders,
    getProviderInstance,
+    disposeProviderInstance,
    resetProviderSettings,
    forceProviderConfigured,
    availableProvidersMetadata,
@@ -4,12 +4,61 @@ import type { StreamTranscriptionDelta, StreamTranscriptionResult } from '@xsai/

 import type { EventStartTranscription, ServerEvent, ServerEvents } from './'

+import { tryCatch } from '@moeru/std'
+import { timeout as promiseTimeout } from 'es-toolkit/promise'
+
 import { createAliyunNLSSession } from './'
 import { nlsWebSocketEndpointFromRegion } from './utils'

 type SessionOptions = NonNullable<Parameters<typeof createAliyunNLSSession>[3]>
 type AudioChunk = ArrayBuffer | ArrayBufferView

+function eventListenerOf(type: string, listener: EventListenerOrEventListenerObject, on?: EventTarget, options?: AddEventListenerOptions) {
+  return {
+    on: () => on?.addEventListener(type, listener, options),
+    off: () => on?.removeEventListener(type, listener, options),
+  }
+}
+
+function promiseOfAbortSignal(signal?: AbortSignal) {
+  if (!signal)
+    return null
+  if (signal.aborted)
+    return Promise.reject(signal.reason ?? new DOMException('Aborted', 'AbortError'))
+
+  return new Promise<never>((_, reject) => {
+    const handler = () => {
+      signal.removeEventListener('abort', handler)
+      reject(signal.reason ?? new DOMException('Aborted', 'AbortError'))
+    }
+
+    signal.addEventListener('abort', handler, { once: true })
+  })
+}
+
+function createWaiter(timeoutMs: number, abortSignal?: AbortSignal) {
+  let resolve!: () => void
+  let reject!: (reason?: unknown) => void
+  const deferred = new Promise<void>((res, rej) => {
+    resolve = res
+    reject = rej
+  })
+
+  function wait() {
+    return Promise.race([
+      deferred,
+      timeoutMs > 0 ? promiseTimeout(timeoutMs) : deferred,
+      abortSignal ? promiseOfAbortSignal(abortSignal) : deferred,
+    ]) as Promise<void>
+  }
+
+  return {
+    wait,
+    trigger: () => resolve?.(),
+    cancel: (reason?: unknown) => reject?.(reason),
+  }
+}
+
 const DEFAULT_SESSION_OPTIONS: Pick<EventStartTranscription['payload'], 'format' | 'sample_rate'> = {
  format: 'pcm',
  sample_rate: 16000,
@@ -133,22 +182,8 @@ function resolveAudioStream(options: AliyunStreamTranscriptionOptions): Readable

 interface InternalRealtimeOptions extends CreateAliyunStreamTranscriptionOptions {
  onSentenceFinal?: (payload: ServerEvents['SentenceEnd']) => Promise<void> | void
-}
-
-function mayThrow(fn: () => void | Promise<void>) {
-  try {
-    return fn()
-  }
-  catch {
-    return undefined
-  }
-}
-
-function eventListenerOf(type: string, listener: EventListenerOrEventListenerObject, on?: EventTarget) {
-  return {
-    on: () => on?.addEventListener(type, listener),
-    off: () => on?.removeEventListener(type, listener),
-  }
+  idleTimeoutMs?: number
+  stopAckTimeoutMs?: number
 }

 async function startRealtimeSession(options: InternalRealtimeOptions): Promise<AliyunStreamTranscriptionHandle> {
@@ -163,32 +198,70 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
    hooks,
    onSessionTerminated,
    onSentenceFinal,
+    idleTimeoutMs = 8000,
+    stopAckTimeoutMs = 2000,
  } = options

  const session = createAliyunNLSSession(accessKeyId, accessKeySecret, appKey, { region })
  const reader = audioStream.getReader()
  const url = await session.websocketUrl()

-  mayThrow(() => hooks?.onWebSocketConnecting?.())
+  await tryCatch(() => hooks?.onWebSocketConnecting?.())

  const websocket = new WebSocket(url)
  websocket.binaryType = 'arraybuffer'

  const abortHandler = abortSignal
-    ? eventListenerOf('abort', () => cleanup(abortSignal?.reason ?? new DOMException('Aborted', 'AbortError')), abortSignal)
+    ? eventListenerOf('abort', () => cleanup(abortSignal.reason ?? new DOMException('Aborted', 'AbortError')), abortSignal, { once: true })
    : undefined
+
  abortHandler?.on()

+  const stopWaiter = createWaiter(stopAckTimeoutMs, abortSignal)
+  let stopping = false
+
+  async function requestStop(reason?: unknown) {
+    if (stopping)
+      return
+    stopping = true
+    try {
+      if (websocket?.readyState === WebSocket.OPEN)
+        await tryCatch(() => session.stop(websocket))
+
+      await Promise.race([
+        stopWaiter.wait(),
+        new Promise(resolve => setTimeout(resolve, stopAckTimeoutMs)),
+      ])
+    }
+    catch (error) {
+      await cleanup(error, { sendStop: false })
+      return
+    }
+
+    await cleanup(reason, { sendStop: false })
+  }
+
+  let idleTimer: ReturnType<typeof setTimeout> | undefined
+  const bumpIdle = () => {
+    if (idleTimer)
+      clearTimeout(idleTimer)
+    idleTimer = setTimeout(() => {
+      void requestStop(new DOMException('Idle timeout', 'AbortError'))
+    }, idleTimeoutMs)
+  }
+
+  bumpIdle()
+
  async function cleanup(error?: unknown, options?: { sendStop?: boolean, closeSocket?: boolean }) {
    const { sendStop = true, closeSocket = true } = options ?? {}
    abortHandler?.off()
-    mayThrow(async () => await reader.cancel())
+    await tryCatch(async () => await reader.cancel())

    if (websocket && closeSocket) {
      switch (websocket.readyState) {
        case WebSocket.OPEN:
          if (sendStop)
-            mayThrow(() => session.stop(websocket))
+            await tryCatch(() => session.stop(websocket))
          websocket.close(1000, 'client closed')
          break
        case WebSocket.CONNECTING:
@@ -215,13 +288,16 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
        }

        const { done, value } = await reader.read()
-
        if (done)
          break
-
        if (value)
          websocket!.send(toArrayBuffer(value))
+
+        bumpIdle()
      }
+
+      // Allow a grace period for server to flush final events before stop.
+      bumpIdle()
    }
    catch (error) {
      await cleanup(error)
@@ -231,7 +307,9 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
  async function onMessage(message: MessageEvent) {
    const data = JSON.parse(message.data)
    session.onEvent(data, async (event: ServerEvent) => {
-      mayThrow(async () => await hooks?.onServerEvent?.(event))
+      await tryCatch(async () => await hooks?.onServerEvent?.(event))
+
+      bumpIdle()

      try {
        switch (event.header.name) {
@@ -242,6 +320,7 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
            await onSentenceFinal?.(event.payload as ServerEvents['SentenceEnd'])
            break
          case 'TranscriptionCompleted':
+            stopWaiter.trigger()
            await cleanup(undefined, { sendStop: false, closeSocket: false })
            break
          default:
@@ -255,7 +334,7 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
  }

  async function onOpen() {
-    mayThrow(() => hooks?.onWebSocketOpen?.())
+    await tryCatch(() => hooks?.onWebSocketOpen?.())

    session.start(websocket!, {
      enable_intermediate_result: true,
@@ -265,10 +344,13 @@ async function startRealtimeSession(options: InternalRealtimeOptions): Promise<A
    })
  }

-  websocket.onerror = event => mayThrow(() => hooks?.onWebSocketError?.(event))
-  websocket.onclose = close => mayThrow(() => hooks?.onWebSocketClose?.(close?.code ?? 1006, close?.reason ?? ''))
-  websocket.onopen = () => mayThrow(async () => onOpen())
-  websocket.onmessage = event => mayThrow(async () => onMessage(event))
+  websocket.onerror = event => tryCatch(() => hooks?.onWebSocketError?.(event))
+  websocket.onclose = (close) => {
+    stopWaiter.trigger()
+    return tryCatch(() => hooks?.onWebSocketClose?.(close?.code ?? 1006, close?.reason ?? ''))
+  }
+  websocket.onopen = () => tryCatch(async () => onOpen())
+  websocket.onmessage = event => tryCatch(async () => onMessage(event))

  if (abortSignal?.aborted)
    throw abortSignal.reason ?? new DOMException('Aborted', 'AbortError')
@@ -364,7 +446,7 @@ export function createAliyunNLSProvider(
  options?: {
    region?: SessionOptions['region']
  },
-): SpeechProviderWithExtraOptions<string, AliyunRealtimeSpeechExtraOptions> {
+): SpeechProviderWithExtraOptions<string, AliyunRealtimeSpeechExtraOptions> & { dispose: () => Promise<void> } {
  return {
    speech(_, extraOptions) {
      return {
@@ -393,6 +475,7 @@ export function createAliyunNLSProvider(
                  controllerClosed = true
                  try {
                    await extraOptions?.onSessionTerminated?.(error)
+                    controller.enqueue(encodeSSE({ delta: '', type: 'transcript.text.done' }))
                  }
                  finally {
                    if (error)
@@ -405,6 +488,7 @@ export function createAliyunNLSProvider(
                  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' }))
                },
              }).then((handle) => {
@@ -434,5 +518,9 @@ export function createAliyunNLSProvider(
        },
      }
    },
+    // Allow external caches to dispose provider instances; no persistent resources to release here.
+    async dispose() {
+
+    },
  }
 }
@@ -237,7 +237,10 @@ export const useSettingsAudioDevice = defineStore('settings-audio-devices', () =
  })

  onMounted(() => {
-    if (selectedAudioInputEnabledPersist.value && selectedAudioInputPersist.value) {
+    const hasSelectedInput = selectedAudioInputPersist.value
+      && audioInputs.value.some(device => device.deviceId === selectedAudioInputPersist.value)
+
+    if (selectedAudioInputEnabledPersist.value && hasSelectedInput) {
      startStream()
    }
    if (selectedAudioInputNonPersist.value && !selectedAudioInputEnabledPersist.value) {
@@ -1,2 +1,2 @@
-export type { VADAudioOptions } from '@proj-airi/stage-ui/libs/audio/vad'
-export { createVADStates } from '@proj-airi/stage-ui/libs/audio/vad'
+export type { VADAudioOptions } from '../../libs/audio/vad'
+export { createVADStates } from '../../libs/audio/vad'
@@ -48,12 +48,10 @@ export class VAD implements BaseVAD {
    try {
      this.emit('status', { type: 'info', message: 'Loading VAD model...' })

-      this.model = await AutoModel.from_pretrained('onnx-community/silero-vad', {
-        config: { model_type: 'custom' } as any,
-        dtype: 'fp32', // Full-precision
-      })
-
+      // Full-precision
+      this.model = await AutoModel.from_pretrained('onnx-community/silero-vad', { config: { model_type: 'custom' } as any, dtype: 'fp32' })
      this.isReady = true
+
      this.emit('status', { type: 'info', message: 'VAD model loaded successfully' })
    }
    catch (error) {
@@ -118,7 +116,9 @@ export class VAD implements BaseVAD {
      if (this.prevBuffers.length >= maxPrevBuffers) {
        this.prevBuffers.shift()
      }
+
      this.prevBuffers.push(inputBuffer.slice(0))
+
      return
    }

@@ -132,6 +132,7 @@ export class VAD implements BaseVAD {
      // Process and reset with overflow
      const overflow = inputBuffer.subarray(remaining)
      this.processSpeechSegment(overflow)
+
      return
    }
    else {
@@ -151,6 +152,7 @@ export class VAD implements BaseVAD {
      // Update state
      this.isRecording = true
      this.postSpeechSamples = 0
+
      return
    }

@@ -163,6 +165,7 @@ export class VAD implements BaseVAD {
      if (this.bufferPointer < minSpeechDurationSamples) {
        // Too short, reset without processing
        this.reset()
+
        return
      }

@@ -5,6 +5,7 @@ export default defineConfig({
    projects: [
      'packages/stage-ui',
      'packages/vite-plugin-warpdrive',
+      'packages/audio-pipelines-transcribe',
    ],
  },
 })