diff --git a/apps/desktop/src/lib/voice-playback.test.ts b/apps/desktop/src/lib/voice-playback.test.ts new file mode 100644 index 000000000000..62aa7ecf51d1 --- /dev/null +++ b/apps/desktop/src/lib/voice-playback.test.ts @@ -0,0 +1,133 @@ +import { afterEach, beforeEach, describe, expect, it, vi } from 'vitest' + +import { startSpeechStream, stopVoicePlayback } from './voice-playback' + +const mocks = vi.hoisted(() => ({ + getApiRequestProfile: vi.fn(() => null), + resolveGatewayWsUrl: vi.fn(async () => 'ws://localhost/api/ws?token=test'), + speakText: vi.fn() +})) + +vi.mock('@hermes/shared', () => ({ + resolveGatewayWsUrl: mocks.resolveGatewayWsUrl +})) + +vi.mock('@/hermes', () => ({ + getApiRequestProfile: mocks.getApiRequestProfile, + speakText: mocks.speakText +})) + +class FakeWebSocket { + static readonly CLOSED = 3 + static readonly CONNECTING = 0 + static readonly OPEN = 1 + static instances: FakeWebSocket[] = [] + + binaryType = '' + onclose: null | (() => void) = null + onerror: null | (() => void) = null + onmessage: null | ((event: MessageEvent) => void) = null + onopen: null | (() => void) = null + readyState = FakeWebSocket.CONNECTING + readonly sent: string[] = [] + + constructor(readonly url: string) { + FakeWebSocket.instances.push(this) + } + + close() { + this.readyState = FakeWebSocket.CLOSED + } + + emit(data: ArrayBuffer | string) { + this.onmessage?.({ data } as MessageEvent) + } + + open() { + this.readyState = FakeWebSocket.OPEN + this.onopen?.() + } + + send(data: string) { + this.sent.push(data) + } +} + +class FakeAudioContext { + static instances: FakeAudioContext[] = [] + + readonly close = vi.fn(async () => undefined) + readonly createBufferSource = vi.fn(() => ({ + buffer: null, + connect: vi.fn(), + start: vi.fn() + })) + currentTime = 0 + readonly destination = {} + readonly resume = vi.fn(async () => undefined) + state: AudioContextState = 'running' + + constructor() { + FakeAudioContext.instances.push(this) + } + + async decodeAudioData(data: ArrayBuffer): Promise { + const marker = new Uint8Array(data)[0] + + if (marker === 2) { + throw new Error('invalid encoded clip') + } + + return { duration: 0.01 } as AudioBuffer + } +} + +async function flushPromises() { + await Promise.resolve() + await Promise.resolve() + await Promise.resolve() + await Promise.resolve() +} + +describe('encoded voice playback', () => { + beforeEach(() => { + FakeWebSocket.instances = [] + FakeAudioContext.instances = [] + vi.stubGlobal('WebSocket', FakeWebSocket) + vi.stubGlobal('AudioContext', FakeAudioContext) + Object.defineProperty(window, 'hermesDesktop', { + configurable: true, + value: { getConnection: vi.fn(async () => ({ wsUrl: 'ws://localhost/api/ws?token=test' })) } + }) + }) + + afterEach(() => { + stopVoicePlayback() + vi.unstubAllGlobals() + Reflect.deleteProperty(window, 'hermesDesktop') + vi.clearAllMocks() + }) + + it('stops decoding after an invalid middle clip and drains scheduled audio', async () => { + const session = await startSpeechStream({ source: 'voice-conversation' }) + expect(session).not.toBeNull() + + const socket = FakeWebSocket.instances[0] + expect(new URL(socket.url).searchParams.get('audio_protocol')).toBe('2') + socket.open() + socket.emit(JSON.stringify({ type: 'start', encoding: 'encoded' })) + socket.emit(Uint8Array.of(1).buffer) + socket.emit(Uint8Array.of(2).buffer) + socket.emit(Uint8Array.of(3).buffer) + socket.emit(JSON.stringify({ type: 'end' })) + + await flushPromises() + + const context = FakeAudioContext.instances[0] + expect(context.createBufferSource).toHaveBeenCalledTimes(1) + expect(context.close).not.toHaveBeenCalled() + + await expect(session?.done).resolves.toBe('done') + expect(context.close).toHaveBeenCalledTimes(1) + }) +}) diff --git a/apps/desktop/src/lib/voice-playback.ts b/apps/desktop/src/lib/voice-playback.ts index 48a51d370817..2b1664c8c312 100644 --- a/apps/desktop/src/lib/voice-playback.ts +++ b/apps/desktop/src/lib/voice-playback.ts @@ -91,9 +91,9 @@ 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. +// Streaming path — /api/audio/speak-stream WebSocket, raw int16 PCM chunks or +// complete encoded sentence clips scheduled through Web Audio. Speech starts +// on the provider's first sentence instead of after full-response synthesis. // --------------------------------------------------------------------------- async function resolveSpeakStreamUrl(): Promise { @@ -105,8 +105,8 @@ async function resolveSpeakStreamUrl(): Promise { try { // Mint a fresh credential (single-use ticket in OAuth mode) for the - // ACTIVE profile's backend, then swap the gateway endpoint for the PCM - // one — auth is shared across WS routes. + // ACTIVE profile's backend, then swap the gateway endpoint for the speech + // stream — auth is shared across WS routes. const profile = getApiRequestProfile() const wsUrl = await resolveGatewayWsUrl(desktop, await desktop.getConnection(profile)) const url = new URL(wsUrl) @@ -123,6 +123,11 @@ async function resolveSpeakStreamUrl(): Promise { url.searchParams.set('profile', profile) } + // Protocol v2 explicitly opts into browser-decodable encoded sentence + // frames. Older clients omit this and receive fallback for sync providers + // instead of misinterpreting MP3 bytes as raw PCM. + url.searchParams.set('audio_protocol', '2') + return url.toString() } catch { return null @@ -145,17 +150,22 @@ export interface SpeechStreamSession { /** * 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. + * streams audio back while generation continues, so speech overlaps the text + * stream (ChatGPT-style). Chunked providers send PCM; providers such as Edge + * send one browser-decodable audio file per sentence over the same socket. */ function openSpeechStream(wsUrl: string, options: VoicePlaybackOptions): SpeechStreamSession { const ws = new WebSocket(wsUrl) ws.binaryType = 'arraybuffer' let context: AudioContext | null = null + let encoding: 'encoded' | 'pcm' = 'pcm' let streamRate = 24_000 let nextStartAt = 0 let carry: null | Uint8Array = null + let decodeQueue = Promise.resolve() + let decodeFailed = false + let receivedAudio = false let started = false let settled = false let finished = false @@ -203,7 +213,26 @@ function openSpeechStream(wsUrl: string, options: VoicePlaybackOptions): SpeechS window.setTimeout(() => settle('done'), remainingMs + 100) } - const schedule = (data: ArrayBuffer) => { + const scheduleBuffer = (buffer: AudioBuffer) => { + if (!context) { + return + } + + 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)) + } + } + + const schedulePcm = (data: ArrayBuffer) => { if (!context) { return } @@ -237,18 +266,37 @@ function openSpeechStream(wsUrl: string, options: VoicePlaybackOptions): SpeechS channel[index] = pcm[index] / 32_768 } - const source = context.createBufferSource() - source.buffer = buffer - source.connect(context.destination) + scheduleBuffer(buffer) + } - const startAt = Math.max(context.currentTime + 0.05, nextStartAt) - source.start(startAt) - nextStartAt = startAt + buffer.duration + const scheduleEncoded = (data: ArrayBuffer) => { + const owner = context - if (!started) { - started = true - setVoicePlaybackState(currentState('speaking', options)) + if (!owner || decodeFailed) { + return } + + // Decoding is asynchronous. Chaining sentences preserves provider order + // even when clips take different amounts of time to decode. + decodeQueue = decodeQueue + .then(async () => { + if (decodeFailed || settled || context !== owner) { + return + } + + const buffer = await owner.decodeAudioData(data.slice(0)) + + if (!settled && context === owner) { + scheduleBuffer(buffer) + } + }) + .catch(() => { + decodeFailed = true + + if (!started) { + settle('fallback') + } + }) } ws.onopen = () => { @@ -257,12 +305,18 @@ function openSpeechStream(wsUrl: string, options: VoicePlaybackOptions): SpeechS ws.onmessage = event => { if (typeof event.data !== 'string') { - schedule(event.data as ArrayBuffer) + receivedAudio = true + + if (encoding === 'encoded') { + scheduleEncoded(event.data as ArrayBuffer) + } else { + schedulePcm(event.data as ArrayBuffer) + } return } - let frame: { channels?: number; sample_rate?: number; type?: string } + let frame: { channels?: number; encoding?: 'encoded' | 'pcm'; sample_rate?: number; type?: string } try { frame = JSON.parse(event.data) as typeof frame @@ -271,6 +325,7 @@ function openSpeechStream(wsUrl: string, options: VoicePlaybackOptions): SpeechS } if (frame.type === 'start') { + encoding = frame.encoding || 'pcm' streamRate = frame.sample_rate || 24_000 context = new AudioContext() @@ -285,7 +340,7 @@ function openSpeechStream(wsUrl: string, options: VoicePlaybackOptions): SpeechS nextStartAt = 0 } else if (frame.type === 'end') { - finishWhenDrained() + void decodeQueue.then(finishWhenDrained) } else if (frame.type === 'fallback') { settle(started ? 'done' : 'fallback') } @@ -294,8 +349,21 @@ function openSpeechStream(wsUrl: string, options: VoicePlaybackOptions): SpeechS // 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')) + ws.onerror = () => { + if (receivedAudio) { + void decodeQueue.then(finishWhenDrained) + } else { + settle(started ? 'done' : 'fallback') + } + } + + ws.onclose = () => { + if (receivedAudio) { + void decodeQueue.then(finishWhenDrained) + } else { + settle(started ? 'done' : 'fallback') + } + } return { // Raw deltas — the server strips markdown/emoji per *sentence*, which is diff --git a/hermes_cli/web_server.py b/hermes_cli/web_server.py index 767882ba9d1a..8386339f3f29 100644 --- a/hermes_cli/web_server.py +++ b/hermes_cli/web_server.py @@ -4487,12 +4487,12 @@ def _split_text_for_speak_stream(text: str, cap: int) -> list: @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. + """Streaming TTS for Desktop: text in, PCM or encoded sentence audio 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, + streams each one's audio the moment it's ready. Speech overlaps generation, exactly like the token→sentence→TTS pipelining the realtime-voice literature converges on. @@ -4501,9 +4501,13 @@ async def speak_stream_ws(ws: "WebSocket") -> None: ``{"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. + binary PCM frames, then ``{"type": "end"}`` for chunked providers + server → ``{"type": "start", "encoding": "encoded"}``, then one complete + browser-decodable audio file per sentence for sync providers. + + Encoded audio is opt-in via ``?audio_protocol=2``. Older Desktop clients + treat every binary frame as PCM, so sync providers must fall back rather + than send them encoded files they would play as noise. """ if not _ws_auth_ok(ws): await ws.close(code=4401) @@ -4528,34 +4532,69 @@ def _resolve(): with _config_profile_scope(profile): 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 + provider = _get_provider(cfg) + cap = _resolve_max_text_length(provider, cfg) + return streamer, cap, provider try: - streamer, cap = await loop.run_in_executor(None, _resolve) + streamer, cap, provider = 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} - ) + supports_encoded_audio = ws.query_params.get("audio_protocol") == "2" + if streamer is None and not supports_encoded_audio: + await ws.send_json({"type": "fallback"}) + await ws.close() + return + if streamer is None: + await ws.send_json({"type": "start", "encoding": "encoded"}) + else: + 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 + synthesis_failed = object() + chunks: asyncio.Queue = asyncio.Queue() # bytes; failure marker; None = synthesis done def _produce(): from tools.tts_streaming import SentenceChunker - from tools.tts_tool import _strip_markdown_for_tts + from tools.tts_tool import _strip_markdown_for_tts, text_to_speech_tool chunker = SentenceChunker() + def _sync_sentence_audio(sentence: str) -> bytes: + requested_path = None + generated_path = None + try: + handle = tempfile.NamedTemporaryFile(suffix=".mp3", delete=False) + requested_path = handle.name + handle.close() + with _config_profile_scope(profile): + result = json.loads( + text_to_speech_tool( + text=sentence, + output_path=requested_path, + provider=provider, + ) + ) + if not result.get("success"): + raise RuntimeError(result.get("error") or "TTS synthesis failed") + generated_path = result.get("file_path") or requested_path + audio = Path(generated_path).read_bytes() + if not audio: + raise RuntimeError("TTS synthesis produced empty audio") + return audio + finally: + for candidate in {requested_path, generated_path} - {None}: + with contextlib.suppress(OSError): + Path(candidate).unlink(missing_ok=True) + # The session stays open for a whole agent turn, and the client only # sends `done` when the turn ends. During tool execution no text # arrives, so without an idle flush a narration line with no trailing @@ -4592,12 +4631,18 @@ def _sentences(): if not cleaned: continue for piece in _split_text_for_speak_stream(cleaned, cap): - for chunk in streamer.stream(piece): + audio_chunks = ( + streamer.stream(piece) + if streamer is not None + else (_sync_sentence_audio(piece),) + ) + for chunk in audio_chunks: 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) + loop.call_soon_threadsafe(chunks.put_nowait, synthesis_failed) finally: loop.call_soon_threadsafe(chunks.put_nowait, None) @@ -4622,13 +4667,22 @@ async def _pump_client(): pump = asyncio.ensure_future(_pump_client()) try: + sent_audio = False + failed = False while True: chunk = await chunks.get() if chunk is None: break + if chunk is synthesis_failed: + failed = True + continue await ws.send_bytes(chunk) + sent_audio = True if not stop.is_set(): - await ws.send_json({"type": "end"}) + if failed and not sent_audio: + await ws.send_json({"type": "fallback"}) + else: + await ws.send_json({"type": "end"}) except (WebSocketDisconnect, RuntimeError): pass finally: diff --git a/tests/hermes_cli/test_web_server_speak_stream.py b/tests/hermes_cli/test_web_server_speak_stream.py index 01220e7e44dd..f98fb680b3f1 100644 --- a/tests/hermes_cli/test_web_server_speak_stream.py +++ b/tests/hermes_cli/test_web_server_speak_stream.py @@ -4,6 +4,7 @@ import json import time +from pathlib import Path from urllib.parse import urlencode import pytest @@ -32,8 +33,11 @@ def stream_client(monkeypatch, _isolate_hermes_home): 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})}" +def _url(token: str | None = None, *, audio_protocol: int | None = None) -> str: + params = {"token": token or web_server._SESSION_TOKEN} + if audio_protocol is not None: + params["audio_protocol"] = str(audio_protocol) + return f"/api/audio/speak-stream?{urlencode(params)}" class _FakeStreamer: @@ -56,6 +60,20 @@ def _patch_provider(monkeypatch, streamer, cap=4000): monkeypatch.setattr("tools.tts_tool._resolve_max_text_length", lambda provider, cfg: cap) +def _patch_sync_edge_provider(monkeypatch): + monkeypatch.setattr("tools.tts_streaming.resolve_streaming_provider", lambda cfg: None) + monkeypatch.setattr("tools.tts_tool._load_tts_config", lambda: {"provider": "edge"}) + monkeypatch.setattr("tools.tts_tool._get_provider", lambda cfg: "edge") + monkeypatch.setattr("tools.tts_tool._resolve_max_text_length", lambda provider, cfg: 5000) + + def fake_tts(*, text, output_path, provider): + path = Path(output_path) + path.write_bytes(f"audio:{text}".encode()) + return json.dumps({"success": True, "file_path": str(path), "provider": provider}) + + monkeypatch.setattr("tools.tts_tool.text_to_speech_tool", fake_tts) + + @@ -76,6 +94,63 @@ def test_streams_pcm_frames_then_end(stream_client, monkeypatch): assert streamer.requests == ["Hello there."] +def test_sync_edge_streams_first_sentence_before_reply_finishes(stream_client, monkeypatch): + _patch_sync_edge_provider(monkeypatch) + + with stream_client.websocket_connect(_url(audio_protocol=2)) as conn: + assert conn.receive_json() == {"type": "start", "encoding": "encoded"} + + conn.send_text(json.dumps({"text": "The first sentence is ready. "})) + assert conn.receive_bytes() == b"audio:The first sentence is ready." + + conn.send_text(json.dumps({"text": "The second sentence follows.", "done": True})) + assert conn.receive_bytes() == b"audio:The second sentence follows." + assert conn.receive_json() == {"type": "end"} + + +def test_sync_provider_falls_back_for_legacy_desktop_client(stream_client, monkeypatch): + _patch_sync_edge_provider(monkeypatch) + + with stream_client.websocket_connect(_url()) as conn: + assert conn.receive_json() == {"type": "fallback"} + + +def test_sync_provider_failure_before_audio_requests_fallback(stream_client, monkeypatch): + _patch_sync_edge_provider(monkeypatch) + monkeypatch.setattr( + "tools.tts_tool.text_to_speech_tool", + lambda **_kwargs: json.dumps({"success": False, "error": "synthetic failure"}), + ) + + with stream_client.websocket_connect(_url(audio_protocol=2)) as conn: + assert conn.receive_json() == {"type": "start", "encoding": "encoded"} + conn.send_text(json.dumps({"text": "This will fail.", "done": True})) + assert conn.receive_json() == {"type": "fallback"} + + +def test_sync_provider_failure_after_audio_ends_without_replaying(stream_client, monkeypatch): + _patch_sync_edge_provider(monkeypatch) + calls = 0 + + def fail_second_sentence(*, text, output_path, provider): + nonlocal calls + calls += 1 + if calls == 2: + return json.dumps({"success": False, "error": "synthetic failure"}) + path = Path(output_path) + path.write_bytes(f"audio:{text}".encode()) + return json.dumps({"success": True, "file_path": str(path), "provider": provider}) + + monkeypatch.setattr("tools.tts_tool.text_to_speech_tool", fail_second_sentence) + + with stream_client.websocket_connect(_url(audio_protocol=2)) as conn: + assert conn.receive_json() == {"type": "start", "encoding": "encoded"} + conn.send_text(json.dumps({"text": "The first sentence works. "})) + assert conn.receive_bytes() == b"audio:The first sentence works." + conn.send_text(json.dumps({"text": "The second sentence fails.", "done": True})) + assert conn.receive_json() == {"type": "end"} + +