diff --git a/dockerfiles/Dockerfile.hermes-webui b/dockerfiles/Dockerfile.hermes-webui index 345d838c..be26f5de 100644 --- a/dockerfiles/Dockerfile.hermes-webui +++ b/dockerfiles/Dockerfile.hermes-webui @@ -17,6 +17,7 @@ COPY dockerfiles/hermes-webui-atlas-patch.py /tmp/hermes-webui-atlas-patch.py COPY dockerfiles/hermes-webui-stt-patch.py /tmp/hermes-webui-stt-patch.py COPY dockerfiles/hermes-webui-telegram-project-patch.py /tmp/hermes-webui-telegram-project-patch.py COPY dockerfiles/hermes-webui-atlas-voice.js /opt/hermes-webui/static/atlas-voice.js +COPY dockerfiles/hermes-webui-atlas-voice-worklet.js /opt/hermes-webui/static/atlas-voice-worklet.js COPY dockerfiles/hermes-webui-atlas-voice.css /opt/hermes-webui/static/atlas-voice.css COPY dockerfiles/hermes-webui-router-patch.py /tmp/hermes-webui-router-patch.py COPY dockerfiles/hermes-webui-router.js /opt/hermes-webui/static/atlas-router.js @@ -60,6 +61,9 @@ RUN /opt/hermes/.venv/bin/python -c 'import cryptography, yaml' \ && grep -Fq 'def _atlas_tts_language(body):' /opt/hermes-webui/api/routes.py \ && grep -Fq 'request_payload["language"] = _atlas_language' /opt/hermes-webui/api/routes.py \ && grep -Fq 'takeSttLanguage(token)' /opt/hermes-webui/static/atlas-voice.js \ + && grep -Fq "registerProcessor('atlas-pcm-playback'" /opt/hermes-webui/static/atlas-voice-worklet.js \ + && grep -Fq "const TTS_STREAM_URL='/api/tts/stream'" /opt/hermes-webui/static/atlas-voice.js \ + && grep -Fq "const STT_STREAM_PATH='/api/transcribe/stream'" /opt/hermes-webui/static/atlas-voice.js \ && grep -Fq 'Hermes Chat' /opt/hermes-webui/static/index.html \ && grep -Fq 'id="hermesBrandStyles"' /opt/hermes-webui/static/index.html \ && grep -Fq 'static/hermes-agent-512.png' /opt/hermes-webui/static/index.html \ diff --git a/dockerfiles/hermes-jetson-stt-server.py b/dockerfiles/hermes-jetson-stt-server.py index 6e0efd08..76722cb2 100644 --- a/dockerfiles/hermes-jetson-stt-server.py +++ b/dockerfiles/hermes-jetson-stt-server.py @@ -4,11 +4,17 @@ from __future__ import annotations import cgi +import base64 +import hashlib import json import os import re +import socket +import struct import tempfile import threading +import time +import wave from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from pathlib import Path @@ -21,7 +27,20 @@ PORT = int(os.getenv("HERMES_STT_PORT", "9000")) MODEL_NAME = os.getenv("HERMES_STT_MODEL", "small") CACHE_DIR = Path(os.getenv("HERMES_STT_CACHE", "/cache/whisper")) MAX_AUDIO_BYTES = 30 * 1024 * 1024 +STREAM_SAMPLE_RATE = 16_000 +STREAM_SAMPLE_WIDTH = 2 +STREAM_CHANNELS = 1 +MAX_STREAM_SECONDS = int(os.getenv("HERMES_STT_STREAM_MAX_SECONDS", "90")) +MAX_STREAM_AUDIO_BYTES = min( + MAX_AUDIO_BYTES, + STREAM_SAMPLE_RATE * STREAM_SAMPLE_WIDTH * MAX_STREAM_SECONDS, +) +MAX_WS_MESSAGE_BYTES = 256 * 1024 +STREAM_IDLE_SECONDS = int(os.getenv("HERMES_STT_STREAM_IDLE_SECONDS", "120")) +VAD_END_SILENCE_MS = int(os.getenv("HERMES_STT_VAD_END_SILENCE_MS", "650")) +VAD_TAIL_MS = int(os.getenv("HERMES_STT_VAD_TAIL_MS", "220")) MODEL_LOCK = threading.Lock() +_WS_GUID = "258EAFA5-E914-47DA-95CA-C5AB0DC85B11" def _repetitive_token(token: str) -> bool: @@ -37,7 +56,10 @@ def _repetitive_token(token: str) -> bool: ) if matches / len(letters) >= 0.86: return True - return max(letters.count(character) for character in set(letters)) / len(letters) >= 0.78 + return ( + max(letters.count(character) for character in set(letters)) / len(letters) + >= 0.78 + ) def _clean_transcript(result: dict) -> str: @@ -90,6 +112,392 @@ def _transcription_payload(result: dict) -> dict: } +def _transcribe_path(model: object, path: str, language: str) -> dict: + """Run the one canonical full-utterance Whisper decode configuration.""" + with MODEL_LOCK: + result = model.transcribe( + path, + language=None if language in {"", "auto"} else language, + task="transcribe", + fp16=torch.cuda.is_available(), + condition_on_previous_text=False, + temperature=0, + compression_ratio_threshold=2.0, + logprob_threshold=-0.8, + no_speech_threshold=0.5, + verbose=False, + ) + return _transcription_payload(result) + + +def _transcribe_pcm(model: object, pcm: bytes, language: str) -> dict: + """Decode one complete 16-kHz mono PCM snapshot through the canonical path.""" + temp_path = "" + try: + with tempfile.NamedTemporaryFile( + prefix="atlas-stt-stream-", suffix=".wav", delete=False + ) as temp: + temp_path = temp.name + with wave.open(temp_path, "wb") as wav_file: + wav_file.setnchannels(STREAM_CHANNELS) + wav_file.setsampwidth(STREAM_SAMPLE_WIDTH) + wav_file.setframerate(STREAM_SAMPLE_RATE) + wav_file.writeframes(pcm) + return _transcribe_path(model, temp_path, language) + finally: + if temp_path: + try: + os.unlink(temp_path) + except OSError: + pass + + +class WebSocketError(Exception): + """Protocol error carrying an RFC 6455 close code safe to expose.""" + + def __init__(self, message: str, close_code: int = 1002) -> None: + super().__init__(message) + self.close_code = close_code + + +class WebSocketConnection: + """Minimal server-side RFC 6455 framing for the private streaming endpoint.""" + + def __init__(self, handler: BaseHTTPRequestHandler) -> None: + self.handler = handler + self._send_lock = threading.Lock() + self._closed = False + + @staticmethod + def accept_value(key: str) -> str: + """Return the RFC 6455 handshake response for a client nonce.""" + digest = hashlib.sha1((key + _WS_GUID).encode("ascii")).digest() + return base64.b64encode(digest).decode("ascii") + + def _read_exact(self, size: int) -> bytes: + data = bytearray() + while len(data) < size: + chunk = self.handler.rfile.read(size - len(data)) + if not chunk: + raise EOFError("websocket peer disconnected") + data.extend(chunk) + return bytes(data) + + def receive(self) -> tuple[int, bytes]: + """Read and reassemble one client message, answering ping inline.""" + opcode = 0 + payload = bytearray() + while True: + header = self._read_exact(2) + first, second = header + final = bool(first & 0x80) + reserved = first & 0x70 + frame_opcode = first & 0x0F + masked = bool(second & 0x80) + length = second & 0x7F + if reserved or not masked: + raise WebSocketError("invalid websocket frame") + if length == 126: + length = struct.unpack("!H", self._read_exact(2))[0] + elif length == 127: + length = struct.unpack("!Q", self._read_exact(8))[0] + if frame_opcode >= 0x8 and (not final or length > 125): + raise WebSocketError("invalid websocket control frame") + if ( + length > MAX_WS_MESSAGE_BYTES + or len(payload) + length > MAX_WS_MESSAGE_BYTES + ): + raise WebSocketError("websocket message is too large", 1009) + mask = self._read_exact(4) + frame = bytearray(self._read_exact(length)) + for index in range(length): + frame[index] ^= mask[index % 4] + + if frame_opcode == 0x8: + raise EOFError("websocket peer closed") + if frame_opcode == 0x9: + self.send(bytes(frame), opcode=0xA) + continue + if frame_opcode == 0xA: + continue + if frame_opcode in {0x1, 0x2}: + if opcode: + raise WebSocketError("nested websocket message") + opcode = frame_opcode + elif frame_opcode != 0x0 or not opcode: + raise WebSocketError("unexpected websocket continuation") + payload.extend(frame) + if final: + return opcode, bytes(payload) + + def send(self, payload: bytes, opcode: int) -> None: + """Send one unmasked server frame while serializing background writers.""" + with self._send_lock: + if self._closed: + return + length = len(payload) + header = bytearray([0x80 | opcode]) + if length < 126: + header.append(length) + elif length <= 0xFFFF: + header.extend((126,)) + header.extend(struct.pack("!H", length)) + else: + header.extend((127,)) + header.extend(struct.pack("!Q", length)) + self.handler.wfile.write(bytes(header) + payload) + self.handler.wfile.flush() + + def send_json(self, payload: dict) -> None: + """Send a compact UTF-8 JSON message.""" + self.send(json.dumps(payload, separators=(",", ":")).encode("utf-8"), 0x1) + + def close(self, code: int = 1000, reason: str = "") -> None: + """Close once; callers may safely race disconnect and worker completion.""" + reason_bytes = reason.encode("utf-8")[:123] + with self._send_lock: + if self._closed: + return + try: + payload = struct.pack("!H", code) + reason_bytes + self.handler.wfile.write(bytes([0x88, len(payload)]) + payload) + self.handler.wfile.flush() + except (BrokenPipeError, ConnectionError, OSError): + pass + self._closed = True + + +def _pcm_rms(pcm: bytes) -> float: + """Return normalized RMS for aligned signed-16-bit little-endian PCM.""" + sample_count = len(pcm) // 2 + if not sample_count: + return 0.0 + samples = struct.unpack(f"<{sample_count}h", pcm[: sample_count * 2]) + square_mean = sum(sample * sample for sample in samples) / sample_count + return (square_mean**0.5) / 32768.0 + + +class StreamingTranscription: + """Own one turn's bounded PCM, VAD state and speculative Whisper work.""" + + def __init__(self, connection: WebSocketConnection, model: object) -> None: + self.connection = connection + self.model = model + self.turn_id = "" + self.language = "auto" + self._pcm = bytearray() + self._lock = threading.Condition() + self._started = False + self._closed = False + self._committing = False + self._heard_speech = False + self._at_eos = False + self._silence_bytes = 0 + self._last_speech_byte = 0 + self._noise_floor = 0.004 + self._client_active = False + self._epoch = 0 + self._pending_key = "" + self._cached_key = "" + self._cached_payload: dict | None = None + + def _response(self, message_type: str, **values: object) -> dict: + return {"type": message_type, "turn_id": self.turn_id, **values} + + def error(self, message: str, *, fatal: bool = False) -> None: + """Report a turn-scoped error without leaking implementation details.""" + self.connection.send_json(self._response("error", error=message, fatal=fatal)) + + def start(self, message: dict) -> None: + """Validate and initialize the only accepted browser audio contract.""" + turn_id = str(message.get("turn_id") or "") + language = str(message.get("language") or "auto").strip().lower() + if self._started: + raise WebSocketError("stream is already started", 1008) + if not re.fullmatch(r"[A-Za-z0-9_.:-]{1,128}", turn_id): + raise WebSocketError("invalid turn_id", 1008) + if message.get("format") != "pcm_s16le": + raise WebSocketError("pcm_s16le audio is required", 1003) + try: + sample_rate = int(message.get("sample_rate") or 0) + except (TypeError, ValueError) as exc: + raise WebSocketError("invalid sample rate", 1003) from exc + if sample_rate != STREAM_SAMPLE_RATE: + raise WebSocketError("16000 Hz audio is required", 1003) + if language != "auto" and not re.fullmatch(r"[a-z]{2,3}", language): + raise WebSocketError("invalid language", 1008) + self.turn_id = turn_id + self.language = language + self._started = True + + def _snapshot_locked(self) -> tuple[bytes, str]: + """Keep the full utterance and discard only bounded post-speech silence. + + Browser ``resume`` is authoritative: once sent, every following PCM + frame is preserved until the next speculative EOS marker, even when a + quiet syllable falls below the server's conservative energy threshold. + """ + cutoff = len(self._pcm) + if self._heard_speech and self._last_speech_byte: + tail_bytes = STREAM_SAMPLE_RATE * STREAM_SAMPLE_WIDTH * VAD_TAIL_MS // 1000 + cutoff = min(cutoff, self._last_speech_byte + tail_bytes) + pcm = bytes(self._pcm[:cutoff]) + key = f"{self._epoch}:{hashlib.sha256(pcm).hexdigest()}" + return pcm, key + + def append(self, pcm: bytes) -> None: + """Append PCM, track server VAD and begin speculation at probable EOS.""" + if not self._started: + raise WebSocketError("start must precede audio", 1008) + if not pcm or len(pcm) % STREAM_SAMPLE_WIDTH: + raise WebSocketError("unaligned PCM audio", 1003) + auto_speculate = False + with self._lock: + if self._closed or self._committing: + raise WebSocketError("turn is no longer accepting audio", 1008) + if len(self._pcm) + len(pcm) > MAX_STREAM_AUDIO_BYTES: + raise WebSocketError("stream audio is too long", 1009) + self._pcm.extend(pcm) + rms = _pcm_rms(pcm) + threshold = max(0.012, self._noise_floor * 2.5 + 0.004) + speech = self._client_active or rms >= threshold + if not self._heard_speech and not speech: + self._noise_floor = self._noise_floor * 0.96 + rms * 0.04 + if speech: + if self._at_eos: + self._epoch += 1 + self._cached_key = "" + self._cached_payload = None + self._heard_speech = True + self._at_eos = False + self._silence_bytes = 0 + self._last_speech_byte = len(self._pcm) + elif self._heard_speech: + self._silence_bytes += len(pcm) + required = ( + STREAM_SAMPLE_RATE + * STREAM_SAMPLE_WIDTH + * VAD_END_SILENCE_MS + // 1000 + ) + if self._silence_bytes >= required and not self._at_eos: + self._at_eos = True + auto_speculate = True + if auto_speculate: + self.speculate() + + def resume(self) -> None: + """Invalidate an EOS snapshot when client-side VAD hears resumed speech.""" + with self._lock: + if not self._started or self._closed: + return + self._at_eos = False + self._silence_bytes = 0 + self._epoch += 1 + self._client_active = True + self._cached_key = "" + self._cached_payload = None + + def speculate(self) -> None: + """Decode a stable snapshot once; only an unchanged turn may consume it.""" + with self._lock: + if not self._started or self._closed or self._committing or not self._pcm: + return + self._client_active = False + self._at_eos = True + pcm, key = self._snapshot_locked() + if not pcm or key in {self._pending_key, self._cached_key}: + return + self._pending_key = key + + def worker() -> None: + try: + payload = _transcribe_pcm(self.model, pcm, self.language) + except Exception as exc: + print(f"[stt] speculative transcription failed: {exc}", flush=True) + with self._lock: + if self._pending_key == key: + self._pending_key = "" + self._lock.notify_all() + if not self._closed and not self._committing: + self.error("speculative transcription failed") + return + + should_send = False + with self._lock: + if self._pending_key == key: + self._pending_key = "" + current_pcm, current_key = self._snapshot_locked() + del current_pcm + if not self._closed and current_key == key: + self._cached_key = key + self._cached_payload = payload + should_send = not self._committing + self._lock.notify_all() + if should_send: + try: + self.connection.send_json( + self._response( + "partial", + transcript=payload["text"], + language=payload["language"], + speculative=True, + ) + ) + except (BrokenPipeError, ConnectionError, OSError): + self.cancel() + + threading.Thread( + target=worker, + name=f"stt-speculate-{self.turn_id}", + daemon=True, + ).start() + + def commit(self) -> None: + """Emit one final result from the complete stable utterance snapshot.""" + with self._lock: + if not self._started or self._closed or self._committing: + raise WebSocketError("turn cannot be committed", 1008) + if not self._pcm: + raise WebSocketError("turn has no audio", 1008) + self._committing = True + pcm, key = self._snapshot_locked() + deadline = time.monotonic() + STREAM_IDLE_SECONDS + while self._pending_key == key and not self._cached_payload: + remaining = deadline - time.monotonic() + if remaining <= 0: + break + self._lock.wait(timeout=remaining) + payload = self._cached_payload if self._cached_key == key else None + + try: + if payload is None: + payload = _transcribe_pcm(self.model, pcm, self.language) + self.connection.send_json( + self._response( + "final", + transcript=payload["text"], + language=payload["language"], + model=payload["model"], + ) + ) + except Exception as exc: + print(f"[stt] streaming transcription failed: {exc}", flush=True) + self.error("transcription failed", fatal=True) + finally: + with self._lock: + self._closed = True + self._lock.notify_all() + + def cancel(self) -> None: + """Invalidate queued work and prevent background writers after disconnect.""" + with self._lock: + self._closed = True + self._cached_payload = None + self._cached_key = "" + self._lock.notify_all() + + def _json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> None: body = json.dumps(payload).encode("utf-8") handler.send_response(status) @@ -109,7 +517,11 @@ class SpeechHandler(BaseHTTPRequestHandler): print(f"[stt] {self.address_string()} {message % args}", flush=True) def do_GET(self) -> None: - if self.path != "/health": + path = self.path.split("?", 1)[0] + if path == "/v1/audio/transcriptions/stream": + self._serve_stream() + return + if path != "/health": _json(self, 404, {"error": "not found"}) return _json( @@ -119,9 +531,101 @@ class SpeechHandler(BaseHTTPRequestHandler): "ok": True, "model": MODEL_NAME, "device": "cuda" if torch.cuda.is_available() else "cpu", + "streaming": { + "enabled": True, + "path": "/v1/audio/transcriptions/stream", + "format": "pcm_s16le", + "sample_rate": STREAM_SAMPLE_RATE, + "max_seconds": MAX_STREAM_SECONDS, + "server_vad": True, + "speculative": True, + }, }, ) + def _serve_stream(self) -> None: + """Upgrade one request and process a single, isolated voice turn.""" + upgrade = self.headers.get("Upgrade", "").strip().lower() + connection_tokens = { + token.strip().lower() + for token in self.headers.get("Connection", "").split(",") + } + key = self.headers.get("Sec-WebSocket-Key", "").strip() + version = self.headers.get("Sec-WebSocket-Version", "").strip() + if upgrade != "websocket" or "upgrade" not in connection_tokens or not key: + _json(self, 426, {"error": "websocket upgrade is required"}) + return + if version != "13": + self.send_response(426) + self.send_header("Sec-WebSocket-Version", "13") + self.send_header("Content-Length", "0") + self.end_headers() + return + try: + decoded_key = base64.b64decode(key, validate=True) + except (ValueError, TypeError): + decoded_key = b"" + if len(decoded_key) != 16: + _json(self, 400, {"error": "invalid websocket key"}) + return + + self.send_response(101, "Switching Protocols") + self.send_header("Upgrade", "websocket") + self.send_header("Connection", "Upgrade") + self.send_header("Sec-WebSocket-Accept", WebSocketConnection.accept_value(key)) + self.end_headers() + self.wfile.flush() + self.close_connection = True + self.connection.settimeout(STREAM_IDLE_SECONDS) + websocket = WebSocketConnection(self) + session = StreamingTranscription( + websocket, + self.server.model, # type: ignore[attr-defined] + ) + close_code = 1000 + close_reason = "" + try: + while True: + opcode, payload = websocket.receive() + if opcode == 0x2: + session.append(payload) + continue + try: + message = json.loads(payload.decode("utf-8")) + except (UnicodeDecodeError, json.JSONDecodeError): + raise WebSocketError("control message must be JSON", 1007) + if not isinstance(message, dict): + raise WebSocketError("control message must be an object", 1007) + message_type = str(message.get("type") or "") + if message_type == "start": + session.start(message) + continue + if message.get("turn_id") != session.turn_id: + raise WebSocketError("turn_id does not match", 1008) + if message_type == "speculate": + session.speculate() + elif message_type == "resume": + session.resume() + elif message_type == "commit": + session.commit() + break + elif message_type == "cancel": + break + else: + raise WebSocketError("unknown stream control", 1008) + except WebSocketError as exc: + close_code = exc.close_code + close_reason = str(exc) + try: + session.error(str(exc), fatal=True) + except (BrokenPipeError, ConnectionError, OSError): + pass + except (EOFError, BrokenPipeError, ConnectionError, socket.timeout, OSError): + close_reason = "peer disconnected" + finally: + session.cancel() + websocket.close(close_code, close_reason) + def do_POST(self) -> None: if self.path != "/v1/audio/transcriptions": _json(self, 404, {"error": "not found"}) @@ -151,11 +655,15 @@ class SpeechHandler(BaseHTTPRequestHandler): return suffix = Path(getattr(audio, "filename", "audio.wav") or "audio.wav").suffix - suffix = suffix if suffix in {".wav", ".webm", ".ogg", ".mp3", ".m4a"} else ".wav" + suffix = ( + suffix if suffix in {".wav", ".webm", ".ogg", ".mp3", ".m4a"} else ".wav" + ) language = str(form.getfirst("language", "auto") or "auto").strip().lower() temp_path = "" try: - with tempfile.NamedTemporaryFile(prefix="atlas-stt-", suffix=suffix, delete=False) as temp: + with tempfile.NamedTemporaryFile( + prefix="atlas-stt-", suffix=suffix, delete=False + ) as temp: temp_path = temp.name while True: chunk = audio.file.read(1024 * 1024) @@ -163,20 +671,12 @@ class SpeechHandler(BaseHTTPRequestHandler): break temp.write(chunk) - with MODEL_LOCK: - result = self.server.model.transcribe( # type: ignore[attr-defined] - temp_path, - language=None if language in {"", "auto"} else language, - task="transcribe", - fp16=torch.cuda.is_available(), - condition_on_previous_text=False, - temperature=0, - compression_ratio_threshold=2.0, - logprob_threshold=-0.8, - no_speech_threshold=0.5, - verbose=False, - ) - _json(self, 200, _transcription_payload(result)) + payload = _transcribe_path( + self.server.model, # type: ignore[attr-defined] + temp_path, + language, + ) + _json(self, 200, payload) except Exception as exc: print(f"[stt] transcription failed: {exc}", flush=True) _json(self, 500, {"error": "transcription failed"}) diff --git a/dockerfiles/hermes-jetson-tts-server.py b/dockerfiles/hermes-jetson-tts-server.py index 0b962ae2..aff6afdb 100644 --- a/dockerfiles/hermes-jetson-tts-server.py +++ b/dockerfiles/hermes-jetson-tts-server.py @@ -6,6 +6,7 @@ from __future__ import annotations import io import json import os +import re import threading import wave from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer @@ -20,7 +21,13 @@ PORT = int(os.getenv("HERMES_TTS_PORT", "9001")) CACHE_DIR = Path(os.getenv("HERMES_TTS_CACHE", "/cache/piper")) MAX_TEXT_CHARS = 5000 ONNX_THREADS = max(1, int(os.getenv("HERMES_TTS_ONNX_THREADS", "4"))) +STREAM_WRITE_BYTES = 32 * 1024 +STREAM_WRITE_TIMEOUT_SECONDS = max( + 1.0, + float(os.getenv("HERMES_TTS_STREAM_WRITE_TIMEOUT_SECONDS", "15")), +) VOICE_LOCK = threading.Lock() +TURN_ID_PATTERN = re.compile(r"[A-Za-z0-9._:-]{1,128}\Z") # Fixed, allow-listed language -> baked voice mapping. This is the ONLY path # from a client-supplied string to a model name: client input is looked up @@ -62,6 +69,16 @@ def resolve_voice_name(language: object) -> str: return LANGUAGE_VOICE_MAP.get(normalized, DEFAULT_VOICE_NAME) +def normalize_turn_id(value: object) -> str | None: + """Return a bounded header-safe turn identifier, or discard it.""" + if not isinstance(value, str): + return None + normalized = value.strip() + if not TURN_ID_PATTERN.fullmatch(normalized): + return None + return normalized + + def _json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> None: body = json.dumps(payload).encode("utf-8") handler.send_response(status) @@ -76,6 +93,7 @@ class SpeechHandler(BaseHTTPRequestHandler): """Serve health and bounded local speech synthesis.""" server_version = "AtlasPiper/1" + protocol_version = "HTTP/1.1" def log_message(self, message: str, *args: object) -> None: print(f"[tts] {self.address_string()} {message % args}", flush=True) @@ -92,19 +110,29 @@ class SpeechHandler(BaseHTTPRequestHandler): "voices": sorted(self.server.voices), # type: ignore[attr-defined] "default_voice": self.server.default_voice_name, # type: ignore[attr-defined] "device": "cpu", + "streaming": { + "path": "/v1/audio/speech/stream", + "format": "pcm_s16le", + }, }, ) def do_POST(self) -> None: - if self.path != "/v1/audio/speech": + if self.path not in {"/v1/audio/speech", "/v1/audio/speech/stream"}: _json(self, 404, {"error": "not found"}) return - content_length = int(self.headers.get("Content-Length", "0") or "0") + try: + content_length = int(self.headers.get("Content-Length", "0") or "0") + except (TypeError, ValueError): + _json(self, 400, {"error": "invalid Content-Length"}) + return if content_length <= 0 or content_length > 64 * 1024: _json(self, 413, {"error": "request is missing or too large"}) return try: payload = json.loads(self.rfile.read(content_length).decode("utf-8")) + if not isinstance(payload, dict): + raise TypeError("request must be an object") text = str(payload.get("input") or payload.get("text") or "").strip() speed = float(payload.get("speed") or 1.0) except (TypeError, ValueError, json.JSONDecodeError): @@ -124,6 +152,21 @@ class SpeechHandler(BaseHTTPRequestHandler): voice_name = resolve_voice_name(payload.get("language")) voice = self.server.voices[voice_name] # type: ignore[attr-defined] + if self.path == "/v1/audio/speech/stream": + self._stream_pcm( + text, + speed, + voice_name, + voice, + normalize_turn_id(payload.get("turn_id")), + ) + return + + self._synthesize_wav(text, speed, voice_name, voice) + + def _synthesize_wav(self, text: str, speed: float, voice_name: str, voice: PiperVoice) -> None: + """Return the compatibility WAV response after complete synthesis.""" + output = io.BytesIO() try: with VOICE_LOCK, wave.open(output, "wb") as wav_file: @@ -144,6 +187,100 @@ class SpeechHandler(BaseHTTPRequestHandler): print(f"[tts] synthesis failed: {exc}", flush=True) _json(self, 500, {"error": "speech synthesis failed"}) + def _write_stream_chunk(self, audio: memoryview) -> bool: + """Write one bounded HTTP chunk, returning false after disconnect.""" + try: + self.wfile.write(f"{len(audio):X}\r\n".encode("ascii")) + self.wfile.write(audio) + self.wfile.write(b"\r\n") + self.wfile.flush() + return True + except (BrokenPipeError, ConnectionResetError, TimeoutError, OSError): + self.close_connection = True + return False + + def _stream_pcm( + self, + text: str, + speed: float, + voice_name: str, + voice: PiperVoice, + turn_id: str | None, + ) -> None: + """Progressively stream sentence PCM with bounded transport writes. + + Piper yields one PCM result per sentence. Each result is split further + before writing so the HTTP socket, rather than application buffers, + provides back-pressure. Aborting the browser fetch closes the socket; + the server then drops the current turn at the next synthesis yield or + bounded write and releases the voice lock. + """ + sample_rate = int(voice.config.sample_rate) + self.send_response(200) + self.send_header( + "Content-Type", + f"audio/pcm;rate={sample_rate};channels=1;encoding=signed-integer;bits=16;endian=little", + ) + self.send_header("Transfer-Encoding", "chunked") + self.send_header("Cache-Control", "no-store") + self.send_header("X-Audio-Format", "pcm_s16le") + self.send_header("X-Audio-Sample-Rate", str(sample_rate)) + self.send_header("X-Audio-Channels", "1") + self.send_header("X-Audio-Sample-Width", "2") + self.send_header("X-TTS-Voice", voice_name) + if turn_id is not None: + self.send_header("X-TTS-Turn-ID", turn_id) + self.end_headers() + + try: + self.connection.settimeout(STREAM_WRITE_TIMEOUT_SECONDS) + except (AttributeError, OSError): + # Unit-test handlers and already-closed clients may not expose a + # live socket. The writes below remain the cancellation boundary. + pass + + connected = True + try: + audio_chunks = iter( + voice.synthesize( + text, + SynthesisConfig(length_scale=1.0 / speed), + ) + ) + while connected: + # Serialize Piper/ONNX access, but never hold the model lock + # while a slow browser applies network back-pressure. + with VOICE_LOCK: + try: + audio_chunk = next(audio_chunks) + except StopIteration: + break + if ( + audio_chunk.sample_rate != sample_rate + or audio_chunk.sample_width != 2 + or audio_chunk.sample_channels != 1 + ): + raise RuntimeError("Piper returned an unexpected PCM format") + pcm = memoryview(audio_chunk.audio_int16_bytes) + for offset in range(0, len(pcm), STREAM_WRITE_BYTES): + if not self._write_stream_chunk(pcm[offset : offset + STREAM_WRITE_BYTES]): + connected = False + break + except Exception as exc: + # Headers are already committed, so a JSON error would corrupt the + # PCM stream. Closing produces an incomplete chunked response that + # the browser can discard or replace with the compatibility path. + print(f"[tts] streaming synthesis failed: {exc}", flush=True) + self.close_connection = True + return + + if connected: + try: + self.wfile.write(b"0\r\n\r\n") + self.wfile.flush() + except (BrokenPipeError, ConnectionResetError, TimeoutError, OSError): + self.close_connection = True + def _load_voice(cache_dir: Path, voice_name: str, threads: int) -> PiperVoice: model_path = cache_dir / f"{voice_name}.onnx" diff --git a/dockerfiles/hermes-webui-atlas-patch.py b/dockerfiles/hermes-webui-atlas-patch.py index c39e236b..186e7cfa 100644 --- a/dockerfiles/hermes-webui-atlas-patch.py +++ b/dockerfiles/hermes-webui-atlas-patch.py @@ -111,6 +111,16 @@ replace_exact( ''', ) +service_worker = ROOT / "static/sw.js" +replace_exact( + service_worker, + " './static/style.css' + VQ,\n", + " './static/style.css' + VQ,\n" + " './static/atlas-voice.css' + VQ,\n" + " './static/atlas-voice.js' + VQ,\n" + " './static/atlas-voice-worklet.js' + VQ,\n", +) + ui = ROOT / "static/ui.js" replace_exact( ui, @@ -294,6 +304,11 @@ replace_exact( ) routes = ROOT / "api/routes.py" +replace_exact( + routes, + "import html as _html\n", + "import base64\nimport html as _html\nimport secrets\n", +) replace_exact( routes, "def _tts_open(req, *, timeout=30, opener_factory=None):", @@ -322,6 +337,312 @@ def _atlas_tts_language(body): def _tts_open(req, *, timeout=30, opener_factory=None):''', ) + +replace_exact( + routes, + "def _tts_open(req, *, timeout=30, opener_factory=None):", + '''ATLAS_TTS_STREAM_URL = "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech/stream" +ATLAS_STT_STREAM_URL = "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions/stream" +ATLAS_VOICE_WS_PROTOCOL = "hermes-voice-v1" +ATLAS_VOICE_MAX_BYTES = 8 * 1024 * 1024 +ATLAS_VOICE_DEADLINE_SECONDS = 100 + + +def _atlas_exact_stream_url(env_name, expected): + """Return an exact private service URL, never a browser-selected target.""" + configured = os.getenv(env_name, "").strip() + return configured if configured == expected else "" + + +def _handle_atlas_streaming_capability(handler): + """Advertise only transports whose immutable in-cluster URLs are configured.""" + tts = bool(_atlas_exact_stream_url("HERMES_WEBUI_ATLAS_TTS_STREAM_URL", ATLAS_TTS_STREAM_URL)) + stt = bool(_atlas_exact_stream_url("HERMES_WEBUI_ATLAS_STT_STREAM_URL", ATLAS_STT_STREAM_URL)) + j(handler, { + "tts": { + "available": tts, + "transport": "http", + "format": "pcm_s16le", + "sample_rate": 22050, + }, + "stt": { + "available": stt, + "transport": "websocket", + "path": "/api/transcribe/stream", + "format": "pcm_s16le", + "sample_rate": 16000, + }, + }, extra_headers={"Cache-Control": "no-store"}) + return True + + +def _atlas_tts_stream_payload(data): + """Build the narrow Piper payload used by the raw-PCM stream endpoint.""" + if not isinstance(data, dict): + raise ValueError("invalid request body") + text = data.get("text") + if not isinstance(text, str) or not text.strip(): + raise ValueError("text is required") + text = text.strip() + if len(text) > 500: + raise ValueError("text too long (max 500 characters)") + payload = {"model": "piper", "input": text, "speed": 1.0} + language = _atlas_tts_language(data) + if language: + payload["language"] = language + turn_id = data.get("turn_id") + if isinstance(turn_id, str) and re.fullmatch(r"[A-Za-z0-9._-]{1,64}", turn_id): + payload["turn_id"] = turn_id + return payload + + +def _handle_atlas_tts_stream(handler): + """Relay bounded private Piper PCM using HTTP/1.1 chunk framing.""" + target = _atlas_exact_stream_url("HERMES_WEBUI_ATLAS_TTS_STREAM_URL", ATLAS_TTS_STREAM_URL) + if not target: + return bad(handler, "Atlas streaming TTS is not configured", 503) + headers_sent = False + try: + payload = _atlas_tts_stream_payload(read_body(handler)) + except (TypeError, ValueError) as exc: + return bad(handler, str(exc), 400) + request = Request( + target, + data=json.dumps(payload).encode("utf-8"), + headers={"Content-Type": "application/json", "Accept": "audio/pcm"}, + ) + try: + upstream = _tts_open( + request, + timeout=45, + opener_factory=lambda: build_opener(ProxyHandler({}), _NoRedirectTtsHandler()), + ) + with upstream: + content_type = str(upstream.headers.get("Content-Type", "")).split(";", 1)[0].lower() + if content_type not in ("audio/l16", "audio/pcm", "application/octet-stream"): + raise ValueError("Atlas streaming TTS returned an unsupported format") + try: + sample_rate = int(upstream.headers.get("X-Audio-Sample-Rate", "22050")) + channels = int(upstream.headers.get("X-Audio-Channels", "1")) + except (TypeError, ValueError): + raise ValueError("Atlas streaming TTS returned invalid audio metadata") + if not 8000 <= sample_rate <= 96000 or channels != 1: + raise ValueError("Atlas streaming TTS returned invalid audio metadata") + handler.send_response(200) + handler.send_header( + "Content-Type", + f"audio/pcm;rate={sample_rate};channels=1;encoding=signed-integer;bits=16;endian=little", + ) + handler.send_header("X-Audio-Sample-Rate", str(sample_rate)) + handler.send_header("X-Audio-Channels", "1") + handler.send_header("Cache-Control", "no-store") + handler.send_header("Transfer-Encoding", "chunked") + handler.end_headers() + headers_sent = True + sent = 0 + while True: + chunk = upstream.read(16384) + if not chunk: + break + sent += len(chunk) + if sent > ATLAS_VOICE_MAX_BYTES: + raise ValueError("Atlas streaming TTS exceeded its audio limit") + handler.wfile.write(("%x\\r\\n" % len(chunk)).encode("ascii")) + handler.wfile.write(chunk) + handler.wfile.write(b"\\r\\n") + handler.wfile.flush() + handler.wfile.write(b"0\\r\\n\\r\\n") + handler.wfile.flush() + return True + except (BrokenPipeError, ConnectionResetError): + return True + except Exception: + logger.exception("Atlas streaming TTS generation failed") + if not headers_sent: + return bad(handler, "Atlas streaming TTS generation failed", 502) + handler.close_connection = True + return True + + +def _atlas_ws_protocols(handler): + return [value.strip() for value in handler.headers.get("Sec-WebSocket-Protocol", "").split(",") if value.strip()] + + +def _atlas_ws_authorized(handler): + """Require same-origin plus the rendered session CSRF token when enabled.""" + origin = handler.headers.get("Origin", "").strip() + if not origin or not _check_same_origin_browser_request(handler): + return False + protocols = _atlas_ws_protocols(handler) + if ATLAS_VOICE_WS_PROTOCOL not in protocols: + return False + from api.auth import csrf_token_for_session, is_auth_enabled, parse_cookie, verify_session + if not is_auth_enabled(): + return True + cookie = parse_cookie(handler) + if not cookie: + cookie = getattr(handler, "_trusted_auth_session_cookie_value", None) + if not cookie or not verify_session(cookie): + return False + expected = csrf_token_for_session(cookie) or "" + supplied = next((value.removeprefix("hermes-csrf.") for value in protocols if value.startswith("hermes-csrf.")), "") + return bool(expected and supplied and secrets.compare_digest(expected, supplied)) + + +def _atlas_ws_upstream_handshake(target): + parsed = urlsplit(target) + if parsed.scheme != "http" or not parsed.hostname or parsed.port is None: + raise ValueError("invalid private streaming STT URL") + upstream = _socket.create_connection((parsed.hostname, parsed.port), timeout=5) + upstream.settimeout(5) + key = base64.b64encode(os.urandom(16)).decode("ascii") + path = parsed.path or "/" + if parsed.query: + path += "?" + parsed.query + request = ( + f"GET {path} HTTP/1.1\\r\\nHost: {parsed.hostname}:{parsed.port}\\r\\n" + "Upgrade: websocket\\r\\nConnection: Upgrade\\r\\n" + f"Sec-WebSocket-Key: {key}\\r\\nSec-WebSocket-Version: 13\\r\\n\\r\\n" + ).encode("ascii") + upstream.sendall(request) + response = bytearray() + while b"\\r\\n\\r\\n" not in response and len(response) < 16384: + chunk = upstream.recv(4096) + if not chunk: + break + response.extend(chunk) + header, separator, remainder = bytes(response).partition(b"\\r\\n\\r\\n") + if not separator or not header.startswith(b"HTTP/1.1 101"): + upstream.close() + raise ConnectionError("private streaming STT rejected WebSocket upgrade") + expected = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii")).digest()).decode("ascii") + headers = {} + for line in header.split(b"\\r\\n")[1:]: + name, marker, value = line.partition(b":") + if marker: + headers[name.decode("ascii", "ignore").strip().lower()] = value.decode("ascii", "ignore").strip() + if not secrets.compare_digest(headers.get("sec-websocket-accept", ""), expected): + upstream.close() + raise ConnectionError("private streaming STT returned an invalid handshake") + upstream.settimeout(1) + return upstream, remainder + + +def _atlas_ws_relay(source, destination, stop, deadline): + transferred = 0 + while not stop.is_set() and time.monotonic() < deadline: + try: + chunk = source.recv(16384) + except (_socket.timeout, TimeoutError): + continue + except OSError: + break + if not chunk: + break + transferred += len(chunk) + if transferred > ATLAS_VOICE_MAX_BYTES: + break + try: + destination.sendall(chunk) + except OSError: + break + stop.set() + + +def _handle_atlas_stt_stream(handler): + """Bridge one authenticated same-origin browser WebSocket to private STT.""" + target = _atlas_exact_stream_url("HERMES_WEBUI_ATLAS_STT_STREAM_URL", ATLAS_STT_STREAM_URL) + if not target: + return bad(handler, "Atlas streaming STT is not configured", 503) + if handler.headers.get("Upgrade", "").strip().lower() != "websocket": + return bad(handler, "WebSocket upgrade required", 426) + if not _atlas_ws_authorized(handler): + return bad(handler, "WebSocket origin or CSRF validation failed", 403) + browser_key = handler.headers.get("Sec-WebSocket-Key", "").strip() + if handler.headers.get("Sec-WebSocket-Version", "").strip() != "13": + return bad(handler, "WebSocket version 13 required", 426) + try: + decoded = base64.b64decode(browser_key, validate=True) + except Exception: + decoded = b"" + if len(decoded) != 16: + return bad(handler, "Invalid WebSocket key", 400) + try: + upstream, remainder = _atlas_ws_upstream_handshake(target) + except Exception: + logger.exception("Atlas streaming STT connection failed") + return bad(handler, "Atlas streaming STT is unavailable", 502) + browser_accept = base64.b64encode(hashlib.sha1((browser_key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii")).digest()).decode("ascii") + handler.send_response(101, "Switching Protocols") + handler.send_header("Upgrade", "websocket") + handler.send_header("Connection", "Upgrade") + handler.send_header("Sec-WebSocket-Accept", browser_accept) + handler.send_header("Sec-WebSocket-Protocol", ATLAS_VOICE_WS_PROTOCOL) + handler.end_headers() + handler.wfile.flush() + handler.close_connection = True + browser = handler.connection + browser.settimeout(1) + stop = threading.Event() + deadline = time.monotonic() + ATLAS_VOICE_DEADLINE_SECONDS + if remainder: + browser.sendall(remainder) + reverse = threading.Thread( + target=_atlas_ws_relay, + args=(upstream, browser, stop, deadline), + name="atlas-stt-ws-upstream", + daemon=True, + ) + reverse.start() + try: + _atlas_ws_relay(browser, upstream, stop, deadline) + finally: + stop.set() + for sock in (upstream, browser): + try: + sock.shutdown(_socket.SHUT_RDWR) + except OSError: + pass + upstream.close() + reverse.join(timeout=2) + return True + + +def _tts_open(req, *, timeout=30, opener_factory=None):''', +) + +replace_exact( + routes, + '''def handle_get(handler, parsed) -> bool: + """Handle all GET routes. Returns True if handled, False for 404.""" +''', + '''def handle_get(handler, parsed) -> bool: + """Handle all GET routes. Returns True if handled, False for 404.""" + if parsed.path == "/api/voice/streaming/capability": + return _handle_atlas_streaming_capability(handler) + if parsed.path == "/api/transcribe/stream": + return _handle_atlas_stt_stream(handler) +''', +) + +replace_exact( + routes, + ''' if parsed.path == "/api/transcribe": + return handle_transcribe(handler) + + if parsed.path == "/api/tts": + return _handle_tts(handler, parsed) +''', + ''' if parsed.path == "/api/transcribe": + return handle_transcribe(handler) + + if parsed.path == "/api/tts/stream": + return _handle_atlas_tts_stream(handler) + + if parsed.path == "/api/tts": + return _handle_tts(handler, parsed) +''', +) marker = " # ── ElevenLabs TTS ──────────────────────────────────────────────────\n" atlas = ''' # ── Atlas private Jetson TTS ───────────────────────────────────────── if engine == "atlas": diff --git a/dockerfiles/hermes-webui-atlas-voice-worklet.js b/dockerfiles/hermes-webui-atlas-voice-worklet.js new file mode 100644 index 00000000..e0747f68 --- /dev/null +++ b/dockerfiles/hermes-webui-atlas-voice-worklet.js @@ -0,0 +1,119 @@ +/* PCM capture/playback processors for the optional low-latency voice path. */ +'use strict'; + +class AtlasPcmCaptureProcessor extends AudioWorkletProcessor { + constructor(){ + super(); + this.pending=[]; + this.pendingLength=0; + this.port.onmessage=(event)=>{ + if((event.data||{}).type!=='flush') return; + this.flush(); + this.port.postMessage({type:'flushed'}); + }; + } + + flush(){ + if(!this.pendingLength) return; + const samples=new Float32Array(this.pendingLength); + let offset=0; + this.pending.forEach(function(chunk){samples.set(chunk,offset);offset+=chunk.length;}); + this.pending=[]; + this.pendingLength=0; + this.port.postMessage({type:'pcm',samples:samples.buffer},[samples.buffer]); + } + + process(inputs){ + const input=inputs[0]&&inputs[0][0]; + if(!input||!input.length) return true; + this.pending.push(new Float32Array(input)); + this.pendingLength+=input.length; + if(this.pendingLength>=1024){ + this.flush(); + } + return true; + } +} + +class AtlasPcmPlaybackProcessor extends AudioWorkletProcessor { + constructor(){ + super(); + this.queue=[]; + this.queueOffset=0; + this.phase=0; + this.sourceRate=22050; + this.ended=false; + this.cancelled=false; + this.queuedFrames=0; + this.reportCountdown=0; + this.drainedSent=false; + this.port.onmessage=(event)=>{ + const data=event.data||{}; + if(data.type==='push'&&data.samples){ + const pcm=new Int16Array(data.samples); + const samples=new Float32Array(pcm.length); + for(let index=0;index=this.queue[0].length){ + this.queue.shift(); + this.queueOffset=0; + } + return this.queue.length?this.queue[0][this.queueOffset]:null; + } + + nextSample(){ + if(!this.queue.length) return null; + if(this.queueOffset+11&&this.queue[1].length?this.queue[1][0]:this.queue[0][this.queueOffset]; + } + + advance(){ + if(!this.queue.length) return; + this.queueOffset+=1; + this.queuedFrames=Math.max(0,this.queuedFrames-1); + if(this.queueOffset>=this.queue[0].length){this.queue.shift();this.queueOffset=0;} + } + + process(_inputs,outputs){ + const output=outputs[0]&&outputs[0][0]; + if(!output) return !this.cancelled; + output.fill(0); + if(this.cancelled) return false; + const ratio=this.sourceRate/sampleRate; + for(let index=0;index=1){this.advance();this.phase-=1;} + } + this.reportCountdown-=1; + if(this.reportCountdown<=0){ + this.reportCountdown=20; + this.port.postMessage({type:'buffer',frames:this.queuedFrames}); + } + if(this.ended&&!this.queue.length&&!this.drainedSent){ + this.drainedSent=true; + this.port.postMessage({type:'drained'}); + } + return true; + } +} + +registerProcessor('atlas-pcm-capture',AtlasPcmCaptureProcessor); +registerProcessor('atlas-pcm-playback',AtlasPcmPlaybackProcessor); diff --git a/dockerfiles/hermes-webui-atlas-voice.js b/dockerfiles/hermes-webui-atlas-voice.js index b8f0aadd..962d488f 100644 --- a/dockerfiles/hermes-webui-atlas-voice.js +++ b/dockerfiles/hermes-webui-atlas-voice.js @@ -13,16 +13,30 @@ let active=false; let state='idle'; let generation=0; + let turnSequence=0; let recorder=null; let stream=null; let audioContext=null; + let captureNode=null; let vadTimer=null; + let responsePollTimer=null; let currentAudio=null; + let playbackSession=null; + let streamingStt=null; + let captureTurnId=''; + let thinkingTurnId=''; let thinkingSession=null; + let assistantBaseline=null; + let speechTurn=null; let errorTimer=null; let visualInputLevel=0; + let streamingCapability={tts:null,stt:null}; const reducedMotion=window.matchMedia?window.matchMedia('(prefers-reduced-motion: reduce)'):{matches:false}; const ERROR_VISIBLE_MS=3200; + const STREAMING_CAPABILITY_URL='/api/voice/streaming/capability'; + const TTS_STREAM_URL='/api/tts/stream'; + const STT_STREAM_PATH='/api/transcribe/stream'; + const WORKLET_URL='static/atlas-voice-worklet.js'; const STATE_LABELS={ listening:'Listening', transcribing:'Transcribing…', @@ -51,7 +65,7 @@ sttLanguageToken=-1; } - function rememberSttLanguage(language, token){ + function rememberSttLanguage(language,token){ sttLanguage=language||''; sttLanguageToken=sttLanguage?token:-1; } @@ -66,7 +80,7 @@ if(typeof window.showToast==='function') window.showToast(message,3000); } - function setState(next, customLabel){ + function setState(next,customLabel){ state=next; indicator.className='voice-mode-indicator '+next; bar.dataset.voiceState=next; @@ -94,14 +108,31 @@ if(errorTimer){window.clearTimeout(errorTimer);errorTimer=null;} } - function errorMessage(error, fallback){ + function errorMessage(error,fallback){ return String((error&&error.message)||fallback||'Voice unavailable').trim(); } + function cancelSpeechTurn(){ + if(!speechTurn) return; + speechTurn.cancelled=true; + speechTurn.final=true; + while(speechTurn.waiters.length) speechTurn.waiters.shift()(null); + speechTurn=null; + } + + function cancelStreamingStt(){ + if(!streamingStt) return; + streamingStt.cancel(); + streamingStt=null; + } + function showUnavailable(message){ generation+=1; active=false; thinkingSession=null; + thinkingTurnId=''; + stopResponseObserver(); + cancelSpeechTurn(); stopCapture(); stopPlayback(); modeBtn.classList.remove('active'); @@ -119,23 +150,48 @@ try{recorder.stop();}catch(_){ } } recorder=null; + captureNode=null; + cancelStreamingStt(); if(stream){stream.getTracks().forEach(function(track){track.stop();});stream=null;} if(audioContext){try{audioContext.close();}catch(_){ }audioContext=null;} } function stopPlayback(){ - if(!currentAudio) return; - try{currentAudio.pause();currentAudio.currentTime=0;}catch(_){ } - currentAudio=null; + if(currentAudio){ + try{currentAudio.pause();currentAudio.currentTime=0;}catch(_){ } + currentAudio=null; + } + if(playbackSession){ + playbackSession.cancelled=true; + if(playbackSession.blobWake){playbackSession.blobWake();playbackSession.blobWake=null;} + playbackSession.controllers.forEach(function(controller){controller.abort();}); + if(playbackSession.lowWaterWake){ + playbackSession.lowWaterWake(); + playbackSession.lowWaterWake=null; + } + if(playbackSession.drainWake){playbackSession.drainWake();playbackSession.drainWake=null;} + if(playbackSession.node){ + try{playbackSession.node.port.postMessage({type:'cancel'});playbackSession.node.disconnect();}catch(_){ } + } + if(playbackSession.context){try{playbackSession.context.close();}catch(_){ }} + playbackSession=null; + } indicator.classList.remove('is-playing'); } + function stopResponseObserver(){ + if(responsePollTimer){clearInterval(responsePollTimer);responsePollTimer=null;} + } + function deactivate(showMessage){ generation+=1; active=false; thinkingSession=null; + thinkingTurnId=''; clearErrorTimer(); clearSttLanguage(); + stopResponseObserver(); + cancelSpeechTurn(); stopCapture(); stopPlayback(); modeBtn.classList.remove('active'); @@ -143,21 +199,40 @@ if(showMessage) toast('Hands-free voice mode off'); } - function restartSoon(token, delay){ + function restartSoon(token,delay){ window.setTimeout(function(){ if(active&&token===generation) startListening(token); },delay||500); } - function sendTranscript(transcript, token, language){ + function assistantRows(){ + return document.querySelectorAll('.msg-row[data-role="assistant"], .assistant-segment[data-raw-text]'); + } + + function readAssistantRow(row){ + if(!row) return ''; + if(row.dataset&&typeof row.dataset.rawText==='string') return row.dataset.rawText; + return typeof row.textContent==='string'?row.textContent:''; + } + + function rememberAssistantBaseline(){ + const rows=assistantRows(); + const row=rows.length?rows[rows.length-1]:null; + assistantBaseline={row:row,text:readAssistantRow(row),count:rows.length}; + } + + function sendTranscript(transcript,token,language){ if(!active||token!==generation) return; const text=String(transcript||'').trim(); if(!text){clearSttLanguage();restartSoon(token,350);return;} composer.value=text; if(typeof window.autoResize==='function') window.autoResize(); thinkingSession=(typeof S!=='undefined'&&S.session)?S.session.session_id:null; + thinkingTurnId=captureTurnId||String(token)+'-'+String(++turnSequence); + rememberAssistantBaseline(); rememberSttLanguage(language,token); setState('thinking'); + startResponseObserver(token); if(typeof window.send==='function') window.send(); } @@ -168,7 +243,7 @@ return 'webm'; } - async function transcribe(blob, token){ + async function transcribe(blob,token){ if(!active||token!==generation) return; setState('transcribing'); const ext=audioExtension(blob.type); @@ -193,10 +268,211 @@ } } + function websocketUrl(path){ + const protocol=window.location&&window.location.protocol==='https:'?'wss:':'ws:'; + return protocol+'//'+window.location.host+path; + } + + function encodePcm16(samples){ + const bytes=new ArrayBuffer(samples.length*2); + const view=new DataView(bytes); + for(let index=0;index1||Date.now()>=commitDeadline){ + reject(new Error('Streaming transcription closed before commit')); + return; + } + if(socket.readyState!==1||queue.length){flush();window.setTimeout(sendCommit,20);return;} + if(!sendJson({type:'commit',turn_id:turnId})) reject(new Error('Streaming transcription commit failed')); + }; + sendCommit(); + return finalPromise; + }, + cancel:function(){ + if(!settled){sendJson({type:'cancel',turn_id:turnId});reject(new Error('Streaming transcription cancelled'));} + if(socket&&socket.readyState<2) socket.close(1000,'cancelled'); + }, + }; + } + + async function installCaptureWorklet(context,source,session){ + if(!session) return false; + if(!context.audioWorklet||!window.AudioWorkletNode){ + session.cancel(); + if(streamingStt===session) streamingStt=null; + return false; + } + try{ + await context.audioWorklet.addModule(WORKLET_URL); + if(!streamingStt||streamingStt!==session) return false; + const node=new AudioWorkletNode(context,'atlas-pcm-capture'); + const silentGain=context.createGain(); + silentGain.gain.value=0; + let flushResolve=null; + node.port.onmessage=function(event){ + if(!event.data) return; + if(event.data.type==='flushed'&&flushResolve){flushResolve();flushResolve=null;return;} + if(!streamingStt||streamingStt!==session||event.data.type!=='pcm') return; + session.push(new Float32Array(event.data.samples)); + }; + node._atlasFlush=function(){ + return new Promise(function(resolve){flushResolve=resolve;node.port.postMessage({type:'flush'});}); + }; + source.connect(node); + node.connect(silentGain); + silentGain.connect(context.destination); + captureNode=node; + session.setWorkletReady(true); + return true; + }catch(_){ + session.cancel(); + if(streamingStt===session) streamingStt=null; + return false; + } + } + + async function transcribeStreamingOrFallback(blob,token,session){ + if(!active||token!==generation) return; + setState('transcribing'); + if(session){ + try{ + const result=await session.commit(); + if(!String(result.transcript||'').trim()) throw new Error('Streaming transcription returned no final text'); + if(active&&token===generation){ + streamingStt=null; + sendTranscript(result.transcript,token,result.language); + return; + } + }catch(_){ + // The finalized browser container below is the quality-preserving path + // whenever rolling PCM transport or speculative inference fails. + } + if(streamingStt===session) streamingStt=null; + } + transcribe(blob,token); + } + async function startListening(token){ if(!active||token!==generation) return; stopCapture(); + stopPlayback(); + cancelSpeechTurn(); clearSttLanguage(); + captureTurnId=String(token)+'-'+String(++turnSequence); setState('listening'); try{ const capture=await navigator.mediaDevices.getUserMedia({ @@ -210,15 +486,26 @@ if(!active||token!==generation){capture.getTracks().forEach(function(track){track.stop();});return;} stream=capture; const Context=window.AudioContext||window.webkitAudioContext; - audioContext=new Context(); + try{ + // Let the browser's native resampler produce Whisper's 16 kHz input. + audioContext=new Context({sampleRate:16000,latencyHint:'interactive'}); + }catch(_){ + audioContext=new Context(); + } const analyser=audioContext.createAnalyser(); analyser.fftSize=2048; const highpass=audioContext.createBiquadFilter(); highpass.type='highpass'; highpass.frequency.value=140; highpass.Q.value=0.7; - audioContext.createMediaStreamSource(stream).connect(highpass); + const mediaSource=audioContext.createMediaStreamSource(stream); + mediaSource.connect(highpass); highpass.connect(analyser); + streamingStt=createStreamingSttSession(captureTurnId,audioContext); + // Whisper receives the browser's full-band processed microphone signal; + // the 140 Hz high-pass remains a VAD-only aid so low voices are not + // needlessly altered before final recognition. + if(streamingStt) await installCaptureWorklet(audioContext,mediaSource,streamingStt); const samples=new Uint8Array(analyser.fftSize); const mimeTypes=['audio/webm;codecs=opus','audio/ogg;codecs=opus','audio/mp4;codecs=mp4a.40.2','audio/mp4','audio/webm']; const mime=mimeTypes.find(function(value){return MediaRecorder.isTypeSupported(value);})||''; @@ -227,6 +514,7 @@ let voiceFrames=0; let noiseFloor=0.008; let lastSpeech=Date.now(); + let speculative=false; const started=Date.now(); recorder=new MediaRecorder(stream,mime?{mimeType:mime}:undefined); let recordedMime=recorder.mimeType||mime||''; @@ -235,16 +523,27 @@ if(event.data.type) recordedMime=event.data.type; chunks.push(event.data); }; - recorder.onstop=function(){ + recorder.onstop=async function(){ if(vadTimer){clearInterval(vadTimer);vadTimer=null;} + const finalCaptureNode=captureNode; + if(finalCaptureNode&&typeof finalCaptureNode._atlasFlush==='function'){ + await Promise.race([ + finalCaptureNode._atlasFlush(), + new Promise(function(resolve){window.setTimeout(resolve,100);}), + ]); + } const recordedStream=stream; stream=null; if(recordedStream) recordedStream.getTracks().forEach(function(track){track.stop();}); - if(audioContext){try{audioContext.close();}catch(_){ }audioContext=null;} + const context=audioContext; + audioContext=null; + if(context){try{context.close();}catch(_){ }} + captureNode=null; recorder=null; - if(!active||token!==generation) return; - if(!heardSpeech||!chunks.length){restartSoon(token,300);return;} - transcribe(new Blob(chunks,{type:recordedMime||'audio/webm'}),token); + if(!active||token!==generation){cancelStreamingStt();return;} + if(!heardSpeech||!chunks.length){cancelStreamingStt();restartSoon(token,300);return;} + const session=streamingStt; + transcribeStreamingOrFallback(new Blob(chunks,{type:recordedMime||'audio/webm'}),token,session); }; // Ask the browser for one finalized container at stop. Android Chromium // can emit timeslice fragments without a reusable EBML initialization @@ -253,12 +552,13 @@ // is small enough to retain as one browser-owned recording. recorder.start(); const silenceMs=Math.max(900,parseInt(localStorage.getItem('hermes-voice-silence-ms')||'1100',10)||1100); + const speculateMs=Math.min(silenceMs-250,Math.max(450,Math.round(silenceMs*0.55))); vadTimer=window.setInterval(function(){ if(!active||token!==generation||!recorder||recorder.state==='inactive') return; analyser.getByteTimeDomainData(samples); let energy=0; - for(let i=0;i=speculateMs&&streamingStt){ + streamingStt.speculate(); + speculative=true; } const finished=heardSpeech&&(now-lastSpeech)>=silenceMs; const timedOut=now-started>=90000; @@ -295,34 +600,120 @@ return String(text||'').replace(/```[\s\S]*?```/g,' code block ').replace(/\s+/g,' ').trim(); } - function playBlob(blob, token){ + function sentenceEnd(text){ + const source=String(text||''); + const boundary=/[.!?…]+["')\]}]*(?:\s|$)/g; + const abbreviation=/(?:\b(?:mr|mrs|ms|dr|prof|sr|jr|st|vs|etc|e\.g|i\.e)|\b[A-Z])\.$/i; + let match; + while((match=boundary.exec(source))){ + const end=match.index+match[0].trimEnd().length; + const prefix=source.slice(0,end).replace(/["')\]}]+$/,''); + if(prefix.endsWith('.')&&abbreviation.test(prefix)) continue; + return end; + } + return -1; + } + + function preferredCut(text,min,target,max){ + const bounded=text.slice(0,max); + const candidates=[]; + const punctuation=/[,;:—–.!?…]+["')\]}]*(?:\s|$)/g; + let match; + while((match=punctuation.exec(bounded))){ + const end=match.index+match[0].trimEnd().length; + if(end>=min) candidates.push({end:end,weight:Math.abs(end-target)}); + } + if(candidates.length){ + candidates.sort(function(left,right){return left.weight-right.weight;}); + return candidates[0].end; + } + const spaces=[]; + const whitespace=/\s+/g; + while((match=whitespace.exec(bounded))){if(match.index>=min) spaces.push(match.index);} + if(spaces.length){ + spaces.sort(function(left,right){return Math.abs(left-target)-Math.abs(right-target);}); + return spaces[0]; + } + return Math.min(max,text.length); + } + + function adaptiveChunks(text,final,firstChunk){ + const source=String(text||'').trim(); + if(!source) return {chunks:[],consumed:0}; + const chunks=[]; + let offset=0; + if(firstChunk!==false){ + const firstSentence=sentenceEnd(source); + if(firstSentence<0&&!final) return {chunks:[],consumed:0}; + const firstAvailable=firstSentence>=0?firstSentence:source.length; + const firstSlice=source.slice(0,firstAvailable); + const firstCut=firstSlice.length<=60?firstSlice.length:preferredCut(firstSlice,40,52,60); + chunks.push(source.slice(0,firstCut).trim()); + offset=firstCut; + } + while(offset96000||channels!==1) throw new Error('Unsupported streaming speech format'); + return {kind:'pcm',response:response,sampleRate:sampleRate,controller:controller,token:token,chunk:chunk,language:language,turnId:turnId,started:false}; + }catch(error){ + if(playbackSession) playbackSession.controllers.delete(controller); + if(error&&error.name==='AbortError') throw error; + // Disable only this optional transport for the browser session. The + // complete WAV endpoint preserves voice quality and availability. + streamingCapability.tts=null; } + } + return {kind:'blob',blob:await fetchSpeech(chunk,language,turnId)}; + } + + async function playPcm(asset,token){ + if(!active||token!==generation) return; + const Context=window.AudioContext||window.webkitAudioContext; + let context; + try{ + // Keep Piper at its native rate and let the browser/audio device own any + // final hardware conversion; the worklet interpolator is a fallback. + context=new Context({sampleRate:asset.sampleRate,latencyHint:'interactive'}); + }catch(_){ + context=new Context(); + } + await context.audioWorklet.addModule(WORKLET_URL); + if(!active||token!==generation){context.close();return;} + const node=new AudioWorkletNode(context,'atlas-pcm-playback'); + const session=playbackSession; + if(!session||session.cancelled){context.close();return;} + session.context=context; + session.node=node; + node.connect(context.destination); + await context.resume(); + let bufferedFrames=0; + let lowWaterResolve=null; + let ended=false; + const drained=new Promise(function(resolve,reject){ + session.drainWake=resolve; + node.port.onmessage=function(event){ + const data=event.data||{}; + if(data.type==='buffer'){ + bufferedFrames=data.frames||0; + if(bufferedFramesasset.sampleRate*2){ + await new Promise(function(resolve){ + let completed=false; + const wake=function(){if(completed)return;completed=true;lowWaterResolve=null;session.lowWaterWake=null;resolve();}; + lowWaterResolve=wake; + session.lowWaterWake=wake; + window.setTimeout(wake,3000); + }); + } + const result=await reader.read(); + if(result.done) break; + let bytes=result.value; + if(carry!==null){ + const joined=new Uint8Array(bytes.length+1); + joined[0]=carry;joined.set(bytes,1);bytes=joined;carry=null; + } + if(bytes.length%2){carry=bytes[bytes.length-1];bytes=bytes.slice(0,-1);} + if(!bytes.length) continue; + const copy=bytes.buffer.slice(bytes.byteOffset,bytes.byteOffset+bytes.byteLength); + bufferedFrames+=bytes.byteLength/2; + asset.started=true; + node.port.postMessage({type:'push',samples:copy,sampleRate:asset.sampleRate},[copy]); + } + node.port.postMessage({type:'end'}); + if(active&&token===generation&&!session.cancelled) await drained; + }finally{ + if(!ended){try{await reader.cancel();}catch(_){ }} + indicator.classList.remove('is-playing'); + try{node.disconnect();context.close();}catch(_){ } + session.node=null; + session.context=null; + session.drainWake=null; + session.controllers.delete(asset.controller); + } + } + + async function playPrepared(asset,token){ + if(asset.kind!=='pcm') return playBlob(asset.blob,token); + try{ + return await playPcm(asset,token); }catch(error){ - if(active&&token===generation){ + if(asset.started) throw error; + streamingCapability.tts=null; + return playBlob(await fetchSpeech(asset.chunk,asset.language,asset.turnId),token); + } + } + + function nextSpeechChunk(turn){ + if(turn.cancelled) return Promise.resolve(null); + if(turn.queue.length) return Promise.resolve(turn.queue.shift()); + if(turn.final) return Promise.resolve(null); + return new Promise(function(resolve){turn.waiters.push(resolve);}); + } + + function enqueueSpeech(turn,chunks){ + chunks.forEach(function(chunk){ + if(!chunk) return; + if(turn.waiters.length) turn.waiters.shift()(chunk); else turn.queue.push(chunk); + }); + } + + function finishSpeechQueue(turn){ + turn.final=true; + while(turn.waiters.length&&!turn.queue.length) turn.waiters.shift()(null); + } + + async function runSpeechQueue(turn){ + if(turn.running) return; + turn.running=true; + playbackSession={turnId:turn.turnId,controllers:new Set(),context:null,node:null,cancelled:false,lowWaterWake:null,drainWake:null,blobWake:null}; + const session=playbackSession; + try{ + let chunk=await nextSpeechChunk(turn); + let current=chunk?prepareSpeech(chunk,turn.language,turn.turnId,turn.token):null; + while(current){ + const asset=await current; + // At most one later synthesis request exists while this asset plays. + // If streaming has not produced it yet, the waiter starts it the + // instant a punctuation-safe chunk arrives. + const next=nextSpeechChunk(turn); + const nextPrepared=next.then(function(value){ + return value?prepareSpeech(value,turn.language,turn.turnId,turn.token):null; + }); + await playPrepared(asset,turn.token); + if(!active||turn.token!==generation||turn.cancelled||session.cancelled) return; + current=await nextPrepared; + } + if(turn.final&&active&&turn.token===generation&&!turn.cancelled) restartSoon(turn.token,300); + }catch(error){ + if(active&&turn.token===generation&&!turn.cancelled){ const message=errorMessage(error,'Local speech is unavailable'); setState('error',message); toast(message); + restartSoon(turn.token,500); } + }finally{ + if(playbackSession===session) stopPlayback(); + if(speechTurn===turn) speechTurn=null; + } + } + + function currentAssistantText(){ + const rows=assistantRows(); + if(!rows.length) return ''; + const row=rows[rows.length-1]; + const text=readAssistantRow(row); + if(assistantBaseline&&row===assistantBaseline.row&&text===assistantBaseline.text) return ''; + if(assistantBaseline&&rows.length0){ + if(!isFinal) return; + thinkingSession=null; + thinkingTurnId=''; + stopResponseObserver(); + const revisedTail=text.slice(Math.min(turn.consumed,text.length)).trim(); + if(revisedTail){ + const revised=adaptiveChunks(revisedTail,true,false); + enqueueSpeech(turn,revised.chunks.length?revised.chunks:[revisedTail]); + turn.consumed=text.length; + } + finishSpeechQueue(turn); + return; + } + turn.sourceText=''; + } + turn.sourceText=text; + const remaining=text.slice(turn.consumed).trimStart(); + const skipped=text.slice(turn.consumed).length-remaining.length; + const extracted=adaptiveChunks(remaining,!!isFinal,turn.first); + if(extracted.chunks.length){ + turn.first=false; + turn.consumed+=skipped+extracted.consumed; + enqueueSpeech(turn,extracted.chunks); + setState('speaking'); + runSpeechQueue(turn); + } + if(isFinal){ + thinkingSession=null; + thinkingTurnId=''; + stopResponseObserver(); + const tail=text.slice(turn.consumed).trim(); + if(tail){enqueueSpeech(turn,[tail]);turn.consumed=text.length;} + finishSpeechQueue(turn); + } + } + + function startResponseObserver(token){ + stopResponseObserver(); + // A bounded poll is deliberately used in addition to the upstream + // completion callback. It sees data-raw-text while SSE is still appending, + // so the first complete sentence can reach Piper before generation ends. + responsePollTimer=window.setInterval(function(){pumpAssistantResponse(token,false);},75); + } + + function speakResponse(token){ + pumpAssistantResponse(token,true); + } + + function validStreamingCapability(payload){ + if(!payload||typeof payload!=='object') return {tts:null,stt:null}; + const tts=payload.tts&&payload.tts.available===true&&payload.tts.transport==='http'&&payload.tts.format==='pcm_s16le'?payload.tts:null; + const stt=payload.stt&&payload.stt.available===true&&payload.stt.transport==='websocket'&&payload.stt.format==='pcm_s16le'&&payload.stt.sample_rate===16000?payload.stt:null; + return {tts:tts,stt:stt}; + } + + async function discoverStreamingCapability(){ + try{ + const response=await fetch(STREAMING_CAPABILITY_URL,{cache:'no-store'}); + if(!response.ok) return; + streamingCapability=validStreamingCapability(await response.json()); + }catch(_){ + streamingCapability={tts:null,stt:null}; } - restartSoon(token,450); } function activate(){ @@ -394,6 +1037,7 @@ const capability=await response.json().catch(function(){return {};}); if(!response.ok||!capability.available||capability.provider!=='local_command') return; ready=true; + discoverStreamingCapability(); if(localStorage.getItem('hermes-atlas-voice-initialized')!=='1'){ localStorage.setItem('hermes-atlas-voice-initialized','1'); localStorage.setItem('hermes-voice-mode-button','true'); @@ -424,7 +1068,7 @@ } }; window.autoReadLastAssistant=function(){ - if(active&&state==='thinking'){speakResponse(generation);return;} + if(active&&(state==='thinking'||state==='speaking')){speakResponse(generation);return;} if(typeof originalAutoRead==='function') originalAutoRead.apply(this,arguments); }; window._voiceModeActive=function(){return active;}; diff --git a/services/hermes/chat-statefulset.yaml b/services/hermes/chat-statefulset.yaml index aff2a068..89a4d111 100644 --- a/services/hermes/chat-statefulset.yaml +++ b/services/hermes/chat-statefulset.yaml @@ -357,8 +357,10 @@ spec: - {name: HERMES_WEBUI_TRUST_FORWARDED_PROTO, value: "1"} - {name: HERMES_ROUTER_PROFILE, value: chat} - {name: HERMES_STT_URL, value: 'http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions'} + - {name: HERMES_WEBUI_ATLAS_STT_STREAM_URL, value: 'http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions/stream'} - {name: HERMES_LOCAL_STT_COMMAND, value: "/opt/hermes/.venv/bin/python /opt/coordinator/hermes_stt_client.py {input_path} --output-dir {output_dir} --language {language} --model {model}"} - {name: HERMES_WEBUI_ATLAS_TTS_URL, value: 'http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech'} + - {name: HERMES_WEBUI_ATLAS_TTS_STREAM_URL, value: 'http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech/stream'} volumeMounts: - {name: home, mountPath: /opt/data} - {name: workspace, mountPath: /opt/data/workspace} diff --git a/services/hermes/deployment.yaml b/services/hermes/deployment.yaml index c1b55876..05db5eec 100644 --- a/services/hermes/deployment.yaml +++ b/services/hermes/deployment.yaml @@ -48,6 +48,7 @@ spec: automountServiceAccountToken: true securityContext: fsGroup: 10000 + fsGroupChangePolicy: OnRootMismatch seccompProfile: type: RuntimeDefault affinity: diff --git a/testing/fixtures/hermes-webui-0.52.181/api/routes.py b/testing/fixtures/hermes-webui-0.52.181/api/routes.py index 499048df..e7f3b377 100644 --- a/testing/fixtures/hermes-webui-0.52.181/api/routes.py +++ b/testing/fixtures/hermes-webui-0.52.181/api/routes.py @@ -1,5 +1,6 @@ """Pinned Atlas TTS route anchors used by image-patch tests.""" +import html as _html import json import os from urllib.request import ProxyHandler, Request, build_opener @@ -39,3 +40,18 @@ def _tts_open(req, *, timeout=30, opener_factory=None): def _handle_tts(handler, data, text, rate_str, engine): # ── ElevenLabs TTS ────────────────────────────────────────────────── return None + + +def handle_get(handler, parsed) -> bool: + """Handle all GET routes. Returns True if handled, False for 404.""" + return False + + +def handle_post(handler, parsed) -> bool: + """Pinned POST route anchors for the Atlas voice patch.""" + if parsed.path == "/api/transcribe": + return handle_transcribe(handler) + + if parsed.path == "/api/tts": + return _handle_tts(handler, parsed) + return False diff --git a/testing/tests/data/atlas_voice_language_probe.js b/testing/tests/data/atlas_voice_language_probe.js index 75966afd..05e071d5 100644 --- a/testing/tests/data/atlas_voice_language_probe.js +++ b/testing/tests/data/atlas_voice_language_probe.js @@ -77,13 +77,14 @@ function makeElement(id) { }; } -function makeHarness() { +function makeHarness(options = {}) { const clock = { now: 1000000 }; const timeouts = []; const intervals = new Map(); let timerId = 1; const ttsRequests = []; + const ttsStreamRequests = []; const transcribeCalls = []; const toasts = []; const sends = []; @@ -94,6 +95,7 @@ function makeHarness() { let assistantRows = []; let loud = false; let recorder = null; + let lastAudio = null; const storage = new Map(); const elements = {}; @@ -134,17 +136,28 @@ function makeHarness() { this.currentTime = 0; this.onended = null; this.onerror = null; - this.pause = () => {}; + this.paused = false; + this.pause = () => { this.paused = true; }; this.play = () => { - setImmediate(() => { if (this.onended) this.onended(); }); + this.paused = false; + if (!options.manualAudio) setImmediate(() => { if (this.onended) this.onended(); }); return Promise.resolve(); }; + this.finish = () => { if (this.onended) this.onended(); }; + lastAudio = this; } async function fetchStub(url, init) { if (url === '/api/transcribe/capability') { return { ok: true, status: 200, json: async () => capability }; } + if (url === '/api/voice/streaming/capability') { + return { + ok: !!options.streamingCapability, + status: options.streamingCapability ? 200 : 404, + json: async () => options.streamingCapability || {}, + }; + } if (url === '/api/transcribe') { transcribeCalls.push({ body: init && init.body }); return { @@ -162,10 +175,20 @@ function makeHarness() { json: async () => ({}), }; } + if (url === '/api/tts/stream') { + ttsStreamRequests.push(JSON.parse(init.body)); + return { + ok: !options.ttsStreamFails, + status: options.ttsStreamFails ? 503 : 200, + body: null, + headers: { get() { return null; } }, + }; + } throw new Error(`unexpected fetch: ${url}`); } const context = { + AbortController, console, Uint8Array, Promise, @@ -188,8 +211,10 @@ function makeHarness() { FormData: function FormData() { this.entries = []; this.append = (k, v) => this.entries.push([k, v]); }, URL: { createObjectURL: () => 'blob:atlas-test', revokeObjectURL() {} }, Audio: AudioElement, + AudioWorkletNode: options.enableAudioWorklet ? function AudioWorkletNode() {} : undefined, MediaRecorder, AudioContext, + ReadableStream: options.enableAudioWorklet ? function ReadableStream() {} : undefined, fetch: fetchStub, localStorage: { getItem: (key) => (storage.has(key) ? storage.get(key) : null), @@ -251,11 +276,13 @@ function makeHarness() { context, elements, ttsRequests, + ttsStreamRequests, transcribeCalls, toasts, sends, clock, recorder: () => recorder, + get lastAudio() { return lastAudio; }, setLoud: (value) => { loud = value; }, setCapability: (value) => { capability = value; }, setTranscribeResponse: (value, status) => { @@ -449,6 +476,89 @@ scenarios.transcribe_error_speaks_nothing = async () => { return { tts: harness.ttsRequests, toasts: harness.toasts }; }; +scenarios.adaptive_chunks_are_sentence_gated = async () => { + const harness = makeHarness(); + await harness.flush(); + const partial = harness.context._atlasAdaptiveChunks( + 'Dr. Rivera is still explaining this opening thought without a safe sentence boundary yet', + false, + ); + const complete = harness.context._atlasAdaptiveChunks( + 'This opening clause gives Hermes a clean and quick place to begin speaking, while the rest of the first sentence remains coherent. ' + + 'The following explanation is deliberately long enough to demonstrate that later punctuation aligned chunks use a much larger target and retain natural prosody for listeners. ' + + 'A compact final tail remains.', + true, + ); + return { partial, complete }; +}; + +scenarios.first_sentence_speaks_before_stream_completion = async () => { + const harness = makeHarness(); + await startVoiceMode(harness); + harness.setTranscribeResponse({ ok: true, transcript: 'Tell me something useful.', language: 'en' }); + await captureSpeech(harness); + harness.setAssistantReply('This answer is still streaming without a complete sentence'); + harness.tick(100); + await harness.flush(); + const beforeBoundary = harness.ttsRequests.length; + harness.setAssistantReply('This answer now has its first complete sentence. The remainder is still'); + harness.tick(100); + await harness.flush(); + const afterBoundary = harness.ttsRequests.length; + harness.setAssistantReply('This answer now has its first complete sentence. The remainder is still being generated and is now complete.'); + harness.context.autoReadLastAssistant(); + await harness.flush(); + return { beforeBoundary, afterBoundary, tts: harness.ttsRequests }; +}; + +scenarios.final_renderer_revision_closes_speech_queue = async () => { + const harness = makeHarness(); + await startVoiceMode(harness); + harness.setTranscribeResponse({ ok: true, transcript: 'Explain this.', language: 'en' }); + await captureSpeech(harness); + harness.setAssistantReply('Hermes starts with a complete sentence. The draft tail is still'); + harness.tick(100); + await harness.flush(); + const beforeRevision = harness.ttsRequests.length; + harness.setAssistantReply('The renderer revised the complete sentence. The final tail is now safe and complete.'); + harness.context.autoReadLastAssistant(); + await harness.flush(); + return { beforeRevision, afterRevision: harness.ttsRequests.length }; +}; + +scenarios.streaming_tts_failure_falls_back_to_wav = async () => { + const harness = makeHarness({ + enableAudioWorklet: true, + ttsStreamFails: true, + streamingCapability: { + tts: { available: true, transport: 'http', format: 'pcm_s16le', sample_rate: 22050 }, + stt: { available: false }, + }, + }); + await startVoiceMode(harness); + await runTurn(harness, { transcript: 'Fallback please.', language: 'en', reply: 'A complete answer.' }); + return { stream: harness.ttsStreamRequests, wav: harness.ttsRequests }; +}; + +scenarios.one_ahead_is_bounded_and_turn_cancel_stops_audio = async () => { + const harness = makeHarness({ manualAudio: true }); + await startVoiceMode(harness); + const reply = 'This opening sentence is sufficiently complete for a short first speech chunk. ' + + 'This second section contains enough carefully chosen words to become a larger punctuation aligned chunk without losing its natural rhythm or clarity for the listener. ' + + 'This third section is also deliberately long enough to require another queued synthesis request after playback advances.'; + await runTurn(harness, { transcript: 'Read it.', language: 'en', reply }); + const beforeFirstEnds = harness.ttsRequests.length; + const firstAudio = harness.lastAudio; + harness.elements.btnVoiceMode.click(); + await harness.flush(); + return { + beforeFirstEnds, + afterCancel: harness.ttsRequests.length, + active: harness.context._voiceModeActive(), + firstAudioPaused: !!(firstAudio && firstAudio.paused), + }; +}; + (async () => { const output = {}; const names = Object.keys(scenarios); diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py index d6fdca7e..fb2624d6 100644 --- a/testing/tests/test_hermes_chat_quality.py +++ b/testing/tests/test_hermes_chat_quality.py @@ -321,9 +321,15 @@ def test_chat_voice_uses_private_jetson_services_and_shared_auto_route(): assert env["HERMES_STT_URL"] == ( "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions" ) + assert env["HERMES_WEBUI_ATLAS_STT_STREAM_URL"] == ( + "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions/stream" + ) assert env["HERMES_WEBUI_ATLAS_TTS_URL"] == ( "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech" ) + assert env["HERMES_WEBUI_ATLAS_TTS_STREAM_URL"] == ( + "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech/stream" + ) assert "hermes_stt_client.py" in env["HERMES_LOCAL_STT_COMMAND"] configmap = _documents(HERMES / "chat-configmap.yaml")[0] @@ -342,11 +348,14 @@ def test_chat_voice_uses_private_jetson_services_and_shared_auto_route(): assert "/api/transcribe" in voice_script assert "/api/tts" in voice_script assert "speakResponse(generation)" in voice_script - assert "window._splitForTTS(text,80)" in voice_script + assert "window._atlasAdaptiveChunks" in voice_script + assert "preferredCut(firstSlice,40,52,60)" in voice_script + assert "preferredCut(remaining,100,120,140)" in voice_script assert "hermes-atlas-voice-latency-v2" in voice_script assert "hermes-voice-silence-ms','1100'" in voice_script - assert "pending=fetchSpeech(chunks[index+1],language)" in voice_script - assert "restartSoon(token,450)" in voice_script + assert "const nextPrepared=next.then" in voice_script + assert "At most one later synthesis request exists" in voice_script + assert "restartSoon(turn.token,300)" in voice_script assert "constraints.voiceIsolation=true" in voice_script assert "highpass.frequency.value=140" in voice_script assert "Math.max(0.04,noiseFloor*2.4+0.006)" in voice_script @@ -359,6 +368,21 @@ def test_chat_voice_uses_private_jetson_services_and_shared_auto_route(): assert "no_speech_threshold=0.5" in stt_server +def test_worker_webui_rollouts_do_not_repeat_completed_state_migration(): + deployment = _documents(HERMES / "deployment.yaml")[0] + pod_spec = deployment["spec"]["template"]["spec"] + init = next( + item + for item in pod_spec["initContainers"] + if item["name"] == "patch-api-server-sessions" + ) + + assert pod_spec["securityContext"]["fsGroupChangePolicy"] == "OnRootMismatch" + command = "\n".join(init["command"]) + assert "patch_api_server_sessions.py" in command + assert "migrate_api_session_lineage.py" not in command + + def test_voice_transcript_filter_removes_fan_hallucinations(monkeypatch): server_path = ROOT / "dockerfiles" / "hermes-jetson-stt-server.py" spec = importlib.util.spec_from_file_location("hermes_jetson_stt_server", server_path) diff --git a/testing/tests/test_hermes_stt_streaming.py b/testing/tests/test_hermes_stt_streaming.py new file mode 100644 index 00000000..6ccba386 --- /dev/null +++ b/testing/tests/test_hermes_stt_streaming.py @@ -0,0 +1,241 @@ +"""Contract tests for Hermes' bounded, speculative PCM transcription stream.""" + +from __future__ import annotations + +import importlib.util +import io +import struct +import sys +import threading +import time +from pathlib import Path +from types import SimpleNamespace + +import pytest + + +ROOT = Path(__file__).resolve().parents[2] +SERVER = ROOT / "dockerfiles" / "hermes-jetson-stt-server.py" + + +def _load_server(monkeypatch): + """Import the server without requiring the image's CUDA dependencies.""" + spec = importlib.util.spec_from_file_location("hermes_stt_stream_server", SERVER) + assert spec and spec.loader + module = importlib.util.module_from_spec(spec) + monkeypatch.setitem(sys.modules, "cgi", SimpleNamespace()) + monkeypatch.setitem( + sys.modules, + "torch", + SimpleNamespace(cuda=SimpleNamespace(is_available=lambda: False)), + ) + monkeypatch.setitem(sys.modules, "whisper", SimpleNamespace()) + spec.loader.exec_module(module) + return module + + +class _Connection: + """Capture server messages without constructing an HTTP handler.""" + + def __init__(self) -> None: + self.messages: list[dict] = [] + self.condition = threading.Condition() + + def send_json(self, payload: dict) -> None: + with self.condition: + self.messages.append(payload) + self.condition.notify_all() + + def wait_for(self, message_type: str, timeout: float = 2.0) -> dict: + deadline = time.monotonic() + timeout + with self.condition: + while time.monotonic() < deadline: + match = next( + ( + message + for message in self.messages + if message.get("type") == message_type + ), + None, + ) + if match: + return match + self.condition.wait(deadline - time.monotonic()) + raise AssertionError(f"no {message_type} response: {self.messages}") + + +class _Model: + """Return a deterministic Whisper-shaped result and retain decode options.""" + + def __init__(self) -> None: + self.calls: list[tuple[str, dict]] = [] + + def transcribe(self, path: str, **options: object) -> dict: + self.calls.append((path, options)) + return { + "text": " hello", + "language": "en", + "segments": [ + { + "text": " hello", + "no_speech_prob": 0.01, + "avg_logprob": -0.1, + } + ], + } + + +def _start(session) -> None: + session.start( + { + "type": "start", + "turn_id": "turn-10", + "format": "pcm_s16le", + "sample_rate": 16_000, + "language": "auto", + } + ) + + +def _speech(samples: int = 1_600) -> bytes: + return struct.pack(f"<{samples}h", *([12_000] * samples)) + + +def _silence(samples: int = 12_000) -> bytes: + return bytes(samples * 2) + + +def test_websocket_handshake_matches_rfc_example(monkeypatch): + module = _load_server(monkeypatch) + + assert ( + module.WebSocketConnection.accept_value("dGhlIHNhbXBsZSBub25jZQ==") + == "s3pPLMBiTxaQ9kYGzzhZRbK+xOo=" + ) + + +def test_websocket_read_exact_accumulates_partial_socket_reads(monkeypatch): + module = _load_server(monkeypatch) + payload = b'{"type":"resume"}' + mask = b"mask" + masked = bytes(value ^ mask[index % 4] for index, value in enumerate(payload)) + frame = bytes([0x81, 0x80 | len(payload)]) + mask + masked + + class ShortReader(io.BytesIO): + def read(self, size: int = -1) -> bytes: + return super().read(1 if size > 0 else size) + + handler = SimpleNamespace(rfile=ShortReader(frame), wfile=io.BytesIO()) + connection = module.WebSocketConnection(handler) + + assert connection.receive() == (0x1, payload) + + +def test_stream_speculation_is_reused_for_unchanged_commit(monkeypatch): + module = _load_server(monkeypatch) + connection = _Connection() + model = _Model() + session = module.StreamingTranscription(connection, model) + _start(session) + + session.append(_speech()) + session.append(_silence()) + partial = connection.wait_for("partial") + assert partial == { + "type": "partial", + "turn_id": "turn-10", + "transcript": "hello", + "language": "en", + "speculative": True, + } + + session.commit() + final = connection.wait_for("final") + assert final["transcript"] == "hello" + assert final["language"] == "en" + assert len(model.calls) == 1 + assert model.calls[0][1]["language"] is None + assert model.calls[0][1]["condition_on_previous_text"] is False + + +def test_resume_invalidates_speculative_snapshot(monkeypatch): + module = _load_server(monkeypatch) + connection = _Connection() + model = _Model() + session = module.StreamingTranscription(connection, model) + _start(session) + + session.append(_speech()) + session.append(_silence()) + connection.wait_for("partial") + session.resume() + session.append(_speech(800)) + session.commit() + + assert len(model.calls) == 2 + assert connection.wait_for("final")["turn_id"] == "turn-10" + + +def test_final_snapshot_preserves_resumed_audio_and_only_trims_tail(monkeypatch): + module = _load_server(monkeypatch) + session = module.StreamingTranscription(_Connection(), _Model()) + _start(session) + voiced = _speech() + session.append(voiced) + session.append(_silence()) + + with session._lock: + trimmed, _ = session._snapshot_locked() + tail_bytes = ( + module.STREAM_SAMPLE_RATE + * module.STREAM_SAMPLE_WIDTH + * module.VAD_TAIL_MS + // 1000 + ) + assert len(trimmed) == len(voiced) + tail_bytes + + session.resume() + quiet_resumed_speech = struct.pack("<800h", *([100] * 800)) + session.append(quiet_resumed_speech) + with session._lock: + resumed, _ = session._snapshot_locked() + assert resumed.endswith(quiet_resumed_speech) + + +def test_stream_contract_rejects_wrong_format_and_oversize(monkeypatch): + module = _load_server(monkeypatch) + session = module.StreamingTranscription(_Connection(), _Model()) + with pytest.raises(module.WebSocketError, match="pcm_s16le"): + session.start( + { + "turn_id": "turn-10", + "format": "webm", + "sample_rate": 16_000, + } + ) + + malformed_rate = module.StreamingTranscription(_Connection(), _Model()) + with pytest.raises(module.WebSocketError, match="invalid sample rate"): + malformed_rate.start( + { + "turn_id": "turn-10", + "format": "pcm_s16le", + "sample_rate": "sixteen kilohertz", + } + ) + + session = module.StreamingTranscription(_Connection(), _Model()) + _start(session) + monkeypatch.setattr(module, "MAX_STREAM_AUDIO_BYTES", 4) + with pytest.raises(module.WebSocketError, match="too long"): + session.append(bytes(6)) + + +def test_stream_health_contract_is_declared_in_source(): + source = SERVER.read_text() + + assert '"path": "/v1/audio/transcriptions/stream"' in source + assert '"format": "pcm_s16le"' in source + assert '"server_vad": True' in source + assert '"speculative": True' in source + assert "MAX_STREAM_AUDIO_BYTES" in source diff --git a/testing/tests/test_hermes_tts_language_routing.py b/testing/tests/test_hermes_tts_language_routing.py index 9e29941b..10c04d4d 100644 --- a/testing/tests/test_hermes_tts_language_routing.py +++ b/testing/tests/test_hermes_tts_language_routing.py @@ -222,6 +222,154 @@ def test_default_voice_name_is_one_of_the_baked_voices(tts): assert tts.DEFAULT_VOICE_NAME in tts.BAKED_VOICE_NAMES +@pytest.mark.parametrize( + "value,expected", + [ + ("turn-42", "turn-42"), + (" session_1:chunk.3 ", "session_1:chunk.3"), + (None, None), + (42, None), + ("", None), + ("turn\r\nX-Injected: true", None), + ("/not/header-safe", None), + ("x" * 129, None), + ], +) +def test_turn_ids_are_safe_to_echo_in_response_headers(tts, value, expected): + assert tts.normalize_turn_id(value) == expected + + +def _decode_chunked_response(data: bytes) -> bytes: + decoded = bytearray() + cursor = 0 + while True: + line_end = data.index(b"\r\n", cursor) + size = int(data[cursor:line_end], 16) + cursor = line_end + 2 + if size == 0: + assert data[cursor:] == b"\r\n" + break + decoded.extend(data[cursor : cursor + size]) + cursor += size + assert data[cursor : cursor + 2] == b"\r\n" + cursor += 2 + return bytes(decoded) + + +def test_stream_endpoint_returns_progressive_pcm_and_preserves_voice_policy(tts): + events: list[str] = [] + first_pcm = b"\x01\x00" * (tts.STREAM_WRITE_BYTES // 2 + 3) + second_pcm = b"\x02\x00\x03\x00" + + class _StreamingVoice: + config = SimpleNamespace(sample_rate=22_050) + + def synthesize(self, text, syn_config): + assert text == "Hola mundo. Otra frase." + assert syn_config.length_scale == pytest.approx(0.8) + assert "headers" in events + events.append("synthesis") + yield SimpleNamespace( + sample_rate=22_050, + sample_width=2, + sample_channels=1, + audio_int16_bytes=first_pcm, + ) + yield SimpleNamespace( + sample_rate=22_050, + sample_width=2, + sample_channels=1, + audio_int16_bytes=second_pcm, + ) + + class _StreamingHandler(tts.SpeechHandler): + def __init__(self): + payload = json.dumps( + { + "text": "Hola mundo. Otra frase.", + "language": "es", + "speed": 1.25, + "turn_id": "turn-9:chunk.1", + } + ).encode("utf-8") + self.path = "/v1/audio/speech/stream" + self.headers = {"Content-Length": str(len(payload))} + self.rfile = io.BytesIO(payload) + self.wfile = io.BytesIO() + self.status = None + self.response_headers = {} + self.close_connection = False + self.server = SimpleNamespace( + voices={CLAUDE: _StreamingVoice()}, + default_voice_name=AMY, + ) + + def send_response(self, status, message=None): + self.status = status + + def send_header(self, name, value): + self.response_headers[name] = value + + def end_headers(self): + events.append("headers") + + handler = _StreamingHandler() + handler.do_POST() + + assert handler.status == 200 + assert handler.response_headers["Transfer-Encoding"] == "chunked" + assert handler.response_headers["X-Audio-Format"] == "pcm_s16le" + assert handler.response_headers["X-Audio-Sample-Rate"] == "22050" + assert handler.response_headers["X-Audio-Channels"] == "1" + assert handler.response_headers["X-Audio-Sample-Width"] == "2" + assert handler.response_headers["X-TTS-Voice"] == CLAUDE + assert handler.response_headers["X-TTS-Turn-ID"] == "turn-9:chunk.1" + assert events == ["headers", "synthesis"] + assert _decode_chunked_response(handler.wfile.getvalue()) == first_pcm + second_pcm + + +def test_stream_disconnect_stops_before_synthesizing_another_sentence(tts): + generated: list[int] = [] + + class _StreamingVoice: + config = SimpleNamespace(sample_rate=16_000) + + def synthesize(self, text, syn_config): + for sentence in (1, 2): + generated.append(sentence) + yield SimpleNamespace( + sample_rate=16_000, + sample_width=2, + sample_channels=1, + audio_int16_bytes=b"\x00\x00", + ) + + class _DisconnectedHandler(tts.SpeechHandler): + def __init__(self): + self.wfile = io.BytesIO() + self.response_headers = {} + self.close_connection = False + + def send_response(self, status, message=None): + return None + + def send_header(self, name, value): + self.response_headers[name] = value + + def end_headers(self): + return None + + def _write_stream_chunk(self, audio): + self.close_connection = True + return False + + handler = _DisconnectedHandler() + handler._stream_pcm("hello", 1.0, AMY, _StreamingVoice(), "turn-1") + + assert generated == [1] + assert handler.close_connection is True + + def test_load_voices_fails_closed_when_a_baked_model_is_missing(tts, tmp_path): with pytest.raises(RuntimeError, match="baked Piper voice is missing"): tts.load_voices(tmp_path, threads=1) diff --git a/testing/tests/test_hermes_voice_instrument.py b/testing/tests/test_hermes_voice_instrument.py index 65e0896d..12d6e20e 100644 --- a/testing/tests/test_hermes_voice_instrument.py +++ b/testing/tests/test_hermes_voice_instrument.py @@ -63,6 +63,10 @@ def test_real_upstream_fixture_receives_visual_instrument_contract(tmp_path: Pat assert layer in index assert ' bool: + """Handle all GET routes. Returns True if handled, False for 404.""" + return False + + +def handle_post(handler, parsed) -> bool: + """Pinned POST route anchors for the Atlas voice patch.""" + if parsed.path == "/api/transcribe": + return handle_transcribe(handler) + + if parsed.path == "/api/tts": + return _handle_tts(handler, parsed) + return False ''' UPLOAD_FIXTURE = '''"""Stand-in carrying the exact upstream /api/transcribe response anchor.""" @@ -607,7 +624,7 @@ def test_voice_mode_drops_hostile_language_values(voice_probe): def test_voice_mode_never_sends_a_voice_field(voice_probe): for request in voice_probe["voice_field_is_never_sent"]["tts"]: - assert set(request) <= {"text", "engine", "language"} + assert set(request) <= {"text", "engine", "language", "turn_id"} assert "voice" not in request @@ -648,6 +665,97 @@ def test_voice_mode_speaks_nothing_when_transcription_fails(voice_probe): assert any("Whisper is down" in toast for toast in result["toasts"]) +def test_adaptive_chunks_wait_for_sentence_then_change_size(voice_probe): + result = voice_probe["adaptive_chunks_are_sentence_gated"] + assert result["partial"] == [] + chunks = result["complete"] + assert 40 <= len(chunks[0]) <= 60 + assert all(100 <= len(chunk) <= 140 for chunk in chunks[1:-1]) + assert "".join(chunks).replace(" ", "") + + +def test_first_complete_sentence_reaches_tts_before_completion_callback(voice_probe): + result = voice_probe["first_sentence_speaks_before_stream_completion"] + assert result["beforeBoundary"] == 0 + assert result["afterBoundary"] == 1 + assert len(result["tts"]) >= 2 + + +def test_final_renderer_revision_closes_queue_without_replaying_prefix(voice_probe): + result = voice_probe["final_renderer_revision_closes_speech_queue"] + assert result["beforeRevision"] == 1 + assert result["afterRevision"] >= 2 + + +def test_streaming_tts_failure_falls_back_to_complete_wav(voice_probe): + result = voice_probe["streaming_tts_failure_falls_back_to_wav"] + assert len(result["stream"]) == 1 + assert len(result["wav"]) == 1 + assert result["stream"][0]["turn_id"] == result["wav"][0]["turn_id"] + + +def test_one_ahead_bound_and_turn_cancellation(voice_probe): + result = voice_probe["one_ahead_is_bounded_and_turn_cancel_stops_audio"] + assert result["beforeFirstEnds"] == 2 + assert result["afterCancel"] == 2 + assert result["active"] is False + assert result["firstAudioPaused"] is True + + +def test_authenticated_voice_websocket_uses_rendered_csrf_subprotocol( + patched_webui, monkeypatch +): + routes = patched_webui.routes + monkeypatch.setattr( + routes, "_check_same_origin_browser_request", lambda _handler: True, raising=False + ) + monkeypatch.setitem( + sys.modules, + "api.auth", + SimpleNamespace( + csrf_token_for_session=lambda _cookie: "csrf-token", + is_auth_enabled=lambda: True, + parse_cookie=lambda _handler: None, + verify_session=lambda value: value == "session-cookie", + ), + ) + handler = SimpleNamespace( + _trusted_auth_session_cookie_value="session-cookie", + headers={ + "Origin": "https://chat.bstein.dev", + "Sec-WebSocket-Protocol": ( + "hermes-voice-v1, hermes-csrf.csrf-token" + ), + } + ) + + assert routes._atlas_ws_authorized(handler) is True + handler.headers["Sec-WebSocket-Protocol"] = "hermes-voice-v1, hermes-csrf.wrong" + assert routes._atlas_ws_authorized(handler) is False + handler.headers["Origin"] = "" + assert routes._atlas_ws_authorized(handler) is False + + +def test_streaming_tts_payload_is_narrow_and_turn_bound(patched_webui): + payload = patched_webui.routes._atlas_tts_stream_payload( + { + "text": "A safe sentence.", + "language": "ru", + "turn_id": "voice-turn-7", + "voice": "../../untrusted", + "speed": 99, + } + ) + + assert payload == { + "model": "piper", + "input": "A safe sentence.", + "speed": 1.0, + "language": "ru", + "turn_id": "voice-turn-7", + } + + # --------------------------------------------------------------------------- # 6. Build-time enforcement and documented semantics # ---------------------------------------------------------------------------