diff --git a/apps/desktop/src/app/chat/composer/hooks/use-voice-conversation.ts b/apps/desktop/src/app/chat/composer/hooks/use-voice-conversation.ts index a8725cac666c1..dc8e7dcdf138c 100644 --- a/apps/desktop/src/app/chat/composer/hooks/use-voice-conversation.ts +++ b/apps/desktop/src/app/chat/composer/hooks/use-voice-conversation.ts @@ -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(null) const spokenSourceLengthRef = useRef(0) - const speechBufferRef = useRef('') + const speechSessionRef = useRef(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) { diff --git a/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts b/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts index 709652dce7bf5..9493227b9da9c 100644 --- a/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts +++ b/apps/desktop/src/app/session/hooks/use-prompt-actions/submit.ts @@ -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. diff --git a/apps/desktop/src/lib/voice-barge-in.ts b/apps/desktop/src/lib/voice-barge-in.ts new file mode 100644 index 0000000000000..c9170e7d98656 --- /dev/null +++ b/apps/desktop/src/lib/voice-barge-in.ts @@ -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 +} diff --git a/apps/desktop/src/lib/voice-playback.ts b/apps/desktop/src/lib/voice-playback.ts index eea1b5b6e0ab2..142423d5d5bd2 100644 --- a/apps/desktop/src/lib/voice-playback.ts +++ b/apps/desktop/src/lib/voice-playback.ts @@ -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 { + 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 { + 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 { + 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((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 { 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((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 diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index f6c43a550b8b2..1181631cb2e1d 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -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.""" diff --git a/tests/hermes_cli/test_web_server.py b/tests/hermes_cli/test_web_server.py index d3e684bb3c2f1..ea19195c2826d 100644 --- a/tests/hermes_cli/test_web_server.py +++ b/tests/hermes_cli/test_web_server.py @@ -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", diff --git a/tests/hermes_cli/test_web_server_speak_stream.py b/tests/hermes_cli/test_web_server_speak_stream.py new file mode 100644 index 0000000000000..50a36f35a90fb --- /dev/null +++ b/tests/hermes_cli/test_web_server_speak_stream.py @@ -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