Spaces:
Running
Running
| """Speech-only WebSocket. Audio in, transcript or Piper audio out. No language model. | |
| Protocol ``sonic-speech-stream-v1``. Everything a v1 client sends and reads is | |
| unchanged; what a session negotiated through ``stream_options`` (see | |
| ``speech_stream.store.OPTION_VALUES``) is additive and off unless asked for. | |
| HV3 hardening, and the failure each part prevents: | |
| * **Backpressure counted what was SENT, not what was waiting.** ``queued_ms`` | |
| grew by one frame per frame and was never reduced, so the limit was a cap on | |
| the length of a reply: with the default 2000 ms, every Piper answer longer | |
| than two seconds died mid-sentence with "outbound audio queue is full". The | |
| limit now bounds audio *ahead of playback* — wall-clock pacing by default, | |
| the client's own ``playback.ack`` when it negotiated ``flow_control: ack`` — | |
| and the server waits when it is ahead instead of ending a healthy turn. | |
| * **One slot for transcription and speech.** A user who spoke while she was | |
| answering had the utterance refused ("a speech turn is already running") in | |
| ``duplex_speech``. Recognition and synthesis now have their own slots. | |
| * **A cancel left the cancelled turn holding the slot** until its Piper thread | |
| noticed, so the ``tts.speak`` that follows a barge-in was refused. A cancel | |
| now releases the slot at once; the old thread finishes unobserved. | |
| * **Input was not checked against what the server decodes.** A 48 kHz frame | |
| was accepted and transcribed as 16 kHz (three times slow, garbage text), and | |
| an odd byte count crashed the PCM decoder. Both are refused with a reason. | |
| """ | |
| from __future__ import annotations | |
| import asyncio | |
| import contextlib | |
| import json | |
| import threading | |
| import time | |
| from collections import deque | |
| from collections.abc import Awaitable, Callable | |
| from uuid import uuid4 | |
| from fastapi import APIRouter, WebSocket, WebSocketDisconnect | |
| from starlette.websockets import WebSocketDisconnected | |
| from sonicagent_rt.audio.pcm import pcm16_to_floats | |
| from sonicagent_rt.auth.service_token import enforce_origin | |
| from sonicagent_rt.config import Settings | |
| from sonicagent_rt.pipeline.cancellation import CancellationToken | |
| from sonicagent_rt.protocol.binary_frame import BinaryAudioFrame, BinaryFrameError, decode_frame, encode_frame | |
| from sonicagent_rt.protocol.constants import FLAG_END_OF_UTTERANCE, Codec, MessageType | |
| from sonicagent_rt.protocol.sequence import SequenceError, SequenceValidator | |
| from sonicagent_rt.speech_stream.chunks import iter_pcm16, split_frames | |
| from sonicagent_rt.speech_stream.store import MAX_TTS_QUEUE, SpeechStreamSession | |
| from sonicagent_rt.stt.errors import Cancelled, SpeechError | |
| from sonicagent_rt.tts.models import validate_voice_id, voice_record | |
| from sonicagent_rt.vad.factory import create_endpoint | |
| router = APIRouter() | |
| _MAX_FRAME_BYTES = 64 * 1024 | |
| #: Longest single input frame accepted. The contract is 20–40 ms; 200 ms leaves | |
| #: room for a client that batches under load without letting one frame carry a | |
| #: whole utterance past the per-utterance limit. | |
| _MAX_FRAME_MS = 200 | |
| #: Transcriptions allowed to wait behind the running one (client endpointing). | |
| _MAX_STT_PENDING = 2 | |
| #: With ``flow_control: ack``, how long the server waits for playback to move | |
| #: before it concludes the client has stopped reading. | |
| _ACK_STALL_SEC = 10.0 | |
| _PACE_STEP_SEC = 0.02 | |
| Emit = Callable[[dict[str, object]], Awaitable[None]] | |
| EmitBytes = Callable[[bytes], Awaitable[None]] | |
| async def speech_stream(websocket: WebSocket, session_id: str) -> None: | |
| settings: Settings = websocket.app.state.settings | |
| token = websocket.query_params.get("resume_token", "") | |
| try: | |
| if not settings.voice_enabled: | |
| raise SpeechError("voice endpoints are disabled") | |
| enforce_origin(websocket.headers.get("origin"), settings) | |
| session = websocket.app.state.speech_streams.acquire(session_id, token) | |
| except SpeechError: | |
| await websocket.close(code=1008) | |
| return | |
| await websocket.accept() | |
| opened = time.monotonic() | |
| send_lock = asyncio.Lock() | |
| async def emit(payload: dict[str, object]) -> None: | |
| payload.setdefault("t_ms", int((time.monotonic() - opened) * 1000)) | |
| async with send_lock: | |
| await websocket.send_json(payload) | |
| async def emit_bytes(payload: bytes) -> None: | |
| async with send_lock: | |
| await websocket.send_bytes(payload) | |
| await emit( | |
| { | |
| "type": "session.ready", | |
| "protocol": "sonic-speech-stream-v1", | |
| "session_id": session.session_id, | |
| "mode": session.mode, | |
| "llm": False, | |
| "partial_results": False, | |
| "options": session.options.as_dict(), | |
| } | |
| ) | |
| stream = _Stream(websocket, session, settings, emit, emit_bytes) | |
| try: | |
| while True: | |
| message = await websocket.receive() | |
| if message.get("type") == "websocket.disconnect": | |
| break | |
| if message.get("text"): | |
| if await stream.on_text(message["text"]): | |
| await websocket.close(code=1000) | |
| break | |
| elif message.get("bytes") is not None: | |
| await stream.on_audio(message["bytes"]) | |
| except (WebSocketDisconnect, WebSocketDisconnected, RuntimeError): | |
| pass | |
| finally: | |
| await stream.close() | |
| websocket.app.state.speech_streams.release(session_id) | |
| class _Slot: | |
| """One kind of work (recognition or synthesis): its task and its cancellation.""" | |
| def __init__(self) -> None: | |
| self.task: asyncio.Task[None] | None = None | |
| self.cancellation = CancellationToken() | |
| self.stop = threading.Event() | |
| def busy(self) -> bool: | |
| return self.task is not None and not self.task.done() | |
| def fresh(self) -> tuple[CancellationToken, threading.Event]: | |
| self.cancellation = CancellationToken() | |
| self.stop = threading.Event() | |
| return self.cancellation, self.stop | |
| def cancel(self) -> None: | |
| """Stop the running work and free the slot NOW. | |
| The worker thread (Piper, faster-whisper) is told through ``stop`` and | |
| finishes on its own; the asyncio task is cancelled so it emits nothing | |
| more, and the slot is free for the next request in the same tick. | |
| """ | |
| self.stop.set() | |
| self.cancellation.cancel() | |
| if self.task is not None and not self.task.done(): | |
| self.task.cancel() | |
| self.task = None | |
| async def drain(self) -> None: | |
| task = self.task | |
| self.cancel() | |
| if task is not None: | |
| with contextlib.suppress(asyncio.CancelledError, Exception): | |
| await task | |
| class _Stream: | |
| def __init__( | |
| self, | |
| websocket: WebSocket, | |
| session: SpeechStreamSession, | |
| settings: Settings, | |
| emit: Emit, | |
| emit_bytes: EmitBytes, | |
| ) -> None: | |
| self.websocket = websocket | |
| self.session = session | |
| self.settings = settings | |
| self.options = session.options | |
| self.emit = emit | |
| self.emit_bytes = emit_bytes | |
| self.endpoint = create_endpoint(settings) | |
| self.sequences = SequenceValidator() | |
| self.started_audio = False | |
| self.stt = _Slot() | |
| self.tts = _Slot() | |
| self.stt_pending: deque[tuple[list[float], int | None]] = deque() | |
| self.tts_pending: deque[tuple[str, str, str]] = deque() | |
| # Client endpointing: the utterance being held until its commit. | |
| self.utterance: list[float] = [] | |
| self.utterance_speech = False | |
| self.max_utterance_samples = int( | |
| min(settings.vad_max_utterance_sec, settings.max_audio_seconds) * settings.input_sample_rate | |
| ) | |
| # Audio sent and not yet played, for the WHOLE session — not per phrase. With the queue, | |
| # phrases follow each other into one playback buffer on the client, so a per-phrase | |
| # budget let a headset hold ~2 s of unplayed audio *per queued phrase* (measured end to | |
| # end: 6 s of speech delivered in 2.2 s against a 2 s budget). | |
| # - flow_control none: a virtual playout clock — when the last sent sample will have | |
| # played if the client plays continuously from now on. | |
| # - flow_control ack: per phrase [sent_ms, played_ms], summed; a phrase leaves the | |
| # table once the client reports it fully played. | |
| self.playout_end = 0.0 | |
| self.unplayed: dict[str, list[float]] = {} | |
| self.last_ack_at: float | None = None | |
| self.current_tts: str | None = None | |
| # ── Control events ─────────────────────────────────────────────── | |
| async def on_text(self, raw: str) -> bool: | |
| try: | |
| event = json.loads(raw) | |
| except json.JSONDecodeError: | |
| await self.emit( | |
| {"type": "session.error", "code": "invalid_request", "message": "control event is not JSON"} | |
| ) | |
| return False | |
| if not isinstance(event, dict): | |
| await self.emit( | |
| { | |
| "type": "session.error", | |
| "code": "invalid_request", | |
| "message": "control event is not an object", | |
| } | |
| ) | |
| return False | |
| kind = event.get("type") | |
| if kind == "ping": | |
| await self.emit({"type": "pong"}) | |
| elif kind == "turn.cancel": | |
| await self._on_cancel(event) | |
| elif kind == "session.close": | |
| self.stt.cancel() | |
| self.tts.cancel() | |
| self.stt_pending.clear() | |
| self.tts_pending.clear() | |
| await self.emit({"type": "session.closed"}) | |
| return True | |
| elif kind == "tts.speak": | |
| await self._on_speak(event) | |
| elif kind == "playback.ack": | |
| self._on_ack(event) | |
| elif kind == "tts.warm": | |
| await self._on_warm(event) | |
| else: | |
| await self.emit( | |
| {"type": "session.error", "code": "invalid_request", "message": "unsupported control event"} | |
| ) | |
| return False | |
| async def _on_cancel(self, event: dict[str, object]) -> None: | |
| # v1 had one slot, so a cancel stopped whatever was running. ``target`` | |
| # narrows it; without one both stop, which is the v1 behaviour. | |
| target = str(event.get("target") or "all").lower() | |
| if target not in ("all", "stt", "tts"): | |
| await self.emit( | |
| { | |
| "type": "session.error", | |
| "code": "invalid_request", | |
| "message": "target must be all, stt or tts", | |
| } | |
| ) | |
| return | |
| if target in ("all", "tts"): | |
| self.tts.cancel() | |
| self.tts_pending.clear() | |
| self._forget_playback() | |
| if target in ("all", "stt"): | |
| self.stt.cancel() | |
| self.stt_pending.clear() | |
| self.utterance = [] | |
| self.utterance_speech = False | |
| with contextlib.suppress(Exception): | |
| self.endpoint.flush() | |
| await self.emit( | |
| { | |
| "type": "turn.cancelled", | |
| "turn_id": event.get("turn_id"), | |
| "target": target, | |
| "clear_audio": target in ("all", "tts"), | |
| } | |
| ) | |
| async def _on_speak(self, event: dict[str, object]) -> None: | |
| if self.session.mode == "stt_only": | |
| await self.emit( | |
| {"type": "session.error", "code": "invalid_request", "message": "this session does not speak"} | |
| ) | |
| return | |
| text = str(event.get("text") or "").strip() | |
| turn_id = str(event.get("turn_id") or f"turn_{uuid4().hex}") | |
| if not text: | |
| await self.emit( | |
| { | |
| "type": "session.error", | |
| "turn_id": turn_id, | |
| "code": "invalid_request", | |
| "message": "text is empty", | |
| } | |
| ) | |
| return | |
| # HV9. A phrase may name its own voice (a preview, or Companion <-> Secretary without a | |
| # new session). Only a registry voice; never a silent substitute for one that is not there. | |
| voice = self.session.voice_id | |
| if event.get("voice_id"): | |
| voice = str(event.get("voice_id")).strip() | |
| try: | |
| validate_voice_id(voice) | |
| except SpeechError: | |
| voice = "" | |
| if not voice or voice_record(voice) is None: | |
| await self.emit( | |
| { | |
| "type": "session.error", | |
| "turn_id": turn_id, | |
| "code": "voice_unavailable", | |
| "message": "voice is not in the voice registry", | |
| } | |
| ) | |
| return | |
| if self.tts.busy(): | |
| if self.options.tts_queue and len(self.tts_pending) < MAX_TTS_QUEUE: | |
| self.tts_pending.append((text, turn_id, voice)) | |
| await self.emit({"type": "tts.queued", "turn_id": turn_id, "position": len(self.tts_pending)}) | |
| return | |
| await self.emit( | |
| { | |
| "type": "session.error", | |
| "turn_id": turn_id, | |
| "code": "rate_limited", | |
| "message": "a speech turn is already running", | |
| } | |
| ) | |
| return | |
| self._start_speech(text, turn_id, voice) | |
| def _start_speech(self, text: str, turn_id: str, voice: str) -> None: | |
| cancellation, stop = self.tts.fresh() | |
| self.tts.task = asyncio.create_task(self._speak_then_next(text, turn_id, voice, cancellation, stop)) | |
| async def _speak_then_next( | |
| self, text: str, turn_id: str, voice: str, cancellation: CancellationToken, stop: threading.Event | |
| ) -> None: | |
| await self._speak(text, turn_id, voice, cancellation, stop) | |
| if cancellation.cancelled or stop.is_set() or not self.tts_pending: | |
| return | |
| next_text, next_turn, next_voice = self.tts_pending.popleft() | |
| # Started from inside the finishing task: the slot is handed straight | |
| # over, so no ``tts.speak`` arriving in between can jump the queue. | |
| cancellation, stop = self.tts.fresh() | |
| self.tts.task = asyncio.create_task( | |
| self._speak_then_next(next_text, next_turn, next_voice, cancellation, stop) | |
| ) | |
| async def _on_warm(self, event: dict[str, object]) -> None: | |
| """HV9. Load the selected voice's model before her first sentence needs it.""" | |
| voice = str(event.get("voice_id") or self.session.voice_id).strip() | |
| try: | |
| validate_voice_id(voice) | |
| if voice_record(voice) is None: | |
| raise SpeechError("voice is not in the voice registry") | |
| warm = getattr(self.websocket.app.state.tts, "warm", None) | |
| if not callable(warm): | |
| raise SpeechError("this speech engine cannot preload voices") | |
| result = await asyncio.to_thread(warm, voice) | |
| except SpeechError as exc: | |
| await self.emit({"type": "session.error", "code": "voice_unavailable", "message": exc.message}) | |
| return | |
| await self.emit({"type": "tts.warmed", **result}) | |
| def _on_ack(self, event: dict[str, object]) -> None: | |
| if self.options.flow_control != "ack": | |
| return # v1 clients never send it; a stray one is harmless | |
| turn_id = str(event.get("turn_id") or "") | |
| try: | |
| played = float(event.get("played_ms")) # type: ignore[arg-type] | |
| except (TypeError, ValueError): | |
| return | |
| entry = self.unplayed.get(turn_id) | |
| if entry is None: | |
| return | |
| if played >= entry[1]: | |
| entry[1] = played | |
| self.last_ack_at = time.monotonic() | |
| if played > 0: | |
| # Playback is sequential: a phrase that has started playing means every phrase | |
| # sent before it has finished. Without this, a client that acknowledges only the | |
| # phrase it is receiving leaves the previous phrase's tail in the budget forever. | |
| for earlier in list(self.unplayed): | |
| if earlier == turn_id: | |
| break | |
| self.unplayed.pop(earlier, None) | |
| # Fully played and no longer being sent: it no longer holds any of the budget. | |
| if entry[1] >= entry[0] and not (self.tts.busy() and self.current_tts == turn_id): | |
| self.unplayed.pop(turn_id, None) | |
| def _ahead_ms(self) -> float: | |
| """Audio handed to the client and not yet played, across every phrase.""" | |
| if self.options.flow_control == "ack": | |
| return sum(max(0.0, sent - played) for sent, played in self.unplayed.values()) | |
| return max(0.0, (self.playout_end - time.monotonic()) * 1000.0) | |
| def _note_sent(self, turn_id: str, ms: float) -> None: | |
| if self.options.flow_control == "ack": | |
| self.unplayed.setdefault(turn_id, [0.0, 0.0])[0] += ms | |
| else: | |
| self.playout_end = max(self.playout_end, time.monotonic()) + ms / 1000.0 | |
| def _forget_playback(self) -> None: | |
| """The client cleared its audio (a tts cancel): nothing it held is waiting any more.""" | |
| self.unplayed.clear() | |
| self.playout_end = 0.0 | |
| # ── Microphone audio ───────────────────────────────────────────── | |
| async def on_audio(self, payload: bytes) -> None: | |
| if self.session.mode == "tts_only": | |
| await self.emit( | |
| { | |
| "type": "session.error", | |
| "code": "invalid_request", | |
| "message": "this session does not accept audio", | |
| } | |
| ) | |
| return | |
| if len(payload) > _MAX_FRAME_BYTES: | |
| await self.emit( | |
| {"type": "session.error", "code": "audio_too_large", "message": "audio frame is too large"} | |
| ) | |
| return | |
| try: | |
| frame = decode_frame(payload) | |
| if frame.message_type != int(MessageType.AUDIO_INPUT): | |
| raise BinaryFrameError("client frames must be microphone audio") | |
| if frame.sample_rate != self.settings.input_sample_rate: | |
| raise BinaryFrameError( | |
| f"unsupported sample rate {frame.sample_rate}: send mono PCM16 at " | |
| f"{self.settings.input_sample_rate} Hz" | |
| ) | |
| if len(frame.payload) % 2: | |
| raise BinaryFrameError("PCM16 payload must be a whole number of 16-bit samples") | |
| if len(frame.payload) // 2 > self.settings.input_sample_rate * _MAX_FRAME_MS // 1000: | |
| raise BinaryFrameError(f"audio frame is longer than {_MAX_FRAME_MS} ms") | |
| self.sequences.accept(frame.sequence) | |
| samples = pcm16_to_floats(frame.payload) | |
| except (BinaryFrameError, SequenceError, ValueError) as exc: | |
| await self.emit({"type": "session.error", "code": "invalid_audio", "message": str(exc)}) | |
| return | |
| if not self.started_audio: | |
| self.started_audio = True | |
| await self.emit({"type": "audio.input.started"}) | |
| committed = bool(frame.flags & FLAG_END_OF_UTTERANCE) | |
| if self.options.endpointing == "client": | |
| await self._client_endpointed(samples, committed, frame.sequence) | |
| return | |
| for event in self.endpoint.push(samples): | |
| if event.kind == "started": | |
| await self.emit({"type": "speech.started"}) | |
| elif event.kind == "utterance": | |
| await self._transcribe(event.samples, None) | |
| if committed: | |
| await self.emit({"type": "audio.input.committed", "commit_seq": frame.sequence}) | |
| for event in self.endpoint.flush(): | |
| if event.kind == "utterance": | |
| await self._transcribe(event.samples, frame.sequence) | |
| async def _client_endpointed(self, samples: list[float], committed: bool, sequence: int) -> None: | |
| # The VAD still runs, but only to say "speech.started" for the UI; the | |
| # client's commit decides where the utterance ends. | |
| for event in self.endpoint.push(samples): | |
| if event.kind == "started" and not self.utterance_speech: | |
| self.utterance_speech = True | |
| await self.emit({"type": "speech.started"}) | |
| self.utterance.extend(samples) | |
| if len(self.utterance) > self.max_utterance_samples: | |
| self.utterance = [] | |
| self.utterance_speech = False | |
| with contextlib.suppress(Exception): | |
| self.endpoint.flush() | |
| await self.emit( | |
| { | |
| "type": "session.error", | |
| "code": "audio_too_long", | |
| "commit_seq": None, | |
| "message": "utterance is longer than the server accepts; it was discarded", | |
| } | |
| ) | |
| return | |
| if not committed: | |
| return | |
| held, self.utterance = self.utterance, [] | |
| self.utterance_speech = False | |
| with contextlib.suppress(Exception): | |
| self.endpoint.flush() | |
| await self.emit({"type": "audio.input.committed", "commit_seq": sequence}) | |
| await self._transcribe(held, sequence) | |
| async def _transcribe(self, samples: list[float], commit_seq: int | None) -> None: | |
| if self.stt.busy(): | |
| if self.options.endpointing == "client" and len(self.stt_pending) < _MAX_STT_PENDING: | |
| self.stt_pending.append((samples, commit_seq)) | |
| return | |
| await self.emit( | |
| { | |
| "type": "session.error", | |
| "code": "rate_limited", | |
| "commit_seq": commit_seq, | |
| "message": "a speech turn is already running", | |
| } | |
| ) | |
| return | |
| self._start_transcription(samples, commit_seq) | |
| def _start_transcription(self, samples: list[float], commit_seq: int | None) -> None: | |
| cancellation, stop = self.stt.fresh() | |
| self.stt.task = asyncio.create_task( | |
| self._transcribe_then_next(samples, commit_seq, cancellation, stop) | |
| ) | |
| async def _transcribe_then_next( | |
| self, | |
| samples: list[float], | |
| commit_seq: int | None, | |
| cancellation: CancellationToken, | |
| stop: threading.Event, | |
| ) -> None: | |
| await self._finish_utterance(samples, commit_seq, cancellation, stop) | |
| if cancellation.cancelled or stop.is_set() or not self.stt_pending: | |
| return | |
| next_samples, next_seq = self.stt_pending.popleft() | |
| cancellation, stop = self.stt.fresh() | |
| self.stt.task = asyncio.create_task( | |
| self._transcribe_then_next(next_samples, next_seq, cancellation, stop) | |
| ) | |
| async def _finish_utterance( | |
| self, | |
| samples: list[float], | |
| commit_seq: int | None, | |
| cancellation: CancellationToken, | |
| stop: threading.Event, | |
| ) -> None: | |
| turn_id = f"turn_{uuid4().hex}" | |
| audio_ms = round(len(samples) * 1000 / self.settings.input_sample_rate) | |
| tag: dict[str, object] = {} if commit_seq is None else {"commit_seq": commit_seq} | |
| try: | |
| await self.emit({"type": "speech.stopped", "turn_id": turn_id, **tag}) | |
| if cancellation.cancelled or stop.is_set(): | |
| return | |
| if not samples: | |
| # A commit with no audio still answers, so the client is never | |
| # left waiting for a final that is not coming. | |
| await self.emit( | |
| { | |
| "type": "transcript.final", | |
| "turn_id": turn_id, | |
| "text": "", | |
| "language": self.session.language, | |
| "is_final": True, | |
| "no_speech": True, | |
| "audio_ms": 0, | |
| **tag, | |
| } | |
| ) | |
| return | |
| started = time.perf_counter() | |
| transcript = await asyncio.to_thread( | |
| self.websocket.app.state.stt.transcribe, | |
| samples, | |
| sample_rate=self.settings.input_sample_rate, | |
| language=self.session.language, | |
| model=self.settings.stt_model, | |
| request_id=turn_id, | |
| cancel=stop, | |
| ) | |
| if cancellation.cancelled or stop.is_set(): | |
| return | |
| elapsed = round((time.perf_counter() - started) * 1000, 1) | |
| await self.emit( | |
| { | |
| "type": "transcript.final", | |
| "turn_id": turn_id, | |
| "text": transcript.text, | |
| "language": transcript.language, | |
| "is_final": True, | |
| "audio_ms": audio_ms, | |
| **tag, | |
| } | |
| ) | |
| await self.emit( | |
| { | |
| "type": "metrics.turn", | |
| "turn_id": turn_id, | |
| "stt_ms": elapsed, | |
| "audio_ms": audio_ms, | |
| "partials": 0, | |
| "llm": False, | |
| **tag, | |
| } | |
| ) | |
| except Cancelled: | |
| return | |
| except asyncio.CancelledError: | |
| raise | |
| except SpeechError as exc: | |
| await self.emit( | |
| {"type": "session.error", "turn_id": turn_id, "code": exc.code, "message": exc.message, **tag} | |
| ) | |
| except (WebSocketDisconnect, WebSocketDisconnected): | |
| return | |
| # ── Synthesis ──────────────────────────────────────────────────── | |
| async def _speak( | |
| self, text: str, turn_id: str, voice: str, cancellation: CancellationToken, stop: threading.Event | |
| ) -> None: | |
| settings = self.settings | |
| frame_ms = int(self.options.output_frame_ms) | |
| limit_ms = settings.max_outbound_queue_ms | |
| started = time.perf_counter() | |
| first_ms: float | None = None | |
| sequence = 0 | |
| sent_ms = 0.0 | |
| self.current_tts = turn_id | |
| synth_s = 0.0 | |
| engine = self.websocket.app.state.tts | |
| is_loaded = getattr(engine, "is_loaded", None) | |
| cold = None | |
| if callable(is_loaded): | |
| with contextlib.suppress(Exception): | |
| cold = not is_loaded(voice) | |
| record = voice_record(voice) | |
| await self.emit( | |
| { | |
| "type": "tts.started", | |
| "turn_id": turn_id, | |
| "voice_id": voice, | |
| "role": record.role if record else None, | |
| "cold": cold, | |
| } | |
| ) | |
| iterator = iter( | |
| iter_pcm16( | |
| engine, | |
| text, | |
| voice=voice, | |
| language=self.session.language, | |
| cancel=stop, | |
| ) | |
| ) | |
| def pull() -> tuple[int, bytes] | None: | |
| try: | |
| return next(iterator) | |
| except StopIteration: | |
| return None | |
| try: | |
| while True: | |
| if cancellation.cancelled or stop.is_set(): | |
| return | |
| t_pull = time.perf_counter() | |
| chunk = await asyncio.to_thread(pull) | |
| synth_s += time.perf_counter() - t_pull | |
| if chunk is None: | |
| break | |
| rate, pcm = chunk | |
| if first_ms is None: | |
| first_ms = round((time.perf_counter() - started) * 1000, 1) | |
| await self.emit( | |
| { | |
| "type": "tts.audio.meta", | |
| "turn_id": turn_id, | |
| "sample_rate": rate, | |
| "codec": "pcm_s16le", | |
| "incremental": "sentence", | |
| "frame_ms": frame_ms, | |
| } | |
| ) | |
| for piece in split_frames(pcm, rate, frame_ms): | |
| if cancellation.cancelled or stop.is_set(): | |
| return | |
| if limit_ms <= 0: | |
| # A zero budget cannot hold even one frame: refuse, as v1 did. | |
| await self._refuse_queue(turn_id, cancellation, stop, "outbound audio queue is full") | |
| return | |
| if not await self._wait_for_room(turn_id, limit_ms, cancellation, stop): | |
| return | |
| timestamp = sequence * frame_ms * 1000 | |
| if self.options.chunk_events == "all": | |
| await self.emit( | |
| { | |
| "type": "tts.audio.chunk", | |
| "turn_id": turn_id, | |
| "sequence": sequence, | |
| "sample_rate": rate, | |
| "timestamp_us": timestamp, | |
| "bytes": len(piece), | |
| } | |
| ) | |
| await self.emit_bytes( | |
| encode_frame( | |
| BinaryAudioFrame( | |
| message_type=int(MessageType.AUDIO_OUTPUT), | |
| codec=int(Codec.PCM16), | |
| flags=0, | |
| sample_rate=rate, | |
| sequence=sequence, | |
| timestamp_us=timestamp, | |
| payload=piece, | |
| ) | |
| ) | |
| ) | |
| piece_ms = len(piece) / 2 / rate * 1000.0 | |
| sent_ms += piece_ms | |
| self._note_sent(turn_id, piece_ms) | |
| sequence += 1 | |
| except Cancelled: | |
| return | |
| except SpeechError as exc: | |
| await self.emit( | |
| {"type": "session.error", "turn_id": turn_id, "code": exc.code, "message": exc.message} | |
| ) | |
| return | |
| except (WebSocketDisconnect, WebSocketDisconnected): | |
| return | |
| await self.emit( | |
| {"type": "tts.audio.done", "turn_id": turn_id, "frames": sequence, "audio_ms": round(sent_ms)} | |
| ) | |
| await self.emit( | |
| { | |
| "type": "metrics.turn", | |
| "turn_id": turn_id, | |
| "tts_first_ms": first_ms, | |
| "frames": sequence, | |
| "audio_ms": round(sent_ms), | |
| # HV9. Per voice and profile: time in synthesis (a cold load included, on the | |
| # first sentence), its real-time factor, and whether the model was cold. | |
| "voice_id": voice, | |
| "role": record.role if record else None, | |
| "cold": cold, | |
| "synth_ms": round(synth_s * 1000, 1), | |
| "rtf": round(synth_s * 1000 / sent_ms, 3) if sent_ms else None, | |
| "llm": False, | |
| } | |
| ) | |
| async def _wait_for_room( | |
| self, | |
| turn_id: str, | |
| limit_ms: int, | |
| cancellation: CancellationToken, | |
| stop: threading.Event, | |
| ) -> bool: | |
| """Hold the next frame while the client has ``limit_ms`` or more unplayed. | |
| Pacing, not refusing: the reply is healthy, the client is simply not | |
| there yet. Only a client that negotiated acks and then stopped | |
| acknowledging for ``_ACK_STALL_SEC`` is treated as gone. | |
| """ | |
| waiting_since = time.monotonic() | |
| while self._ahead_ms() >= limit_ms: | |
| if cancellation.cancelled or stop.is_set(): | |
| return False | |
| if self.options.flow_control == "ack": | |
| last = self.last_ack_at or waiting_since | |
| if time.monotonic() - max(last, waiting_since) > _ACK_STALL_SEC: | |
| await self._refuse_queue( | |
| turn_id, cancellation, stop, "client stopped acknowledging playback" | |
| ) | |
| return False | |
| await asyncio.sleep(_PACE_STEP_SEC) | |
| return True | |
| async def _refuse_queue( | |
| self, turn_id: str, cancellation: CancellationToken, stop: threading.Event, message: str | |
| ) -> None: | |
| cancellation.cancel() | |
| stop.set() | |
| await self.emit( | |
| {"type": "session.error", "turn_id": turn_id, "code": "rate_limited", "message": message} | |
| ) | |
| async def close(self) -> None: | |
| self.stt_pending.clear() | |
| self.tts_pending.clear() | |
| await self.stt.drain() | |
| await self.tts.drain() | |