feat(voice): desktop speech-stream sessions with barge-in capture
/api/audio/speak-stream WebSocket: one socket + one Web Audio clock per reply. The renderer feeds raw LLM deltas as they arrive; the server cuts sentences with the shared chunker and streams int16 PCM back while generation continues — speech overlaps generation with no per-sentence connection or synthesis gaps. Falls back to the POST data-URL path for old backends / non-chunked providers. Barge-in runs a MediaRecorder on the monitor's stream the whole time playback is live (rotated while quiet to bound pre-roll); talking over the agent cuts playback and the complete utterance goes straight to transcription and submit. /api/audio/transcribe returns 200/"" for no-speech results so quiet turns re-listen instead of toasting a 400.
This commit is contained in:
Brooklyn Nicholson
parent
68e1fedd2d
commit
93e9061f15
|
|
@ -1,7 +1,13 @@
|
|||
import { useCallback, useEffect, useRef, useState } from 'react'
|
||||
|
||||
import { useI18n } from '@/i18n'
|
||||
import { playSpeechText, stopVoicePlayback } from '@/lib/voice-playback'
|
||||
import { monitorSpeechDuringPlayback } from '@/lib/voice-barge-in'
|
||||
import {
|
||||
playSpeechText,
|
||||
type SpeechStreamSession,
|
||||
startSpeechStream,
|
||||
stopVoicePlayback
|
||||
} from '@/lib/voice-playback'
|
||||
import { notify, notifyError } from '@/store/notifications'
|
||||
|
||||
import { useMicRecorder } from './use-mic-recorder'
|
||||
|
|
@ -44,7 +50,9 @@ export function useVoiceConversation({
|
|||
const awaitingSpokenResponseRef = useRef(false)
|
||||
const responseIdRef = useRef<string | null>(null)
|
||||
const spokenSourceLengthRef = useRef(0)
|
||||
const speechBufferRef = useRef('')
|
||||
const speechSessionRef = useRef<null | SpeechStreamSession>(null)
|
||||
const stopBargeMonitorRef = useRef<(() => void) | null>(null)
|
||||
const bargeCapturePendingRef = useRef(false)
|
||||
const enabledRef = useRef(enabled)
|
||||
const mutedRef = useRef(muted)
|
||||
const busyRef = useRef(busy)
|
||||
|
|
@ -74,60 +82,13 @@ export function useVoiceConversation({
|
|||
}
|
||||
}
|
||||
|
||||
const resetSpeechBuffer = () => {
|
||||
const dropSpeechSession = () => {
|
||||
stopBargeMonitorRef.current?.()
|
||||
stopBargeMonitorRef.current = null
|
||||
bargeCapturePendingRef.current = false
|
||||
speechSessionRef.current = null
|
||||
responseIdRef.current = null
|
||||
spokenSourceLengthRef.current = 0
|
||||
speechBufferRef.current = ''
|
||||
}
|
||||
|
||||
const appendSpeechText = (text: string) => {
|
||||
if (!text) {
|
||||
return
|
||||
}
|
||||
|
||||
speechBufferRef.current = `${speechBufferRef.current}${text}`
|
||||
}
|
||||
|
||||
const takeSpeechChunk = (force = false): string | null => {
|
||||
const buffer = speechBufferRef.current.replace(/\s+/g, ' ').trim()
|
||||
|
||||
if (!buffer) {
|
||||
speechBufferRef.current = ''
|
||||
|
||||
return null
|
||||
}
|
||||
|
||||
const sentence = buffer.match(/^(.+?[.!?。！？])(?:\s+|$)/)
|
||||
|
||||
if (sentence?.[1] && (sentence[1].length >= 8 || force)) {
|
||||
const chunk = sentence[1].trim()
|
||||
speechBufferRef.current = buffer.slice(sentence[1].length).trim()
|
||||
|
||||
return chunk
|
||||
}
|
||||
|
||||
if (!force && buffer.length > 220) {
|
||||
const softBoundary = Math.max(
|
||||
buffer.lastIndexOf(', ', 180),
|
||||
buffer.lastIndexOf('; ', 180),
|
||||
buffer.lastIndexOf(': ', 180)
|
||||
)
|
||||
|
||||
if (softBoundary > 80) {
|
||||
const chunk = buffer.slice(0, softBoundary + 1).trim()
|
||||
speechBufferRef.current = buffer.slice(softBoundary + 1).trim()
|
||||
|
||||
return chunk
|
||||
}
|
||||
}
|
||||
|
||||
if (!force) {
|
||||
return null
|
||||
}
|
||||
|
||||
speechBufferRef.current = ''
|
||||
|
||||
return buffer
|
||||
}
|
||||
|
||||
const handleTurn = useCallback(
|
||||
|
|
@ -167,7 +128,7 @@ export function useVoiceConversation({
|
|||
}
|
||||
|
||||
awaitingSpokenResponseRef.current = true
|
||||
resetSpeechBuffer()
|
||||
dropSpeechSession()
|
||||
await onSubmit(transcript)
|
||||
setStatus('thinking')
|
||||
} catch (error) {
|
||||
|
|
@ -193,6 +154,10 @@ export function useVoiceConversation({
|
|||
return
|
||||
}
|
||||
|
||||
if (bargeCapturePendingRef.current) {
|
||||
return // the barge monitor is mid-capture and owns the mic
|
||||
}
|
||||
|
||||
if (statusRef.current !== 'idle') {
|
||||
return
|
||||
}
|
||||
|
|
@ -220,24 +185,237 @@ export function useVoiceConversation({
|
|||
}
|
||||
}, [handle, handleTurn, onFatalError, voiceCopy.couldNotStartSession, voiceCopy.microphoneFailed])
|
||||
|
||||
const speak = useCallback(
|
||||
async (text: string) => {
|
||||
setStatus('speaking')
|
||||
const settleAfterSpeech = useCallback(
|
||||
(barged: boolean) => {
|
||||
if (barged || !awaitingSpokenResponseRef.current) {
|
||||
awaitingSpokenResponseRef.current = false
|
||||
consumePendingResponse()
|
||||
}
|
||||
|
||||
if (bargeCapturePendingRef.current) {
|
||||
// The barge monitor is still capturing the user's interruption — it
|
||||
// owns the next turn. Keep it alive and don't re-open the mic; the
|
||||
// utterance callback transcribes and submits when they go quiet.
|
||||
speechSessionRef.current = null
|
||||
responseIdRef.current = null
|
||||
spokenSourceLengthRef.current = 0
|
||||
setStatus('listening')
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
dropSpeechSession()
|
||||
|
||||
if (enabledRef.current) {
|
||||
pendingStartRef.current = true
|
||||
}
|
||||
|
||||
setStatus('idle')
|
||||
},
|
||||
[consumePendingResponse]
|
||||
)
|
||||
|
||||
/**
|
||||
* Submit the utterance the barge monitor captured — the user's interruption
|
||||
* from its first syllable, no re-listen round trip. Empty/failed captures
|
||||
* fall back to normal listening.
|
||||
*/
|
||||
const submitCapturedUtterance = useCallback(
|
||||
async (audio: Blob | null) => {
|
||||
const resumeListening = () => {
|
||||
if (enabledRef.current && !mutedRef.current) {
|
||||
pendingStartRef.current = true
|
||||
}
|
||||
|
||||
setStatus('idle')
|
||||
}
|
||||
|
||||
if (!audio || !onTranscribeAudio) {
|
||||
resumeListening()
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
setStatus('transcribing')
|
||||
|
||||
try {
|
||||
await playSpeechText(text, { source: 'voice-conversation' })
|
||||
} catch (error) {
|
||||
notifyError(error, voiceCopy.playbackFailed)
|
||||
} finally {
|
||||
if (enabledRef.current) {
|
||||
pendingStartRef.current = true
|
||||
setStatus('idle')
|
||||
} else {
|
||||
setStatus('idle')
|
||||
const transcript = (await onTranscribeAudio(audio)).trim()
|
||||
|
||||
if (!transcript) {
|
||||
resumeListening()
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
awaitingSpokenResponseRef.current = true
|
||||
dropSpeechSession()
|
||||
consumePendingResponse()
|
||||
await onSubmit(transcript)
|
||||
setStatus('thinking')
|
||||
} catch (error) {
|
||||
notifyError(error, voiceCopy.transcriptionFailed)
|
||||
resumeListening()
|
||||
}
|
||||
},
|
||||
[voiceCopy.playbackFailed]
|
||||
[consumePendingResponse, onSubmit, onTranscribeAudio, voiceCopy.transcriptionFailed]
|
||||
)
|
||||
|
||||
/** Barge-in monitor wiring shared by the live and fallback speech paths. */
|
||||
const openBargeMonitor = useCallback(
|
||||
(onBarge: () => void) =>
|
||||
monitorSpeechDuringPlayback({
|
||||
onSpeech: () => {
|
||||
bargeCapturePendingRef.current = true
|
||||
onBarge()
|
||||
stopVoicePlayback()
|
||||
},
|
||||
onUtterance: audio => {
|
||||
bargeCapturePendingRef.current = false
|
||||
stopBargeMonitorRef.current = null
|
||||
void submitCapturedUtterance(audio)
|
||||
}
|
||||
}),
|
||||
[submitCapturedUtterance]
|
||||
)
|
||||
|
||||
/** Push any new reply text into the live session; finish when complete. */
|
||||
const feedSpeechSession = useCallback(
|
||||
(responseId: string) => {
|
||||
const session = speechSessionRef.current
|
||||
|
||||
if (!session || responseIdRef.current !== responseId) {
|
||||
return
|
||||
}
|
||||
|
||||
const response = pendingResponse()
|
||||
|
||||
if (response && response.id === responseId) {
|
||||
if (response.text.length > spokenSourceLengthRef.current) {
|
||||
session.append(response.text.slice(spokenSourceLengthRef.current))
|
||||
spokenSourceLengthRef.current = response.text.length
|
||||
}
|
||||
|
||||
if (!response.pending && !busyRef.current) {
|
||||
session.finish()
|
||||
}
|
||||
} else if (!busyRef.current) {
|
||||
// Reply consumed/vanished while we were speaking — close out the turn.
|
||||
session.finish()
|
||||
}
|
||||
},
|
||||
[pendingResponse]
|
||||
)
|
||||
|
||||
/** Whole-text fallback: wait for the reply to complete, then speak it. */
|
||||
const awaitFallbackSpeech = useCallback(
|
||||
(responseId: string) => {
|
||||
const poll = () => {
|
||||
if (responseIdRef.current !== responseId) {
|
||||
return
|
||||
}
|
||||
|
||||
const response = pendingResponse()
|
||||
|
||||
if (!response || response.id !== responseId) {
|
||||
settleAfterSpeech(false)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if (response.pending || busyRef.current) {
|
||||
window.setTimeout(poll, 250)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
let barged = false
|
||||
|
||||
stopBargeMonitorRef.current?.()
|
||||
stopBargeMonitorRef.current = openBargeMonitor(() => {
|
||||
barged = true
|
||||
})
|
||||
|
||||
void playSpeechText(response.text, { source: 'voice-conversation' })
|
||||
.catch(error => notifyError(error, voiceCopy.playbackFailed))
|
||||
.finally(() => {
|
||||
if (responseIdRef.current === responseId) {
|
||||
awaitingSpokenResponseRef.current = false
|
||||
settleAfterSpeech(barged)
|
||||
}
|
||||
})
|
||||
}
|
||||
|
||||
poll()
|
||||
},
|
||||
[openBargeMonitor, pendingResponse, settleAfterSpeech, voiceCopy.playbackFailed]
|
||||
)
|
||||
|
||||
/**
|
||||
* Live-speak the streaming reply: one speech session per response, fed
|
||||
* incremental text as the assistant generates it. Audio overlaps generation
|
||||
* — no wait for the full reply, no per-sentence gaps.
|
||||
*/
|
||||
const openLiveSpeech = useCallback(
|
||||
(responseId: string) => {
|
||||
responseIdRef.current = responseId
|
||||
spokenSourceLengthRef.current = 0
|
||||
setStatus('speaking')
|
||||
|
||||
let barged = false
|
||||
|
||||
// VAD barge-in: the user talking over the reply cuts playback, drops
|
||||
// the not-yet-spoken remainder, AND keeps capturing — the interruption
|
||||
// is transcribed from its first syllable instead of losing the opening
|
||||
// words to a mic re-open.
|
||||
stopBargeMonitorRef.current = openBargeMonitor(() => {
|
||||
barged = true
|
||||
})
|
||||
|
||||
void (async () => {
|
||||
const session = await startSpeechStream({ source: 'voice-conversation' })
|
||||
|
||||
// The session may resolve after the loop moved on (barge, disable).
|
||||
if (responseIdRef.current !== responseId) {
|
||||
if (session) {
|
||||
stopVoicePlayback()
|
||||
}
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if (!session) {
|
||||
// No streaming backend/provider: speak the whole reply once it lands.
|
||||
speechSessionRef.current = null
|
||||
awaitFallbackSpeech(responseId)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
speechSessionRef.current = session
|
||||
|
||||
// Timer-driven feed: reply text flows into the session at delta rate
|
||||
// regardless of React render cadence.
|
||||
const feedTimer = window.setInterval(() => feedSpeechSession(responseId), 150)
|
||||
feedSpeechSession(responseId)
|
||||
|
||||
const outcome = await session.done
|
||||
window.clearInterval(feedTimer)
|
||||
|
||||
if (responseIdRef.current !== responseId) {
|
||||
return
|
||||
}
|
||||
|
||||
if (outcome === 'fallback') {
|
||||
awaitFallbackSpeech(responseId)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
awaitingSpokenResponseRef.current = false
|
||||
settleAfterSpeech(barged)
|
||||
})()
|
||||
},
|
||||
[awaitFallbackSpeech, feedSpeechSession, openBargeMonitor, settleAfterSpeech]
|
||||
)
|
||||
|
||||
const start = useCallback(async () => {
|
||||
|
|
@ -254,7 +432,7 @@ export function useVoiceConversation({
|
|||
|
||||
setMuted(false)
|
||||
awaitingSpokenResponseRef.current = false
|
||||
resetSpeechBuffer()
|
||||
dropSpeechSession()
|
||||
consumePendingResponse()
|
||||
pendingStartRef.current = true
|
||||
await startListening()
|
||||
|
|
@ -274,7 +452,7 @@ export function useVoiceConversation({
|
|||
handle.cancel()
|
||||
turnClosingRef.current = false
|
||||
awaitingSpokenResponseRef.current = false
|
||||
resetSpeechBuffer()
|
||||
dropSpeechSession()
|
||||
consumePendingResponse()
|
||||
setMuted(false)
|
||||
setStatus('idle')
|
||||
|
|
@ -325,8 +503,9 @@ export function useVoiceConversation({
|
|||
return () => window.removeEventListener('keydown', onKeyDown, { capture: true })
|
||||
}, [enabled, stopTurn])
|
||||
|
||||
// Drive the loop: after a voice-submitted turn, speak stable chunks as the
|
||||
// assistant stream grows. Otherwise start listening when idle between turns.
|
||||
// Drive the loop: when a voice-submitted reply appears, open a live speech
|
||||
// session (which feeds itself from then on). Otherwise start listening when
|
||||
// idle between turns.
|
||||
useEffect(() => {
|
||||
if (!enabled || muted) {
|
||||
return
|
||||
|
|
@ -336,38 +515,15 @@ export function useVoiceConversation({
|
|||
const response = pendingResponse()
|
||||
|
||||
if (response) {
|
||||
if (response.id !== responseIdRef.current) {
|
||||
resetSpeechBuffer()
|
||||
responseIdRef.current = response.id
|
||||
}
|
||||
openLiveSpeech(response.id)
|
||||
|
||||
if (response.text.length > spokenSourceLengthRef.current) {
|
||||
appendSpeechText(response.text.slice(spokenSourceLengthRef.current))
|
||||
spokenSourceLengthRef.current = response.text.length
|
||||
}
|
||||
|
||||
const chunk = takeSpeechChunk(!response.pending && !busy)
|
||||
|
||||
if (chunk) {
|
||||
void speak(chunk)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
if (!response.pending && !busy) {
|
||||
awaitingSpokenResponseRef.current = false
|
||||
consumePendingResponse()
|
||||
resetSpeechBuffer()
|
||||
pendingStartRef.current = true
|
||||
setStatus('idle')
|
||||
|
||||
return
|
||||
}
|
||||
return
|
||||
}
|
||||
|
||||
if (!busy && status === 'thinking') {
|
||||
// Turn finished without any speakable reply (tool-only, error).
|
||||
awaitingSpokenResponseRef.current = false
|
||||
resetSpeechBuffer()
|
||||
dropSpeechSession()
|
||||
pendingStartRef.current = true
|
||||
setStatus('idle')
|
||||
|
||||
|
|
@ -382,7 +538,7 @@ export function useVoiceConversation({
|
|||
if (pendingStartRef.current) {
|
||||
void startListening()
|
||||
}
|
||||
}, [busy, consumePendingResponse, enabled, muted, pendingResponse, speak, startListening, status])
|
||||
}, [busy, enabled, muted, openLiveSpeech, pendingResponse, startListening, status])
|
||||
|
||||
useEffect(() => {
|
||||
if (enabled && !wasEnabledRef.current) {
|
||||
|
|
|
|||
|
|
@ -6,6 +6,7 @@ import { type ChatMessage, textPart } from '@/lib/chat-messages'
|
|||
import { optimisticAttachmentRef } from '@/lib/chat-runtime'
|
||||
import { sanitizeComposerInput } from '@/lib/composer-input-sanitize'
|
||||
import { setMutableRef } from '@/lib/mutable-ref'
|
||||
import { isVoicePlaybackActive, stopVoicePlayback } from '@/lib/voice-playback'
|
||||
import {
|
||||
$composerAttachments,
|
||||
clearComposerAttachments,
|
||||
|
|
@ -141,6 +142,11 @@ export function useSubmitPrompt(deps: SubmitPromptDeps) {
|
|||
return false
|
||||
}
|
||||
|
||||
// Typing barge-in: a new send silences any in-flight spoken reply.
|
||||
if (isVoicePlaybackActive()) {
|
||||
stopVoicePlayback()
|
||||
}
|
||||
|
||||
// Queue drains carry their source session explicitly. A background drain
|
||||
// must never inherit the currently selected session after the user moves
|
||||
// to another chat.
|
||||
|
|
|
|||
|
|
@ -0,0 +1,238 @@
|
|||
// VAD barge-in: watch the mic while TTS plays, fire the moment the user talks
|
||||
// over it, and CAPTURE what they say. Detection alone loses the first words —
|
||||
// by the time sustained speech trips the trigger and a fresh recorder spins
|
||||
// up, "stop, actually—" has become "actually—". So a MediaRecorder runs on
|
||||
// the monitor's stream the whole time (pre-roll), and once tripped it keeps
|
||||
// rolling until the user goes quiet, delivering the complete utterance.
|
||||
//
|
||||
// Echo cancellation strips the app's own speaker output from the capture, the
|
||||
// noise floor is calibrated while playback is already audible, and the
|
||||
// sustained window filters coughs/thumps — mirrors
|
||||
// tools/voice_mode.listen_for_speech on the Python surfaces.
|
||||
|
||||
const CALIBRATION_MS = 400
|
||||
const SUSTAINED_MS = 300
|
||||
const MIN_TRIGGER_LEVEL = 0.075 // matches the voice loop's silenceLevel
|
||||
const PRE_ROLL_RESTART_MS = 5_000 // cap pre-roll: restart the recorder while quiet
|
||||
const UTTERANCE_SILENCE_MS = 1_250 // matches the voice loop's silenceMs
|
||||
const UTTERANCE_MAX_MS = 30_000
|
||||
|
||||
export interface BargeMonitorCallbacks {
|
||||
/** Sustained speech detected — cut playback now. */
|
||||
onSpeech: () => void
|
||||
/**
|
||||
* The interrupting utterance, complete from its first syllable (pre-roll
|
||||
* included), delivered once the user goes quiet. `null` when capture was
|
||||
* unavailable — fall back to normal listening.
|
||||
*/
|
||||
onUtterance?: (audio: Blob | null) => void
|
||||
}
|
||||
|
||||
export function monitorSpeechDuringPlayback(callbacks: BargeMonitorCallbacks): () => void {
|
||||
let disposed = false
|
||||
let stream: MediaStream | null = null
|
||||
let context: AudioContext | null = null
|
||||
let frame: number | null = null
|
||||
let recorder: MediaRecorder | null = null
|
||||
let chunks: Blob[] = []
|
||||
let mimeType = ''
|
||||
|
||||
const cleanup = () => {
|
||||
disposed = true
|
||||
|
||||
if (frame !== null) {
|
||||
window.cancelAnimationFrame(frame)
|
||||
frame = null
|
||||
}
|
||||
|
||||
if (recorder && recorder.state !== 'inactive') {
|
||||
recorder.ondataavailable = null
|
||||
recorder.onstop = null
|
||||
|
||||
try {
|
||||
recorder.stop()
|
||||
} catch {
|
||||
// already stopped
|
||||
}
|
||||
}
|
||||
|
||||
recorder = null
|
||||
chunks = []
|
||||
void context?.close().catch(() => undefined)
|
||||
context = null
|
||||
stream?.getTracks().forEach(track => track.stop())
|
||||
stream = null
|
||||
}
|
||||
|
||||
const startSegment = () => {
|
||||
if (!stream || typeof MediaRecorder === 'undefined') {
|
||||
return
|
||||
}
|
||||
|
||||
mimeType =
|
||||
['audio/webm;codecs=opus', 'audio/webm', 'audio/mp4', 'audio/ogg;codecs=opus'].find(type =>
|
||||
MediaRecorder.isTypeSupported(type)
|
||||
) ?? ''
|
||||
|
||||
try {
|
||||
recorder = new MediaRecorder(stream, mimeType ? { mimeType } : undefined)
|
||||
} catch {
|
||||
recorder = null
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
chunks = []
|
||||
|
||||
recorder.ondataavailable = event => {
|
||||
if (event.data.size > 0) {
|
||||
chunks.push(event.data)
|
||||
}
|
||||
}
|
||||
|
||||
recorder.start(250)
|
||||
}
|
||||
|
||||
/** Restart the recorder to drop stale pre-roll — only valid while quiet. */
|
||||
const rotateSegment = () => {
|
||||
if (!recorder || recorder.state === 'inactive') {
|
||||
return
|
||||
}
|
||||
|
||||
recorder.ondataavailable = null
|
||||
recorder.onstop = null
|
||||
|
||||
try {
|
||||
recorder.stop()
|
||||
} catch {
|
||||
// already stopped
|
||||
}
|
||||
|
||||
startSegment()
|
||||
}
|
||||
|
||||
const finishCapture = () => {
|
||||
const active = recorder
|
||||
const type = active?.mimeType || mimeType || 'audio/webm'
|
||||
|
||||
if (!active || active.state === 'inactive') {
|
||||
cleanup()
|
||||
callbacks.onUtterance?.(chunks.length ? new Blob(chunks, { type }) : null)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
active.onstop = () => {
|
||||
const audio = chunks.length ? new Blob(chunks, { type }) : null
|
||||
|
||||
cleanup()
|
||||
callbacks.onUtterance?.(audio)
|
||||
}
|
||||
|
||||
active.stop()
|
||||
}
|
||||
void (async () => {
|
||||
try {
|
||||
stream = await navigator.mediaDevices.getUserMedia({
|
||||
audio: { echoCancellation: true, noiseSuppression: true }
|
||||
})
|
||||
|
||||
if (disposed) {
|
||||
cleanup()
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
startSegment()
|
||||
|
||||
context = new AudioContext()
|
||||
const analyser = context.createAnalyser()
|
||||
analyser.fftSize = 256
|
||||
context.createMediaStreamSource(stream).connect(analyser)
|
||||
|
||||
const data = new Uint8Array(analyser.fftSize)
|
||||
const startedAt = Date.now()
|
||||
const floorSamples: number[] = []
|
||||
let segmentStartedAt = Date.now()
|
||||
let speechStartedAt: number | null = null
|
||||
let tripped = false
|
||||
let trippedAt = 0
|
||||
let quietSince: number | null = null
|
||||
|
||||
const tick = () => {
|
||||
if (disposed) {
|
||||
return
|
||||
}
|
||||
|
||||
analyser.getByteTimeDomainData(data)
|
||||
|
||||
let sum = 0
|
||||
|
||||
for (const value of data) {
|
||||
const centered = value - 128
|
||||
sum += centered * centered
|
||||
}
|
||||
|
||||
const level = Math.min(1, Math.sqrt(sum / data.length) / 42)
|
||||
const now = Date.now()
|
||||
|
||||
if (!tripped && now - startedAt < CALIBRATION_MS) {
|
||||
floorSamples.push(level)
|
||||
} else if (!tripped) {
|
||||
const floor = floorSamples.length ? [...floorSamples].sort((a, b) => a - b)[floorSamples.length >> 1] : 0
|
||||
const trigger = Math.max(MIN_TRIGGER_LEVEL, floor * 3.5)
|
||||
|
||||
if (level >= trigger) {
|
||||
speechStartedAt ??= now
|
||||
|
||||
if (now - speechStartedAt >= SUSTAINED_MS) {
|
||||
tripped = true
|
||||
trippedAt = now
|
||||
quietSince = null
|
||||
callbacks.onSpeech()
|
||||
|
||||
if (!callbacks.onUtterance || !recorder) {
|
||||
cleanup()
|
||||
callbacks.onUtterance?.(null)
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
} else {
|
||||
speechStartedAt = null
|
||||
|
||||
// Bound the pre-roll while quiet so the utterance blob doesn't
|
||||
// accumulate the whole playback (rotating mid-speech would lose
|
||||
// the onset — the whole point).
|
||||
if (now - segmentStartedAt >= PRE_ROLL_RESTART_MS) {
|
||||
rotateSegment()
|
||||
segmentStartedAt = now
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Tripped: keep recording until the user goes quiet (endpoint).
|
||||
// Playback is already stopped, so plain silence-vs-speech works.
|
||||
if (level >= MIN_TRIGGER_LEVEL) {
|
||||
quietSince = null
|
||||
} else {
|
||||
quietSince ??= now
|
||||
}
|
||||
|
||||
if ((quietSince && now - quietSince >= UTTERANCE_SILENCE_MS) || now - trippedAt >= UTTERANCE_MAX_MS) {
|
||||
finishCapture()
|
||||
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
frame = window.requestAnimationFrame(tick)
|
||||
}
|
||||
|
||||
tick()
|
||||
} catch {
|
||||
cleanup()
|
||||
}
|
||||
})()
|
||||
|
||||
return cleanup
|
||||
}
|
||||
|
|
@ -1,3 +1,5 @@
|
|||
import { resolveGatewayWsUrl } from '@hermes/shared'
|
||||
|
||||
import { speakText } from '@/hermes'
|
||||
import {
|
||||
$voicePlayback,
|
||||
|
|
@ -58,6 +60,321 @@ export function stopVoicePlayback() {
|
|||
})
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Streaming path — /api/audio/speak-stream WebSocket, raw int16 PCM frames
|
||||
// scheduled through Web Audio. Speech starts on the provider's first chunk
|
||||
// instead of after full synthesis + base64 transfer.
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
async function resolveSpeakStreamUrl(): Promise<null | string> {
|
||||
const desktop = window.hermesDesktop
|
||||
|
||||
if (!desktop?.getConnection) {
|
||||
return null
|
||||
}
|
||||
|
||||
try {
|
||||
// Mint a fresh credential (single-use ticket in OAuth mode), then swap the
|
||||
// gateway endpoint for the PCM one — auth is shared across WS routes.
|
||||
const wsUrl = await resolveGatewayWsUrl(desktop, await desktop.getConnection())
|
||||
const url = new URL(wsUrl)
|
||||
|
||||
if (!url.pathname.endsWith('/api/ws')) {
|
||||
return null
|
||||
}
|
||||
|
||||
url.pathname = url.pathname.replace(/\/api\/ws$/, '/api/audio/speak-stream')
|
||||
|
||||
return url.toString()
|
||||
} catch {
|
||||
return null
|
||||
}
|
||||
}
|
||||
|
||||
export interface SpeechStreamSession {
|
||||
/** Feed more reply text as it streams in. Safe after `finish` (no-op). */
|
||||
append: (text: string) => void
|
||||
/** No more text coming — resolves `done` once the audio drains. */
|
||||
finish: () => void
|
||||
/**
|
||||
* 'done' — audio fully played (or barged via stopVoicePlayback)
|
||||
* 'fallback'— no audio ever produced; caller should speak the accumulated
|
||||
* text through `playSpeechText` instead.
|
||||
*/
|
||||
done: Promise<'done' | 'fallback'>
|
||||
}
|
||||
|
||||
/**
|
||||
* Open a live speech session: one WebSocket + one AudioContext for a whole
|
||||
* reply. Text is appended as LLM deltas arrive; the server cuts sentences and
|
||||
* streams PCM back while generation continues, so speech overlaps the text
|
||||
* stream (ChatGPT-style) with no per-sentence connection or synthesis gaps.
|
||||
*/
|
||||
function openSpeechStream(wsUrl: string, options: VoicePlaybackOptions): SpeechStreamSession {
|
||||
const ws = new WebSocket(wsUrl)
|
||||
ws.binaryType = 'arraybuffer'
|
||||
|
||||
let context: AudioContext | null = null
|
||||
let streamRate = 24_000
|
||||
let nextStartAt = 0
|
||||
let carry: null | Uint8Array = null
|
||||
let started = false
|
||||
let settled = false
|
||||
let finished = false
|
||||
const pendingSends: string[] = []
|
||||
|
||||
let settle: (value: 'done' | 'fallback') => void = () => undefined
|
||||
|
||||
const done = new Promise<'done' | 'fallback'>(resolve => {
|
||||
settle = value => {
|
||||
if (settled) {
|
||||
return
|
||||
}
|
||||
|
||||
settled = true
|
||||
currentStop = null
|
||||
|
||||
try {
|
||||
ws.close()
|
||||
} catch {
|
||||
// already closed
|
||||
}
|
||||
|
||||
void context?.close().catch(() => undefined)
|
||||
context = null
|
||||
resolve(value)
|
||||
}
|
||||
})
|
||||
|
||||
const send = (frame: object) => {
|
||||
const data = JSON.stringify(frame)
|
||||
|
||||
if (ws.readyState === WebSocket.OPEN) {
|
||||
ws.send(data)
|
||||
} else if (ws.readyState === WebSocket.CONNECTING) {
|
||||
pendingSends.push(data)
|
||||
}
|
||||
}
|
||||
|
||||
// stopVoicePlayback() → immediate barge-in: kill the socket (the server
|
||||
// aborts synthesis on disconnect) and the audio context (cuts sound now).
|
||||
currentStop = () => settle('done')
|
||||
|
||||
const finishWhenDrained = () => {
|
||||
const remainingMs = context ? Math.max(0, nextStartAt - context.currentTime) * 1_000 : 0
|
||||
window.setTimeout(() => settle('done'), remainingMs + 100)
|
||||
}
|
||||
|
||||
const schedule = (data: ArrayBuffer) => {
|
||||
if (!context) {
|
||||
return
|
||||
}
|
||||
|
||||
// Provider chunks are not sample-aligned — carry any odd byte over.
|
||||
let bytes = new Uint8Array(data)
|
||||
|
||||
if (carry) {
|
||||
const joined = new Uint8Array(carry.length + bytes.length)
|
||||
joined.set(carry)
|
||||
joined.set(bytes, carry.length)
|
||||
bytes = joined
|
||||
carry = null
|
||||
}
|
||||
|
||||
const usable = bytes.length - (bytes.length % 2)
|
||||
|
||||
if (bytes.length !== usable) {
|
||||
carry = bytes.slice(usable)
|
||||
}
|
||||
|
||||
if (!usable) {
|
||||
return
|
||||
}
|
||||
|
||||
const pcm = new Int16Array(bytes.buffer, bytes.byteOffset, usable / 2)
|
||||
const buffer = context.createBuffer(1, pcm.length, streamRate)
|
||||
const channel = buffer.getChannelData(0)
|
||||
|
||||
for (let index = 0; index < pcm.length; index += 1) {
|
||||
channel[index] = pcm[index] / 32_768
|
||||
}
|
||||
|
||||
const source = context.createBufferSource()
|
||||
source.buffer = buffer
|
||||
source.connect(context.destination)
|
||||
|
||||
const startAt = Math.max(context.currentTime + 0.05, nextStartAt)
|
||||
source.start(startAt)
|
||||
nextStartAt = startAt + buffer.duration
|
||||
|
||||
if (!started) {
|
||||
started = true
|
||||
setVoicePlaybackState(currentState('speaking', options))
|
||||
}
|
||||
}
|
||||
|
||||
ws.onopen = () => {
|
||||
pendingSends.splice(0).forEach(data => ws.send(data))
|
||||
}
|
||||
|
||||
ws.onmessage = event => {
|
||||
if (typeof event.data !== 'string') {
|
||||
schedule(event.data as ArrayBuffer)
|
||||
|
||||
return
|
||||
}
|
||||
|
||||
let frame: { channels?: number; sample_rate?: number; type?: string }
|
||||
|
||||
try {
|
||||
frame = JSON.parse(event.data) as typeof frame
|
||||
} catch {
|
||||
return
|
||||
}
|
||||
|
||||
if (frame.type === 'start') {
|
||||
streamRate = frame.sample_rate || 24_000
|
||||
context = new AudioContext()
|
||||
nextStartAt = 0
|
||||
} else if (frame.type === 'end') {
|
||||
finishWhenDrained()
|
||||
} else if (frame.type === 'fallback') {
|
||||
settle(started ? 'done' : 'fallback')
|
||||
}
|
||||
}
|
||||
|
||||
// A drop before any audio means the endpoint is unavailable (old backend,
|
||||
// auth, network) → fall back. After audio started, replaying the whole
|
||||
// message via POST would stutter — treat what played as the playback.
|
||||
ws.onerror = () => settle(started ? 'done' : 'fallback')
|
||||
ws.onclose = () => (started ? finishWhenDrained() : settle('fallback'))
|
||||
|
||||
return {
|
||||
// Raw deltas — the server strips markdown/emoji per *sentence*, which is
|
||||
// the only safe granularity when constructs span delta boundaries.
|
||||
append: text => {
|
||||
if (text && !finished && !settled) {
|
||||
send({ text })
|
||||
}
|
||||
},
|
||||
finish: () => {
|
||||
if (!finished && !settled) {
|
||||
finished = true
|
||||
send({ done: true })
|
||||
}
|
||||
},
|
||||
done
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Live-speak an in-progress reply: open a session, then `append` deltas and
|
||||
* `finish` when generation completes. Resolves null when streaming is
|
||||
* unavailable (old backend / non-chunked provider) — the caller falls back to
|
||||
* whole-text `playSpeechText`.
|
||||
*/
|
||||
export async function startSpeechStream(options: VoicePlaybackOptions): Promise<null | SpeechStreamSession> {
|
||||
const wsUrl = await resolveSpeakStreamUrl()
|
||||
|
||||
if (!wsUrl) {
|
||||
return null
|
||||
}
|
||||
|
||||
stopVoicePlayback()
|
||||
setVoicePlaybackState(currentState('preparing', options))
|
||||
|
||||
const session = openSpeechStream(wsUrl, options)
|
||||
|
||||
void session.done.then(outcome => {
|
||||
if (outcome === 'done') {
|
||||
setVoicePlaybackState(currentState('idle'))
|
||||
}
|
||||
})
|
||||
|
||||
return session
|
||||
}
|
||||
|
||||
/** One-shot playback of complete text over the streaming WS. */
|
||||
function playSpeechStream(wsUrl: string, text: string, options: VoicePlaybackOptions): Promise<'fallback' | 'played'> {
|
||||
const session = openSpeechStream(wsUrl, options)
|
||||
session.append(text)
|
||||
session.finish()
|
||||
|
||||
return session.done.then(outcome => (outcome === 'done' ? 'played' : 'fallback'))
|
||||
}
|
||||
|
||||
async function playSpeechDataUrl(
|
||||
speakableText: string,
|
||||
options: VoicePlaybackOptions,
|
||||
isCurrent: () => boolean
|
||||
): Promise<boolean> {
|
||||
const response = await speakText(speakableText)
|
||||
|
||||
if (!isCurrent()) {
|
||||
return false
|
||||
}
|
||||
|
||||
const audio = new Audio(response.data_url)
|
||||
currentAudio = audio
|
||||
setVoicePlaybackState(currentState('speaking', options, audio))
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
let stall: number | null = null
|
||||
|
||||
const cleanup = () => {
|
||||
if (stall !== null) {
|
||||
window.clearTimeout(stall)
|
||||
stall = null
|
||||
}
|
||||
|
||||
audio.removeEventListener('ended', onEnded)
|
||||
audio.removeEventListener('error', onError)
|
||||
audio.removeEventListener('timeupdate', armStall)
|
||||
currentStop = null
|
||||
}
|
||||
|
||||
const armStall = () => {
|
||||
if (stall !== null) {
|
||||
window.clearTimeout(stall)
|
||||
}
|
||||
|
||||
stall = window.setTimeout(() => {
|
||||
cleanup()
|
||||
reject(new Error('Playback stalled'))
|
||||
}, PLAYBACK_STALL_MS)
|
||||
}
|
||||
|
||||
const onEnded = () => {
|
||||
cleanup()
|
||||
resolve()
|
||||
}
|
||||
|
||||
const onError = () => {
|
||||
cleanup()
|
||||
reject(new Error('Playback failed'))
|
||||
}
|
||||
|
||||
currentStop = () => {
|
||||
cleanup()
|
||||
resolve()
|
||||
}
|
||||
|
||||
audio.addEventListener('ended', onEnded, { once: true })
|
||||
audio.addEventListener('error', onError, { once: true })
|
||||
audio.addEventListener('timeupdate', armStall)
|
||||
armStall()
|
||||
void audio.play().catch(onError)
|
||||
})
|
||||
|
||||
if (!isCurrent()) {
|
||||
return false
|
||||
}
|
||||
|
||||
currentAudio = null
|
||||
|
||||
return true
|
||||
}
|
||||
|
||||
export async function playSpeechText(text: string, options: VoicePlaybackOptions): Promise<boolean> {
|
||||
stopVoicePlayback()
|
||||
|
||||
|
|
@ -73,72 +390,35 @@ export async function playSpeechText(text: string, options: VoicePlaybackOptions
|
|||
setVoicePlaybackState(currentState('preparing', options))
|
||||
|
||||
try {
|
||||
const response = await speakText(speakableText)
|
||||
// Streaming first; the POST data-URL path is the fallback for backends
|
||||
// without the WS endpoint or providers without a chunked API.
|
||||
const streamUrl = await resolveSpeakStreamUrl()
|
||||
|
||||
if (streamUrl && isCurrent()) {
|
||||
const outcome = await playSpeechStream(streamUrl, speakableText, options)
|
||||
|
||||
if (outcome === 'played') {
|
||||
if (!isCurrent()) {
|
||||
return false
|
||||
}
|
||||
|
||||
setVoicePlaybackState(currentState('idle'))
|
||||
|
||||
return true
|
||||
}
|
||||
}
|
||||
|
||||
if (!isCurrent()) {
|
||||
return false
|
||||
}
|
||||
|
||||
const audio = new Audio(response.data_url)
|
||||
currentAudio = audio
|
||||
setVoicePlaybackState(currentState('speaking', options, audio))
|
||||
const played = await playSpeechDataUrl(speakableText, options, isCurrent)
|
||||
|
||||
await new Promise<void>((resolve, reject) => {
|
||||
let stall: number | null = null
|
||||
|
||||
const cleanup = () => {
|
||||
if (stall !== null) {
|
||||
window.clearTimeout(stall)
|
||||
stall = null
|
||||
}
|
||||
|
||||
audio.removeEventListener('ended', onEnded)
|
||||
audio.removeEventListener('error', onError)
|
||||
audio.removeEventListener('timeupdate', armStall)
|
||||
currentStop = null
|
||||
}
|
||||
|
||||
const armStall = () => {
|
||||
if (stall !== null) {
|
||||
window.clearTimeout(stall)
|
||||
}
|
||||
|
||||
stall = window.setTimeout(() => {
|
||||
cleanup()
|
||||
reject(new Error('Playback stalled'))
|
||||
}, PLAYBACK_STALL_MS)
|
||||
}
|
||||
|
||||
const onEnded = () => {
|
||||
cleanup()
|
||||
resolve()
|
||||
}
|
||||
|
||||
const onError = () => {
|
||||
cleanup()
|
||||
reject(new Error('Playback failed'))
|
||||
}
|
||||
|
||||
currentStop = () => {
|
||||
cleanup()
|
||||
resolve()
|
||||
}
|
||||
|
||||
audio.addEventListener('ended', onEnded, { once: true })
|
||||
audio.addEventListener('error', onError, { once: true })
|
||||
audio.addEventListener('timeupdate', armStall)
|
||||
armStall()
|
||||
void audio.play().catch(onError)
|
||||
})
|
||||
|
||||
if (!isCurrent()) {
|
||||
return false
|
||||
if (played) {
|
||||
setVoicePlaybackState(currentState('idle'))
|
||||
}
|
||||
|
||||
currentAudio = null
|
||||
setVoicePlaybackState(currentState('idle'))
|
||||
|
||||
return true
|
||||
return played
|
||||
} catch (error) {
|
||||
if (isCurrent()) {
|
||||
currentStop = null
|
||||
|
|
|
|||
|
|
@ -9,6 +9,7 @@ Usage:
|
|||
python -m hermes_cli.main web --port 8080
|
||||
"""
|
||||
|
||||
import contextlib
|
||||
from contextlib import asynccontextmanager, contextmanager
|
||||
|
||||
import asyncio
|
||||
|
|
@ -28,6 +29,7 @@ import json
|
|||
import logging
|
||||
import mimetypes
|
||||
import os
|
||||
import queue
|
||||
import re
|
||||
import secrets
|
||||
import shlex
|
||||
|
|
@ -4269,10 +4271,14 @@ async def transcribe_audio_upload(payload: AudioTranscriptionRequest):
|
|||
tmp.write(audio_bytes)
|
||||
temp_path = tmp.name
|
||||
|
||||
from tools.transcription_tools import transcribe_audio
|
||||
# transcribe_recording (not raw transcribe_audio): filters Whisper
|
||||
# hallucinations and maps provider "empty transcript" errors to a
|
||||
# successful empty result — the live voice loop treats "" as silence
|
||||
# and re-listens instead of surfacing a 400 on every quiet turn.
|
||||
from tools.voice_mode import transcribe_recording
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
result = await loop.run_in_executor(None, transcribe_audio, temp_path)
|
||||
result = await loop.run_in_executor(None, transcribe_recording, temp_path)
|
||||
except HTTPException:
|
||||
raise
|
||||
except Exception as exc:
|
||||
|
|
@ -4464,6 +4470,151 @@ async def speak_text(payload: TTSSpeakRequest):
|
|||
}
|
||||
|
||||
|
||||
def _split_text_for_speak_stream(text: str, cap: int) -> list:
|
||||
"""Split *text* into provider-cap-sized pieces on sentence boundaries."""
|
||||
from tools.tts_streaming import SENTENCE_BOUNDARY_RE as _SENTENCE_BOUNDARY_RE
|
||||
|
||||
cap = cap if cap and cap > 0 else 4000
|
||||
pieces, buf = [], ""
|
||||
for sentence in filter(str.strip, _SENTENCE_BOUNDARY_RE.split(text)):
|
||||
while len(sentence) > cap:
|
||||
pieces.append(sentence[:cap])
|
||||
sentence = sentence[cap:]
|
||||
if buf and len(buf) + len(sentence) + 1 > cap:
|
||||
pieces.append(buf)
|
||||
buf = sentence
|
||||
else:
|
||||
buf = f"{buf} {sentence}" if buf else sentence
|
||||
if buf:
|
||||
pieces.append(buf)
|
||||
return pieces
|
||||
|
||||
|
||||
@app.websocket("/api/audio/speak-stream")
|
||||
async def speak_stream_ws(ws: "WebSocket") -> None:
|
||||
"""Streaming TTS for the desktop: text in, raw int16 PCM frames out.
|
||||
|
||||
The socket is a per-reply speech *session*: the client feeds text
|
||||
incrementally as LLM deltas arrive, the server cuts sentences
|
||||
(``SentenceChunker`` — same cutter as the CLI/TUI speaker pipeline) and
|
||||
streams each one's PCM the moment it's ready. Speech overlaps generation,
|
||||
exactly like the token→sentence→TTS pipelining the realtime-voice
|
||||
literature converges on.
|
||||
|
||||
Protocol:
|
||||
client → ``{"text": "..."}`` frames (incremental; may combine with done),
|
||||
``{"done": true}`` when the reply is complete,
|
||||
``{"stop": true}`` or disconnect = barge-in
|
||||
server → ``{"type": "start", "sample_rate": N, "channels": 1}``,
|
||||
binary PCM frames, then ``{"type": "end"}``
|
||||
server → ``{"type": "fallback"}`` when the configured provider has no
|
||||
chunked API — the client uses the POST endpoint instead.
|
||||
"""
|
||||
if not _ws_auth_ok(ws):
|
||||
await ws.close(code=4401)
|
||||
return
|
||||
if not _ws_request_is_allowed(ws):
|
||||
await ws.close(code=4403)
|
||||
return
|
||||
await ws.accept()
|
||||
|
||||
loop = asyncio.get_running_loop()
|
||||
|
||||
def _resolve():
|
||||
from tools.tts_streaming import resolve_streaming_provider
|
||||
from tools.tts_tool import _get_provider, _load_tts_config, _resolve_max_text_length
|
||||
|
||||
cfg = _load_tts_config()
|
||||
streamer = resolve_streaming_provider(cfg)
|
||||
cap = _resolve_max_text_length(_get_provider(cfg), cfg) if streamer else 0
|
||||
return streamer, cap
|
||||
|
||||
try:
|
||||
streamer, cap = await loop.run_in_executor(None, _resolve)
|
||||
except Exception:
|
||||
_log.exception("speak-stream provider resolution failed")
|
||||
streamer, cap = None, 0
|
||||
if streamer is None:
|
||||
with contextlib.suppress(Exception):
|
||||
await ws.send_json({"type": "fallback"})
|
||||
await ws.close()
|
||||
return
|
||||
|
||||
await ws.send_json(
|
||||
{"type": "start", "sample_rate": streamer.sample_rate, "channels": streamer.channels}
|
||||
)
|
||||
|
||||
stop = threading.Event()
|
||||
text_q: queue.Queue = queue.Queue() # str deltas; None = end-of-text
|
||||
chunks: asyncio.Queue = asyncio.Queue() # PCM out; None = synthesis done
|
||||
|
||||
def _produce():
|
||||
from tools.tts_streaming import SentenceChunker
|
||||
from tools.tts_tool import _strip_markdown_for_tts
|
||||
|
||||
chunker = SentenceChunker()
|
||||
|
||||
def _sentences():
|
||||
while not stop.is_set():
|
||||
delta = text_q.get()
|
||||
if delta is None:
|
||||
yield from chunker.flush()
|
||||
return
|
||||
yield from chunker.feed(delta)
|
||||
|
||||
try:
|
||||
for sentence in _sentences():
|
||||
cleaned = _strip_markdown_for_tts(sentence)
|
||||
if not cleaned:
|
||||
continue
|
||||
for piece in _split_text_for_speak_stream(cleaned, cap):
|
||||
for chunk in streamer.stream(piece):
|
||||
if stop.is_set():
|
||||
return
|
||||
loop.call_soon_threadsafe(chunks.put_nowait, chunk)
|
||||
except Exception as exc:
|
||||
_log.warning("speak-stream synthesis failed: %s", exc)
|
||||
finally:
|
||||
loop.call_soon_threadsafe(chunks.put_nowait, None)
|
||||
|
||||
threading.Thread(target=_produce, daemon=True).start()
|
||||
|
||||
async def _pump_client():
|
||||
# Text frames feed synthesis; done ends the text; stop/disconnect
|
||||
# (or any unparseable frame) is barge-in.
|
||||
try:
|
||||
while True:
|
||||
frame = json.loads(await ws.receive_text())
|
||||
if frame.get("text"):
|
||||
text_q.put(str(frame["text"]))
|
||||
if frame.get("stop"):
|
||||
break
|
||||
if frame.get("done"):
|
||||
text_q.put(None)
|
||||
except Exception:
|
||||
pass
|
||||
stop.set()
|
||||
text_q.put(None) # unblock the producer
|
||||
|
||||
pump = asyncio.ensure_future(_pump_client())
|
||||
try:
|
||||
while True:
|
||||
chunk = await chunks.get()
|
||||
if chunk is None:
|
||||
break
|
||||
await ws.send_bytes(chunk)
|
||||
if not stop.is_set():
|
||||
await ws.send_json({"type": "end"})
|
||||
except (WebSocketDisconnect, RuntimeError):
|
||||
pass
|
||||
finally:
|
||||
stop.set()
|
||||
text_q.put(None)
|
||||
pump.cancel()
|
||||
with contextlib.suppress(Exception):
|
||||
await ws.close()
|
||||
|
||||
|
||||
@app.get("/api/actions/{name}/status")
|
||||
async def get_action_status(name: str, lines: int = 200):
|
||||
"""Tail an action log and report whether the process is still running."""
|
||||
|
|
|
|||
|
|
@ -2192,7 +2192,7 @@ class TestWebServerEndpoints:
|
|||
|
||||
captured = {}
|
||||
|
||||
def fake_transcribe_audio(path):
|
||||
def fake_transcribe_audio(path, model=None):
|
||||
captured["path"] = path
|
||||
return {
|
||||
"success": True,
|
||||
|
|
@ -2219,6 +2219,36 @@ class TestWebServerEndpoints:
|
|||
assert captured["path"].endswith(".webm")
|
||||
assert not Path(captured["path"]).exists()
|
||||
|
||||
def test_audio_transcription_no_speech_is_not_an_error(self, monkeypatch):
|
||||
"""A provider hearing silence (empty transcript) must return 200/"" —
|
||||
the live voice loop treats it as a quiet turn and re-listens, instead
|
||||
of surfacing a 400 toast on every pause (the ElevenLabs empty-
|
||||
transcript spam)."""
|
||||
import tools.transcription_tools as transcription_tools
|
||||
|
||||
monkeypatch.setattr(
|
||||
transcription_tools,
|
||||
"transcribe_audio",
|
||||
lambda path, model=None: {
|
||||
"success": False,
|
||||
"transcript": "",
|
||||
"error": "ElevenLabs STT returned empty transcript",
|
||||
"no_speech": True,
|
||||
},
|
||||
)
|
||||
|
||||
resp = self.client.post(
|
||||
"/api/audio/transcribe",
|
||||
json={
|
||||
"data_url": "data:audio/webm;base64,aGVsbG8=",
|
||||
"mime_type": "audio/webm",
|
||||
},
|
||||
)
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert resp.json()["ok"] is True
|
||||
assert resp.json()["transcript"] == ""
|
||||
|
||||
def test_audio_transcription_rejects_invalid_base64(self):
|
||||
resp = self.client.post(
|
||||
"/api/audio/transcribe",
|
||||
|
|
|
|||
|
|
@ -0,0 +1,163 @@
|
|||
"""/api/audio/speak-stream — desktop streaming TTS over WebSocket."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import json
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import pytest
|
||||
from starlette.testclient import TestClient
|
||||
from starlette.websockets import WebSocketDisconnect
|
||||
|
||||
from hermes_cli import web_server
|
||||
|
||||
|
||||
@pytest.fixture
|
||||
def stream_client(monkeypatch, _isolate_hermes_home):
|
||||
previous_auth_required = getattr(web_server.app.state, "auth_required", None)
|
||||
web_server.app.state.auth_required = False
|
||||
|
||||
client = TestClient(web_server.app)
|
||||
try:
|
||||
yield client
|
||||
finally:
|
||||
close = getattr(client, "close", None)
|
||||
if close is not None:
|
||||
close()
|
||||
if previous_auth_required is None:
|
||||
if hasattr(web_server.app.state, "auth_required"):
|
||||
delattr(web_server.app.state, "auth_required")
|
||||
else:
|
||||
web_server.app.state.auth_required = previous_auth_required
|
||||
|
||||
|
||||
def _url(token: str | None = None) -> str:
|
||||
return f"/api/audio/speak-stream?{urlencode({'token': token or web_server._SESSION_TOKEN})}"
|
||||
|
||||
|
||||
class _FakeStreamer:
|
||||
sample_rate = 24000
|
||||
channels = 1
|
||||
|
||||
def __init__(self, chunks):
|
||||
self.chunks = chunks
|
||||
self.requests: list[str] = []
|
||||
|
||||
def stream(self, text):
|
||||
self.requests.append(text)
|
||||
yield from self.chunks
|
||||
|
||||
|
||||
def _patch_provider(monkeypatch, streamer, cap=4000):
|
||||
monkeypatch.setattr("tools.tts_streaming.resolve_streaming_provider", lambda cfg: streamer)
|
||||
monkeypatch.setattr("tools.tts_tool._load_tts_config", lambda: {})
|
||||
monkeypatch.setattr("tools.tts_tool._get_provider", lambda cfg: "fake")
|
||||
monkeypatch.setattr("tools.tts_tool._resolve_max_text_length", lambda provider, cfg: cap)
|
||||
|
||||
|
||||
def test_rejects_bad_token(stream_client):
|
||||
with pytest.raises(WebSocketDisconnect) as exc:
|
||||
with stream_client.websocket_connect(_url(token="wrong")):
|
||||
pass
|
||||
assert exc.value.code == 4401
|
||||
|
||||
|
||||
def test_fallback_frame_when_no_streaming_provider(stream_client, monkeypatch):
|
||||
_patch_provider(monkeypatch, None)
|
||||
with stream_client.websocket_connect(_url()) as conn:
|
||||
assert conn.receive_json() == {"type": "fallback"}
|
||||
|
||||
|
||||
def test_streams_pcm_frames_then_end(stream_client, monkeypatch):
|
||||
streamer = _FakeStreamer([b"\x01\x02\x03\x04", b"\x05\x06"])
|
||||
_patch_provider(monkeypatch, streamer)
|
||||
|
||||
with stream_client.websocket_connect(_url()) as conn:
|
||||
start = conn.receive_json()
|
||||
assert start == {"type": "start", "sample_rate": 24000, "channels": 1}
|
||||
|
||||
conn.send_text(json.dumps({"text": "Hello there.", "done": True}))
|
||||
assert conn.receive_bytes() == b"\x01\x02\x03\x04"
|
||||
assert conn.receive_bytes() == b"\x05\x06"
|
||||
assert conn.receive_json() == {"type": "end"}
|
||||
|
||||
assert streamer.requests == ["Hello there."]
|
||||
|
||||
|
||||
def test_incremental_deltas_are_cut_into_sentences(stream_client, monkeypatch):
|
||||
"""Text fed across frames is chunked and synthesized while more arrives."""
|
||||
streamer = _FakeStreamer([b"\x00\x00"])
|
||||
_patch_provider(monkeypatch, streamer)
|
||||
|
||||
with stream_client.websocket_connect(_url()) as conn:
|
||||
assert conn.receive_json()["type"] == "start"
|
||||
conn.send_text(json.dumps({"text": "This is the first full"}))
|
||||
conn.send_text(json.dumps({"text": " sentence of the reply. And"}))
|
||||
# The first sentence is complete — PCM must arrive before `done`.
|
||||
assert conn.receive_bytes() == b"\x00\x00"
|
||||
conn.send_text(json.dumps({"text": " here is the second one.", "done": True}))
|
||||
assert conn.receive_bytes() == b"\x00\x00"
|
||||
assert conn.receive_json() == {"type": "end"}
|
||||
|
||||
assert streamer.requests == [
|
||||
"This is the first full sentence of the reply.",
|
||||
"And here is the second one.",
|
||||
]
|
||||
|
||||
|
||||
def test_stop_frame_cuts_synthesis(stream_client, monkeypatch):
|
||||
streamer = _FakeStreamer([b"\x00\x00"])
|
||||
_patch_provider(monkeypatch, streamer)
|
||||
|
||||
with stream_client.websocket_connect(_url()) as conn:
|
||||
assert conn.receive_json()["type"] == "start"
|
||||
conn.send_text(json.dumps({"stop": True}))
|
||||
# Socket closes without an "end" frame — barge-in, not completion.
|
||||
with pytest.raises(WebSocketDisconnect):
|
||||
conn.receive_text()
|
||||
assert streamer.requests == []
|
||||
|
||||
|
||||
def test_long_text_is_split_across_provider_requests(stream_client, monkeypatch):
|
||||
streamer = _FakeStreamer([b"\x00\x00"])
|
||||
_patch_provider(monkeypatch, streamer, cap=24)
|
||||
|
||||
with stream_client.websocket_connect(_url()) as conn:
|
||||
assert conn.receive_json()["type"] == "start"
|
||||
conn.send_text(
|
||||
json.dumps(
|
||||
{"text": "First sentence here. Second sentence here. Third one.", "done": True}
|
||||
)
|
||||
)
|
||||
# One PCM frame per split piece, then end.
|
||||
frames = 0
|
||||
while True:
|
||||
message = conn.receive()
|
||||
if message.get("bytes") is not None:
|
||||
frames += 1
|
||||
else:
|
||||
assert json.loads(message["text"]) == {"type": "end"}
|
||||
break
|
||||
|
||||
assert len(streamer.requests) > 1
|
||||
assert frames == len(streamer.requests)
|
||||
# Nothing lost in the split: every sentence reached the provider.
|
||||
joined = " ".join(streamer.requests)
|
||||
for fragment in ("First sentence here.", "Second sentence here.", "Third one."):
|
||||
assert fragment in joined
|
||||
|
||||
|
||||
def test_split_text_respects_cap_and_preserves_content():
|
||||
text = "Alpha beta. Gamma delta epsilon. Zeta eta theta iota kappa."
|
||||
pieces = web_server._split_text_for_speak_stream(text, 30)
|
||||
assert pieces
|
||||
assert all(len(piece) <= 30 for piece in pieces)
|
||||
joined = " ".join(pieces)
|
||||
for word in text.replace(".", "").split():
|
||||
assert word in joined
|
||||
|
||||
|
||||
def test_split_text_hard_splits_oversized_sentence():
|
||||
pieces = web_server._split_text_for_speak_stream("x" * 100, 30)
|
||||
assert all(len(piece) <= 30 for piece in pieces)
|
||||
assert sum(len(piece) for piece in pieces) == 100
|
||||
Loading…
Reference in New Issue