From c690123d48c48673dd0d586fb45b022d6ce0d71f Mon Sep 17 00:00:00 2001 From: Jason McCartney Date: Wed, 23 Sep 2026 15:15:16 -0700 Subject: [PATCH 1/3] Stream Spark speech with bounded audio and canonical voice profiles --- backends/spark-audio/Dockerfile | 56 + backends/spark-audio/README.md | 28 + .../spark-audio/lloom_audio_cuda_server.py | 1174 +++++++++++++++++ .../test_lloom_audio_cuda_server.py | 1018 ++++++++++++++ package.json | 4 +- src/server.mjs | 91 +- test/speech-stream.test.mjs | 273 ++++ 7 files changed, 2634 insertions(+), 10 deletions(-) create mode 100644 backends/spark-audio/Dockerfile create mode 100644 backends/spark-audio/README.md create mode 100644 backends/spark-audio/lloom_audio_cuda_server.py create mode 100644 backends/spark-audio/test_lloom_audio_cuda_server.py create mode 100644 test/speech-stream.test.mjs diff --git a/backends/spark-audio/Dockerfile b/backends/spark-audio/Dockerfile new file mode 100644 index 0000000..733a990 --- /dev/null +++ b/backends/spark-audio/Dockerfile @@ -0,0 +1,56 @@ +# Pinned LLooM local CUDA audio adapter image. +# +# The base is the NGC PyTorch container pinned by digest; it already ships the +# CUDA-matched ``torch``/``torchaudio`` builds, so this Dockerfile NEVER installs +# or upgrades torch -- the parent build handles that. +# +# docker build -t lloom-audio-cuda . +# +# Runtime (single model per process): +# docker run --gpus all -p 8220:8220 \ +# -v /models/Qwen3-TTS-12Hz-0.6B-CustomVoice:/models/tts:ro \ +# lloom-audio-cuda \ +# --kind tts --model Qwen3-TTS-12Hz-0.6B-CustomVoice --model-path /models/tts \ +# --host 0.0.0.0 --port 8220 + +FROM nvcr.io/nvidia/pytorch@sha256:bbc2b67e2533edd63ff1496bb1ed00a00338cdc1478af6c1a0bf9f4b369977e7 + +ENV DEBIAN_FRONTEND=noninteractive \ + PYTHONUNBUFFERED=1 \ + PIP_NO_CACHE_DIR=1 + +# ffmpeg decodes uploaded audio for Whisper; sox + libsndfile cover the +# reference-audio decode path (soundfile) and general audio tooling. +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + ffmpeg \ + sox \ + libsndfile1 \ + libsndfile1-dev \ + && apt-get clean \ + && rm -rf /var/lib/apt/lists/* + +# Pinned Python runtime dependencies. ``qwen-tts`` is installed --no-deps so it +# cannot drag in a clashing torch/transformers build; the versions it depends on +# are pinned explicitly right below. torch / torchaudio stay as shipped by NGC. +RUN python -m pip install --no-cache-dir --no-deps "qwen-tts==0.1.1" \ + && python -m pip install --no-cache-dir \ + "transformers==4.57.3" \ + "accelerate==1.12.0" \ + "huggingface-hub==0.36.2" \ + "openai-whisper==20250625" \ + "fastapi==0.141.1" \ + "uvicorn==0.53.0" \ + "python-multipart==0.0.32" + +# Optional CUDA graph engine, pinned and installed without changing torch. +RUN python -m pip install --no-cache-dir --no-deps "faster-qwen3-tts==0.2.6" + +WORKDIR /opt/lloom-audio +COPY lloom_audio_cuda_server.py /opt/lloom-audio/lloom_audio_cuda_server.py + +EXPOSE 8220 + +# Local-only by default; override --host/--port at ``docker run`` time. +ENTRYPOINT ["python", "/opt/lloom-audio/lloom_audio_cuda_server.py"] +CMD ["--help"] diff --git a/backends/spark-audio/README.md b/backends/spark-audio/README.md new file mode 100644 index 0000000..9bda419 --- /dev/null +++ b/backends/spark-audio/README.md @@ -0,0 +1,28 @@ +# Spark speech adapter + +Serves one existing local Qwen3-TTS or Whisper checkpoint per process. The stock +engine remains the default. Set `LLOOM_TTS_ENGINE=faster` for CUDA graph voice +cloning using `faster-qwen3-tts==0.2.6` (MIT, upstream commit +`5ae964a39fadec39141db779f5c5e4768ddd2b22`). This version matches Qwen TTS 0.1.1 +and Transformers 4.57.3; install it without replacing the CUDA-matched PyTorch. + +`stream: true` with `response_format: pcm` emits mono 24 kHz signed little-endian +PCM16 as synthesis progresses. Four codec frames per chunk balance onset and +decode overhead. WAV remains buffered. Named profiles are resolved by the +LLooM gateway and their bounded reference audio is sent inline; the adapter +never opens client-supplied paths. Profile sampling parameters are preserved. + +The producer owns the single GPU lease until it actually exits. Cancellation +stops further chunks, closes the generator, and cannot release a replacement +request's lease. Queue capacity and reference/prompt caches are bounded. +A failed partial stream closes the connection rather than appending JSON or SSE +to audio. Cold CUDA graph capture and first-use voice conditioning require a +warm-up before latency measurement. A loaded GPU can remain slower than real +time even when first audio arrives earlier; measure both onset and sustained +production against the same concurrent workload. + +Run `pytest -q backends/spark-audio/test_lloom_audio_cuda_server.py` in an isolated +environment with FastAPI, httpx, numpy, soundfile, python-multipart, and pytest. +These tests use fake models. Hardware acceptance must additionally prove PCM +headers, named-profile routing, streaming onset, interruption/recovery, and +transcription through the selected gateway on the actual host. diff --git a/backends/spark-audio/lloom_audio_cuda_server.py b/backends/spark-audio/lloom_audio_cuda_server.py new file mode 100644 index 0000000..3d1a688 --- /dev/null +++ b/backends/spark-audio/lloom_audio_cuda_server.py @@ -0,0 +1,1174 @@ +"""Standalone local FastAPI CUDA audio adapter for LLooM. + +One process serves exactly one pre-selected model: + + * ``--kind tts --model --model-path `` loads a local + Qwen3-TTS checkpoint via ``Qwen3TTSModel.from_pretrained`` on ``cuda:0``. + * ``--kind stt --model --model-path `` loads a + local Whisper checkpoint via ``whisper.load_model(path, device="cuda")``. + +``--model-path`` is a *required* local directory for both kinds: the STT path is +the actual Whisper checkpoint directory, so the process never performs an +implicit HuggingFace/OpenAI download at request time. + +The model id is *fixed at startup*. Requests may echo that exact id; any other +id (or a request that tries to pick a different model) is rejected. The server +never reads a caller-supplied filesystem path and never fetches a caller-supplied +URL: the only audio accepted for Base/clone is inline multipart upload or a +``data:`` base64 payload. Requests are bounded (body, text length, upload size, +reference duration, ``max_new_tokens``) and GPU work is offloaded to a worker +thread. + +Concurrency model +----------------- +Every async route hands the *entire* inference to +``anyio.to_thread.run_sync(..., abandon_on_cancel=False)`` so the event loop is +never blocked and cancellations/disconnects cannot abandon GPU work halfway. +Only one generation may run at a time; a second concurrent request is rejected +with HTTP 429 instead of queueing unboundedly. The "busy" flag is held until the +worker thread *actually* completes, even if the client disconnects. + +Region of the OpenAI speech/transcriptions API implemented (WAV or PCM): + + POST /v1/audio/speech (JSON or multipart) + POST /v1/audio/transcriptions (multipart) + GET /health + GET /v1/models +""" +from __future__ import annotations + +import argparse +import base64 +import binascii +import asyncio +import hashlib +import io +import json +import logging +import math +import os +import queue +import sys +import tempfile +import threading +import wave +from dataclasses import dataclass, field +from concurrent.futures import ThreadPoolExecutor +from typing import Any, AsyncIterator, Dict, Optional, Sequence, Tuple + +import anyio +import numpy as np +from fastapi import FastAPI, Request +from fastapi.responses import JSONResponse, Response, StreamingResponse + +LOG = logging.getLogger("lloom-audio-cuda") + +# --------------------------------------------------------------------------- +# Limits / constants +# --------------------------------------------------------------------------- + +MAX_BODY_BYTES = 25 * 1024 * 1024 # hard cap on a whole request body +MAX_UPLOAD_BYTES = 16 * 1024 * 1024 # per uploaded reference clip / STT file +MAX_TEXT_CHARS = 4000 # synthesis text length +MAX_REF_TEXT_CHARS = 4000 # reference transcript length +MAX_NEW_TOKENS = 2048 # capping generated audio tokens +MAX_STREAM_QUEUE_CHUNKS = 4 # bounded queue provides backpressure +PCM_SAMPLE_RATE = 24000 +PCM_CHANNELS = 1 +MIN_REF_SECONDS = 0.05 # reject empty / degenerate clips +MAX_REF_SECONDS = 30.0 # reject oversized reference audio +MAX_STT_SECONDS = 600.0 # whisper transcribe cap (10 minutes) +DEFAULT_LANGUAGE = "Auto" # Qwen3-TTS language when unset + +# Preset Qwen3-TTS CustomVoice speakers. These are the only speaker ids the +# CustomVoice mode will forward; anything else is rejected (no ad-hoc profiles). +CUSTOM_VOICE_SPEAKERS: Tuple[str, ...] = ( + "serena", + "vivian", + "uncle_fu", + "ryan", + "aiden", + "ono_anna", + "sohee", + "eric", + "dylan", +) + +TTS_MODES: Tuple[str, ...] = ("custom_voice", "voice_design", "voice_clone") + +# The gateway/OpenAI model ids this adapter is willing to serve. These are only +# cosmetic aliases for the fixed local checkpoint. +KNOWN_STT_GATEWAY_IDS: Tuple[str, ...] = ( + "openai/whisper-large-v3-turbo", + "whisper-large-v3-turbo", + "whisper-1", +) +SUPPORTED_AUDIO_SUFFIXES: Tuple[str, ...] = ( + ".wav", + ".mp3", + ".flac", + ".ogg", + ".opus", + ".m4a", + ".aac", + ".webm", +) + + +class ClientError(Exception): + """A request-level fault that maps to an HTTP status (never leaks details).""" + + def __init__(self, message: str, status_code: int = 400) -> None: + super().__init__(message) + self.message = message + self.status_code = status_code + + +class BusyError(Exception): + """Raised when a second concurrent generation is attempted.""" + + +class BodyTooLarge(Exception): + """Raised internally when a streamed request body exceeds the hard cap.""" + + +def _infer_tts_mode(model_id: str) -> str: + """Infer the Qwen3-TTS operating mode from the (fixed) model id.""" + text = (model_id or "").lower() + if "voicedesign" in text or "voice-design" in text or "voice_design" in text: + return "voice_design" + if "customvoice" in text or "custom-voice" in text or "custom_voice" in text: + return "custom_voice" + if "base" in text or "clone" in text: + return "voice_clone" + return "voice_clone" + + +# --------------------------------------------------------------------------- +# Bounded body handling +# --------------------------------------------------------------------------- + + +async def _bounded_receive_body(receive, limit: int = MAX_BODY_BYTES) -> bytes: + """Read a full request body using the ASGI ``receive`` channel, bounded. + + Content-Length is only a hint (and may be absent for chunked bodies), so the + running total is capped while streaming. This enforces the total request cap + for *both* JSON and multipart before any parsing happens, with no unbounded + buffering. + """ + if receive is None: + raise ClientError("request body is unavailable", status_code=400) + + chunks = [] + total = 0 + while True: + message = await receive() + message_type = message.get("type") + if message_type == "http.disconnect": + raise ClientError("client disconnected", status_code=499) + if message_type != "http.request": + if message.get("more_body"): + continue + break + chunk = message.get("body", b"") or b"" + total += len(chunk) + if total > limit: + raise BodyTooLarge("request body too large") + if chunk: + chunks.append(chunk) + if not message.get("more_body", False): + break + return b"".join(chunks) + + +async def _read_bounded_body(request) -> bytes: + """Enforce the total body cap before any parsing (peek + replay).""" + content_length = request.headers.get("content-length") + if content_length is not None: + try: + declared = int(content_length) + except ValueError: + raise ClientError("invalid Content-Length") + if declared > MAX_BODY_BYTES: + raise ClientError("request body too large", status_code=413) + + try: + raw = await _bounded_receive_body(request.receive, MAX_BODY_BYTES) + except BodyTooLarge: + raise ClientError("request body too large", status_code=413) + + # Replay the bounded buffer to downstream parsers (request.body / form()). + state = {"done": False} + + async def _replay(): + if state["done"]: + return {"type": "http.request", "body": b"", "more_body": False} + state["done"] = True + return {"type": "http.request", "body": raw, "more_body": False} + + request._receive = _replay + return raw + + +# --------------------------------------------------------------------------- +# Duration helpers (decoded frames / real sample rate, never a header only) +# --------------------------------------------------------------------------- + + +def _decode_wav_duration(raw: bytes) -> Optional[float]: + """Exact duration from a RIFF/WAVE container using its own frame count.""" + if len(raw) < 44 or raw[0:4] != b"RIFF" or raw[8:12] != b"WAVE": + return None + pos = 12 + sample_rate = None + block_align = None + data_size = None + while pos + 8 <= len(raw): + chunk_id = raw[pos : pos + 4] + chunk_size = int.from_bytes(raw[pos + 4 : pos + 8], "little") + body = pos + 8 + if chunk_id == b"fmt " and chunk_size >= 16 and body + 16 <= len(raw): + channels = int.from_bytes(raw[body + 2 : body + 4], "little") + sample_rate = int.from_bytes(raw[body + 4 : body + 8], "little") + block_align = int.from_bytes(raw[body + 12 : body + 14], "little") + if not block_align: + bits = int.from_bytes(raw[body + 14 : body + 16], "little") + block_align = max(1, channels) * max(1, bits // 8) + elif chunk_id == b"data": + data_size = min(chunk_size, max(0, len(raw) - body)) + if chunk_size <= 0: + break + pos = body + chunk_size + (chunk_size & 1) + if not sample_rate or data_size is None: + return None + frame_count = data_size // max(1, block_align or 1) + return frame_count / float(sample_rate) + + +def _decoded_duration_seconds(audio: Any, sample_rate: int) -> float: + """Duration from the *actual* decoded frames and the true sample rate. + + Codec-agnostic: compressed references (FLAC/OGG/MP3/M4A) report their real + duration here instead of the ``None`` a WAV-header probe would return. + """ + sample_rate = int(sample_rate or 0) + if sample_rate <= 0: + raise ClientError("reference audio has an invalid sample rate") + arr = np.asarray(audio) + frames = int(arr.shape[0]) if arr.ndim else int(arr.size) + return frames / float(sample_rate) + + +def _decode_ref_audio(raw: bytes) -> Tuple[np.ndarray, int]: + """Decode uploaded reference bytes into ``(mono float32 array, sample rate)``.""" + if not raw: + raise ClientError("reference audio is empty") + if len(raw) > MAX_UPLOAD_BYTES: + raise ClientError("reference audio exceeds the upload limit", status_code=413) + + try: # imported lazily so unit tests can stub out decoding + import soundfile as sf + + audio, sample_rate = sf.read(io.BytesIO(raw), dtype="float32", always_2d=False) + except ClientError: + raise + except Exception as exc: # pragma: no cover - depends on installed codecs + raise ClientError("reference audio could not be decoded") from exc + + array = np.asarray(audio, dtype=np.float32) + if array.ndim > 1: + array = array.mean(axis=1) + if array.size == 0: + raise ClientError("reference audio is empty") + + # Enforce the duration window on real decoded frames, not a WAV header. + duration = _decoded_duration_seconds(array, sample_rate) + if duration < MIN_REF_SECONDS: + raise ClientError("reference audio is too short") + if duration > MAX_REF_SECONDS: + raise ClientError("reference audio is too long", status_code=413) + return np.ascontiguousarray(array, dtype=np.float32), int(sample_rate) + + +def _decode_base64_audio(value: str) -> Tuple[np.ndarray, int]: + payload = value.split(",", 1)[1] if value.startswith("data:") else value + try: + raw = base64.b64decode(payload, validate=True) + except (binascii.Error, ValueError) as exc: + raise ClientError("reference audio base64 could not be decoded") from exc + return _decode_ref_audio(raw) + + +def _to_wav_bytes(audio: Any, sample_rate: int) -> bytes: + """Serialise a float waveform to a 16-bit PCM WAV container.""" + array = np.asarray(audio, dtype=np.float32) + if array.ndim > 1: + array = array.reshape(-1) + array = np.clip(np.nan_to_num(array, nan=0.0, posinf=1.0, neginf=-1.0), -1.0, 1.0) + pcm = (array * 32767.0).astype(" bytes: + """Serialise a float waveform to raw PCM16 little-endian mono bytes.""" + array = np.asarray(audio, dtype=np.float32) + if array.ndim > 1: + array = array.reshape(-1) + array = np.clip(np.nan_to_num(array, nan=0.0, posinf=1.0, neginf=-1.0), -1.0, 1.0) + return (array * 32767.0).astype(" Dict[str, str]: + """Describe raw PCM explicitly instead of relying on network L16 byte order.""" + return { + "x-audio-sample-rate": str(sample_rate), + "x-audio-channels": str(PCM_CHANNELS), + "x-audio-format": "pcm_s16le", + } + + +# --------------------------------------------------------------------------- +# Server configuration + fixed model holder +# --------------------------------------------------------------------------- + + +@dataclass(frozen=True) +class ServerConfig: + kind: str + model_id: str + model_path: Optional[str] = None + host: str = "127.0.0.1" + port: int = 8220 + hf_home: Optional[str] = None + engine: Optional[str] = None + + def __post_init__(self) -> None: + kind = (self.kind or "").strip().lower() + if kind not in ("tts", "stt"): + raise ValueError("kind must be 'tts' or 'stt'") + if not self.model_id or not str(self.model_id).strip(): + raise ValueError("a model id is required") + + +class ModelHolder: + """Owns the single configured model, its load state and the inference lock.""" + + def __init__(self, config: ServerConfig) -> None: + self.config = config + self.kind = config.kind + self.model_id = config.model_id + self.mode = _infer_tts_mode(config.model_id) if config.kind == "tts" else None + self.model: Any = None + engine = (config.engine or os.getenv("LLOOM_TTS_ENGINE", "stock")).lower() + self.engine = engine + self.uses_faster_engine = engine in ("faster", "faster-qwen3-tts", "fasterqwen3tts") + self.ready = False + self.fatal = False + self.load_error: Optional[str] = None + self.gpu_lock = threading.Lock() + self.executor = ThreadPoolExecutor(max_workers=1, thread_name_prefix="audio-gpu") + # Guards model access and tracks whether a generation is in flight. + self._busy_lock = threading.Lock() + self._busy = False + self._busy_generation = 0 + self._reference_lock = threading.Lock() + # Process-lifetime staging for FasterQwen3TTS. Stable content-addressed + # paths let its in-memory voice-prompt cache be reused; this is not a + # persistent configuration/model cache and is removed with the holder. + self._reference_tmp = tempfile.TemporaryDirectory(prefix="lloom-qwen-tts-ref-") + self._reference_paths = set() + self._stream_queue: queue.Queue = queue.Queue(maxsize=MAX_STREAM_QUEUE_CHUNKS) + self._stream_stop = threading.Event() + + # -- id validation ---------------------------------------------------- + def resolve_model_id(self, requested: Optional[str]) -> str: + """Return the fixed id, rejecting any attempt to select another model.""" + if requested is None: + return self.model_id + requested = str(requested).strip() + if not requested: + return self.model_id + if requested == self.model_id: + return self.model_id + # For STT, accept canonical gateway aliases for this same checkpoint. + if self.kind == "stt": + fixed = self.model_id.strip().lower() + if requested.lower() in KNOWN_STT_GATEWAY_IDS and fixed in KNOWN_STT_GATEWAY_IDS: + return self.model_id + raise ClientError("unknown model", status_code=404) + + # -- lifecycle -------------------------------------------------------- + def load(self) -> None: + """Warm-load the fixed model. Called at startup; errors are fatal.""" + try: + if self.kind == "tts": + self.model = self._load_tts() + else: + self.model = self._load_stt() + except Exception as exc: # keep only the type name; never leak details + self.fatal = True + self.ready = False + self.load_error = type(exc).__name__ + LOG.error("model load failed (%s)", self.load_error) + raise + self.ready = True + self.fatal = False + + def _load_tts(self) -> Any: + import torch + + if self.uses_faster_engine: + if self.mode != "voice_clone": + raise ValueError("the optional faster engine is available for voice_clone only") + from faster_qwen3_tts import FasterQwen3TTS + + if not self.config.model_path or not os.path.isdir(self.config.model_path): + raise ValueError("TTS requires an existing local model directory") + # A local path prevents model download. The faster wrapper loads the + # same local checkpoint through Qwen3TTSModel. + return FasterQwen3TTS.from_pretrained( + self.config.model_path, + device="cuda:0", + dtype=torch.bfloat16, + attn_implementation="sdpa", + ) + + from qwen_tts import Qwen3TTSModel + + if not self.config.model_path or not os.path.isdir(self.config.model_path): + raise ValueError("TTS requires an existing local model directory") + return Qwen3TTSModel.from_pretrained( + self.config.model_path, + device_map="cuda:0", + dtype=torch.bfloat16, + attn_implementation="sdpa", + local_files_only=True, + ) + + def _load_stt(self) -> Any: + import whisper + + # ``--model-path`` is the actual local Whisper checkpoint. Passing it as + # the model name means whisper loads the local checkpoint file directly and + # never performs an implicit download (a bare alias such as "turbo" + # would trigger one), so a real path is required here. + path = self.config.model_path + if not path: + raise ValueError("STT requires --model-path pointing at a local checkpoint") + if not os.path.isfile(path): + raise ValueError("STT --model-path must be a checkpoint file") + return whisper.load_model(path, device="cuda") + + # -- busy gate -------------------------------------------------------- + def ensure_idle(self) -> None: + """Raise 429 when another generation is already in flight.""" + with self._busy_lock: + if self._busy: + raise BusyError("another generation is already running") + self._busy = True + self._busy_generation += 1 + return self._busy_generation + + def clear_busy(self, generation=None) -> None: + with self._busy_lock: + if generation is None or generation == self._busy_generation: + self._busy = False + + @property + def busy(self) -> bool: + with self._busy_lock: + return self._busy + + # -- execution -------------------------------------------------------- + def run(self, fn, *args, **kwargs): + """Run a blocking callable on the calling thread with the lock held.""" + with self.gpu_lock: + return fn(*args, **kwargs) + + async def run_in_thread(self, fn, *args, **kwargs): + """Offload blocking inference to a worker thread. + + ``abandon_on_cancel=False`` keeps the worker running to completion even + when the awaiting task is cancelled (e.g. the client disconnects), so the + busy flag / GPU lock are only released once inference has actually + finished. + """ + lease = self._busy_generation + future = self.executor.submit(lambda: self.run(fn, *args, **kwargs)) + future.add_done_callback(lambda _: self.clear_busy(lease)) + with anyio.CancelScope(shield=True): + return await anyio.to_thread.run_sync(future.result, abandon_on_cancel=False) + + def start_stream_job(self, fn): + """Consume the route's busy lease; each response owns its queue/event.""" + lease = self._busy_generation + chunks = queue.Queue(maxsize=MAX_STREAM_QUEUE_CHUNKS) + stopped = threading.Event() + self._stream_queue, self._stream_stop = chunks, stopped + + def put(item): + while not stopped.is_set(): + try: + chunks.put(item, timeout=0.05) + return True + except queue.Full: + continue + return False + + def job(): + with self.gpu_lock: + try: + fn(put, stopped) + except Exception as exc: + LOG.error("streaming generation failed (%s)", type(exc).__name__) + put(RuntimeError("speech generation failed")) + finally: + put(None) + + try: + future = self.executor.submit(job) + except BaseException: + self.clear_busy(lease) + raise + future.add_done_callback(lambda _: self.clear_busy(lease)) + return chunks, stopped + + + +# --------------------------------------------------------------------------- +# Request parsing helpers +# --------------------------------------------------------------------------- + + +def _first_str(body: Dict[str, Any], *keys: str) -> Optional[str]: + """Return the first non-empty value among ``keys`` (alias support).""" + for key in keys: + value = body.get(key) + if value is None: + continue + if isinstance(value, str) and not value.strip(): + continue + return value + return None + + +def _coerce_int(value: Any, default: Optional[int] = None) -> Optional[int]: + if value is None or value == "": + return default + try: + return int(value) + except (TypeError, ValueError): + raise ClientError("a numeric parameter was not an integer") + + +def _as_bool(value: Any) -> bool: + if value is None: + return False + return str(value).strip().lower() in ("1", "true", "yes", "on") + + +async def _read_json_body(request) -> Dict[str, Any]: + raw = await _read_bounded_body(request) + if not raw: + return {} + try: + parsed = json.loads(raw) + except json.JSONDecodeError as exc: + raise ClientError("request body was not valid JSON") from exc + if not isinstance(parsed, dict): + raise ClientError("request body must be a JSON object") + return parsed + + +@dataclass +class SpeechRequest: + model_id: str + text: str + language: Optional[str] + instruct: Optional[str] + speaker: Optional[str] + ref_audio: Optional[Tuple[np.ndarray, int]] + ref_text: Optional[str] + x_vector_only_mode: bool + max_new_tokens: int + stream: bool + response_format: str = "wav" + sampling: Dict[str, Any] = field(default_factory=dict) + + +def _build_speech_request(holder: ModelHolder, body: Dict[str, Any]) -> SpeechRequest: + model_id = holder.resolve_model_id(_first_str(body, "model")) + + text = _first_str(body, "input", "text") + if text is None: + raise ClientError("'input' (or 'text') is required") + text = str(text) + if len(text) > MAX_TEXT_CHARS: + raise ClientError("input text is too long", status_code=413) + + instruct = _first_str(body, "instructions", "instruct") + if instruct is not None: + instruct = str(instruct) + if len(instruct) > MAX_TEXT_CHARS: + raise ClientError("instructions are too long", status_code=413) + + language = _first_str(body, "language") + if language is not None: + language = str(language).strip() + if not language: + language = DEFAULT_LANGUAGE # "Auto" + + speaker = _first_str(body, "speaker", "voice") + ref_text = _first_str(body, "ref_text") + if ref_text is not None: + ref_text = str(ref_text) + if len(ref_text) > MAX_REF_TEXT_CHARS: + raise ClientError("ref_text is too long", status_code=413) + + max_new_tokens = _coerce_int(_first_str(body, "max_new_tokens"), default=MAX_NEW_TOKENS) + if max_new_tokens is None or max_new_tokens <= 0: + max_new_tokens = MAX_NEW_TOKENS + max_new_tokens = min(max_new_tokens, MAX_NEW_TOKENS) + + x_vector_only_mode = _as_bool(body.get("x_vector_only_mode")) + stream = _as_bool(body.get("stream")) + + sampling = {} + for key, low, high in [("temperature", 0.01, 2.0), ("top_p", 0.01, 1.0), + ("top_k", 1, 1000), ("repetition_penalty", 0.1, 3.0)]: + if body.get(key) is None: + continue + try: + value = float(body[key]) + except (ValueError, TypeError): + raise ClientError("invalid " + key) + if not math.isfinite(value) or not low <= value <= high: + raise ClientError("invalid " + key) + if key == "top_k" and value != int(value): + raise ClientError("top_k must be an integer") + sampling[key] = int(value) if key == "top_k" else value + + mode = holder.mode + ref_audio: Optional[Tuple[np.ndarray, int]] = None + + if mode == "custom_voice": + # Instructions are OPTIONAL for CustomVoice: a preset speaker alone is a + # valid request, and the preset default speaker is used when none is set. + selected = speaker if speaker is not None else CUSTOM_VOICE_SPEAKERS[0] + selected = str(selected).strip().lower() + if selected not in CUSTOM_VOICE_SPEAKERS: + raise ClientError("unknown speaker for CustomVoice", status_code=404) + speaker = selected + if body.get("ref_audio") is not None: + raise ClientError("CustomVoice does not accept reference audio") + elif mode == "voice_design": + # VoiceDesign *requires* an instruction. + if instruct is None: + raise ClientError("VoiceDesign requires 'instructions'") + speaker = None + elif mode == "voice_clone": + supplied = body.get("ref_audio") + if supplied is None: + raise ClientError("Base/clone requires reference audio") + if isinstance(supplied, str): + if supplied.startswith("data:") or ";base64," in supplied: + ref_audio = _decode_base64_audio(supplied) + else: + # Never read caller-specified paths or fetch caller URLs. + raise ClientError("ref_audio must be uploaded or base64-encoded, not a path") + elif isinstance(supplied, (bytes, bytearray, memoryview)): + ref_audio = _decode_ref_audio(bytes(supplied)) + else: + raise ClientError("ref_audio must be uploaded or base64-encoded") + if not x_vector_only_mode and not ref_text: + raise ClientError("Base/clone requires ref_text unless x_vector_only_mode") + speaker = None + else: # pragma: no cover - guarded by config validation + raise ClientError("unsupported model mode", status_code=500) + + return SpeechRequest( + model_id=model_id, + text=text, + language=language, + instruct=instruct, + speaker=speaker, + ref_audio=ref_audio, + ref_text=ref_text, + x_vector_only_mode=x_vector_only_mode, + max_new_tokens=max_new_tokens, + stream=stream, + sampling=sampling, + response_format=str(body.get("response_format") or "wav").lower(), + ) + + +def _reject_unsupported_speech_format(body: Dict[str, Any]) -> None: + """Speech synthesis supports WAV and little-endian PCM16.""" + requested = _first_str(body, "response_format", "format") + if requested is None: + return + fmt = str(requested).strip().lower() + if fmt in ("wav", "audio/wav", "pcm"): + return + raise ClientError("only 'wav' or 'pcm' response_format is supported") + + +def _reject_malformed_stream_mode(body: Dict[str, Any]) -> None: + requested = _first_str(body, "response_format", "format") + if _as_bool(body.get("stream")) and str(requested or "").strip().lower() != "pcm": + raise ClientError("stream=true requires response_format=pcm") + + +def _stable_reference_path(holder: ModelHolder, request: SpeechRequest) -> str: + """Stage one reference by content hash so faster's prompt cache can be reused.""" + if request.ref_audio is None: + raise ClientError("reference audio is required", status_code=400) + waveform, sample_rate = request.ref_audio + payload = _to_wav_bytes(waveform, int(sample_rate)) + digest = hashlib.sha256(payload).hexdigest() + path = os.path.join(holder._reference_tmp.name, f"{digest}.wav") + with holder._reference_lock: + if path not in holder._reference_paths or not os.path.exists(path): + if len(holder._reference_paths) >= 16: + for old in holder._reference_paths: + _safe_unlink(old) + holder._reference_paths.clear() + cache = getattr(holder.model, "_voice_prompt_cache", None) + if isinstance(cache, dict): + cache.clear() + with open(path, "wb") as handle: + handle.write(payload) + holder._reference_paths.add(path) + return path + + +def _run_faster_tts_streaming(holder: ModelHolder, request: SpeechRequest) -> Response: + """Start FasterQwen3TTS clone streaming and return a bounded PCM consumer.""" + model = holder.model + if model is None: + raise ClientError("model is not loaded", status_code=503) + reference_path = _stable_reference_path(holder, request) + + def _produce(put, stopped) -> None: + generator = None + try: + cache = getattr(model, "_voice_prompt_cache", None) + if isinstance(cache, dict) and len(cache) >= 16: + cache.clear() + generator = model.generate_voice_clone_streaming( + text=request.text, + language=request.language, + ref_audio=reference_path, + ref_text=request.ref_text or "", + max_new_tokens=request.max_new_tokens, + **request.sampling, + xvec_only=request.x_vector_only_mode, + instruct=request.instruct, + chunk_size=4, + ) + for audio_chunk, _sample_rate, _timing in generator: + if stopped.is_set(): + break + if int(_sample_rate) != PCM_SAMPLE_RATE: + raise RuntimeError("unexpected model sample rate") + pcm = _to_pcm16_le(audio_chunk) + if not pcm: + continue + # PCM_SAMPLE_RATE is the adapter contract; the bounded queue + # applies backpressure before more GPU work is converted. + if not put(pcm): + break + finally: + close = getattr(generator, "close", None) + if callable(close): + try: + close() + except Exception: + LOG.error("streaming generator cleanup failed") + + chunks, stopped = holder.start_stream_job(_produce) + + async def _consume() -> AsyncIterator[bytes]: + try: + while True: + try: + item = chunks.get_nowait() + except queue.Empty: + await asyncio.sleep(0.01) + continue + if item is None: + return + if isinstance(item, Exception): + raise item + yield item + finally: + stopped.set() + + return StreamingResponse( + _consume(), + media_type="audio/pcm; rate=24000; channels=1", + headers=_pcm_response_headers(PCM_SAMPLE_RATE), + ) + + +async def _run_tts(holder: ModelHolder, request: SpeechRequest) -> Response: + model = holder.model + if model is None: + raise ClientError("model is not loaded", status_code=503) + + mode = holder.mode + if mode == "voice_clone" and holder.uses_faster_engine and request.stream: + return _run_faster_tts_streaming(holder, request) + if mode == "custom_voice": + wavs, sample_rate = await holder.run_in_thread( + model.generate_custom_voice, + text=request.text, + language=request.language, + speaker=request.speaker, + instruct=request.instruct, + max_new_tokens=request.max_new_tokens, + **request.sampling, + ) + elif mode == "voice_design": + wavs, sample_rate = await holder.run_in_thread( + model.generate_voice_design, + text=request.text, + language=request.language, + instruct=request.instruct, + max_new_tokens=request.max_new_tokens, + **request.sampling, + ) + elif holder.uses_faster_engine: + cache = getattr(model, "_voice_prompt_cache", None) + if isinstance(cache, dict) and len(cache) >= 16: + cache.clear() + reference_path = _stable_reference_path(holder, request) + wavs, sample_rate = await holder.run_in_thread( + model.generate_voice_clone, + text=request.text, language=request.language, + ref_audio=reference_path, ref_text=request.ref_text or "", + xvec_only=request.x_vector_only_mode, + max_new_tokens=request.max_new_tokens, + **request.sampling, + ) + else: + wavs, sample_rate = await holder.run_in_thread( + model.generate_voice_clone, + text=request.text, + language=request.language, + ref_audio=request.ref_audio, + ref_text=request.ref_text, + x_vector_only_mode=request.x_vector_only_mode, + max_new_tokens=request.max_new_tokens, + **request.sampling, + ) + + waveform = wavs[0] if isinstance(wavs, (list, tuple)) and len(wavs) else wavs + if waveform is None: + raise ClientError("model returned no audio", status_code=500) + if request.response_format == "pcm": + return Response(content=_to_pcm16_le(waveform), media_type="audio/pcm", + headers=_pcm_response_headers(int(sample_rate))) + payload = _to_wav_bytes(waveform, int(sample_rate or 24000)) + return Response( + content=payload, + media_type="audio/wav", + headers={"Content-Disposition": 'attachment; filename="speech.wav"'}, + ) + + +async def _run_stt( + holder: ModelHolder, + raw: bytes, + suffix: str, + language: Optional[str], + prompt: Optional[str], +) -> Any: + """Decode + transcribe on the worker thread, cleaning up the temp file last. + + ``whisper.load_audio`` decodes any ffmpeg-supported container to a 16 kHz + mono numpy array; a bad decode surfaces as a client error. The ``finally`` + runs on the worker thread, so cleanup can only happen after the worker + completes -- even if the awaiting caller was cancelled or disconnected. + """ + model = holder.model + if model is None: + raise ClientError("model is not loaded", status_code=503) + if suffix not in SUPPORTED_AUDIO_SUFFIXES: + # Reject a request that would need an opaque temp path. + raise ClientError("uploaded audio must have a recognised file extension") + + def _job() -> Any: + import whisper + + tmp_path: Optional[str] = None + try: + with tempfile.NamedTemporaryFile(suffix=suffix, delete=False) as handle: + handle.write(raw) + tmp_path = handle.name + try: + audio = whisper.load_audio(tmp_path) + except ClientError: + raise + except Exception as exc: + raise ClientError("uploaded audio could not be decoded") from exc + array = np.asarray(audio, dtype=np.float32).reshape(-1) + if array.size == 0: + raise ClientError("uploaded audio is empty") + sample_rate = getattr(getattr(whisper, "audio", None), "SAMPLE_RATE", 16000) + duration = array.size / float(sample_rate) + if duration > MAX_STT_SECONDS: + raise ClientError("uploaded audio is too long", status_code=413) + kwargs: Dict[str, Any] = {} + if language: + kwargs["language"] = language + if prompt: + kwargs["initial_prompt"] = prompt + return model.transcribe(array, **kwargs) + finally: + if tmp_path: + _safe_unlink(tmp_path) + + return await holder.run_in_thread(_job) + + +def _format_transcription(result: Any, fmt: str) -> Response: + if isinstance(result, dict): + text = str(result.get("text") or "").strip() + segments = result.get("segments") + language = result.get("language") + else: + text = str(getattr(result, "text", "") or "").strip() + segments = getattr(result, "segments", None) + language = getattr(result, "language", None) + if fmt == "text": + return Response(content=text, media_type="text/plain") + payload: Dict[str, Any] = {"text": text} + if fmt == "verbose_json": + if segments is not None: + payload["segments"] = segments + if language is not None: + payload["language"] = language + return JSONResponse(payload) + + +def _error_response(exc: ClientError) -> JSONResponse: + error_type = "rate_limit_error" if exc.status_code == 429 else "invalid_request_error" + return JSONResponse( + {"error": {"message": exc.message, "type": error_type}}, + status_code=exc.status_code, + ) + + +def _safe_unlink(path: str) -> None: + try: + os.unlink(path) + except OSError: + pass + + +# --------------------------------------------------------------------------- +# Application factory +# --------------------------------------------------------------------------- + + +def create_app(holder: ModelHolder) -> FastAPI: + app = FastAPI(title="LLooM Audio CUDA Adapter", version="1.2.0") + app.state.holder = holder + + @app.get("/health") + async def health() -> JSONResponse: + state = "ready" if holder.ready else ("fatal" if holder.fatal else "loading") + payload = { + "ok": holder.ready, + "status": state, + "loaded": holder.ready, + "kind": holder.kind, + "mode": holder.mode, + "model": holder.model_id, + "busy": holder.busy, + } + status_code = 200 if holder.ready else 503 + return JSONResponse(payload, status_code=status_code) + + @app.get("/v1/models") + async def list_models() -> JSONResponse: + data = [{"id": holder.model_id, "object": "model", "owned_by": "lloom-local"}] + return JSONResponse({"object": "list", "data": data}) + + @app.post("/v1/audio/speech") + async def speech(request: Request) -> Response: + if holder.fatal: + return JSONResponse( + {"error": {"message": "model failed to load", "type": "server_error"}}, + status_code=503, + ) + if holder.kind != "tts" or not holder.ready: + return JSONResponse( + {"error": {"message": "speech is unavailable", "type": "invalid_request_error"}}, + status_code=400, + ) + + try: + # Cap the whole body *before* parsing (JSON or multipart). + await _read_bounded_body(request) + content_type = (request.headers.get("content-type") or "").lower() + if "multipart/form-data" in content_type: + body = await _parse_speech_multipart(request) + else: + body = await _read_json_body(request) + speech_request = _build_speech_request(holder, body) + _reject_unsupported_speech_format(body) + _reject_malformed_stream_mode(body) + except ClientError as exc: + return _error_response(exc) + + try: + lease = holder.ensure_idle() + except BusyError as exc: + return _error_response(ClientError(str(exc), status_code=429)) + + try: + return await _run_tts(holder, speech_request) + except ClientError as exc: + holder.clear_busy(lease) + return _error_response(exc) + except Exception as exc: # never leak internals + holder.clear_busy(lease) + LOG.error("speech failed (%s)", type(exc).__name__) + return JSONResponse( + {"error": {"message": "internal error", "type": "server_error"}}, + status_code=500, + ) + + @app.post("/v1/audio/transcriptions") + async def transcriptions(request: Request) -> Response: + if holder.fatal: + return JSONResponse( + {"error": {"message": "model failed to load", "type": "server_error"}}, + status_code=503, + ) + if holder.kind != "stt" or not holder.ready: + return JSONResponse( + { + "error": { + "message": "transcriptions are unavailable", + "type": "invalid_request_error", + } + }, + status_code=400, + ) + + try: + # Cap the whole multipart body before the form parser runs. + await _read_bounded_body(request) + form = await request.form(max_files=1, max_fields=16) + upload = form.get("file") + model = form.get("model") + language = form.get("language") + prompt = form.get("prompt") + response_format = form.get("response_format") or "json" + + if not hasattr(upload, "read"): + raise ClientError("form field 'file' must be an uploaded file") + if model is not None: + holder.resolve_model_id(str(model)) + fmt = str(response_format).strip().lower() + if fmt not in ("json", "text", "verbose_json"): + raise ClientError("unsupported response_format") + raw = await upload.read(MAX_UPLOAD_BYTES + 1) + if not raw: + raise ClientError("uploaded audio is empty") + if len(raw) > MAX_UPLOAD_BYTES: + raise ClientError("uploaded audio is too large", status_code=413) + + suffix = os.path.splitext(upload.filename or "")[1].lower() + if suffix not in SUPPORTED_AUDIO_SUFFIXES: + raise ClientError("unsupported audio file extension") + language = str(language).strip() if language else None + if language and language.lower() == "auto": + language = None + + holder.ensure_idle() + result = await _run_stt(holder, raw, suffix, language, prompt) + except BusyError as exc: + return _error_response(ClientError(str(exc), status_code=429)) + except ClientError as exc: + return _error_response(exc) + except Exception as exc: # never leak internals + LOG.exception("transcription failed: %s", exc) + return JSONResponse( + {"error": {"message": "internal error", "type": "server_error"}}, + status_code=500, + ) + return _format_transcription(result, fmt) + + return app + + +async def _parse_speech_multipart(request) -> Dict[str, Any]: + form = await request.form(max_files=1, max_fields=16) + body: Dict[str, Any] = {} + for key, value in form.multi_items(): + if hasattr(value, "read"): + raw = await value.read(MAX_UPLOAD_BYTES + 1) + if len(raw) > MAX_UPLOAD_BYTES: + raise ClientError("uploaded reference audio is too large", status_code=413) + body[key] = raw + else: + body[key] = value + return body + + +def main(argv: Optional[Sequence[str]] = None) -> int: + parser = argparse.ArgumentParser(description="LLooM local CUDA audio adapter") + parser.add_argument("--kind", required=True, choices=("tts", "stt"), help="model kind") + parser.add_argument("--model", required=True, help="fixed model id served by this process") + parser.add_argument( + "--model-path", + default=None, + help="local model directory (TTS) or local Whisper checkpoint (STT)", + ) + parser.add_argument("--host", default="127.0.0.1") + parser.add_argument("--port", type=int, default=8220) + parser.add_argument( + "--hf-home", default=None, help="cache root override (kept for compatibility)" + ) + args = parser.parse_args(argv) + + logging.basicConfig(level=logging.INFO) + + config = ServerConfig( + kind=args.kind, + model_id=args.model, + model_path=args.model_path, + host=args.host, + port=args.port, + hf_home=args.hf_home, + ) + holder = ModelHolder(config) + + try: + holder.load() + except Exception: + # Fatal: refuse to serve a half-initialised process. + return 1 + + app = create_app(holder) + + import uvicorn + + uvicorn.run(app, host=config.host, port=config.port, log_level="info", workers=1) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/backends/spark-audio/test_lloom_audio_cuda_server.py b/backends/spark-audio/test_lloom_audio_cuda_server.py new file mode 100644 index 0000000..18230ff --- /dev/null +++ b/backends/spark-audio/test_lloom_audio_cuda_server.py @@ -0,0 +1,1018 @@ +"""Focused tests for lloom_audio_cuda_server. + +Coverage targets the previously-missed failure modes: + + * ``/health`` stays responsive (ASGI-level) while a slow inference runs; + * a second concurrent generation is rejected with HTTP 429, not queued; + * temp-file cleanup happens only after the worker completes, even on cancel; + * ``max_new_tokens`` reaches every Qwen ``generate_*`` method; + * STT loads the *local* Whisper checkpoint (``--model-path``), no download; + * reference duration is enforced from decoded frames for compressed audio; + * the total body cap applies to JSON and multipart before parsing; + * unsupported speech ``response_format`` is rejected. + +Stubs stand in for torch/soundfile/whisper so no GPU or network is used. +""" +from __future__ import annotations + +import asyncio +import base64 +import io +import os +import sys +import struct +import threading +import time +import wave + +import anyio +import httpx +import numpy as np +import pytest + +from fastapi.testclient import TestClient + +import lloom_audio_cuda_server as mod + + +class FakeTTSModel: + """Records calls and tracks concurrency to prove serialization.""" + + def __init__(self) -> None: + self.calls = [] + self.active = 0 + self.max_active = 0 + self.delay = 0.0 + + def _record(self, name, kwargs): + self.active += 1 + self.max_active = max(self.max_active, self.active) + try: + if self.delay: + time.sleep(self.delay) + self.calls.append((name, kwargs)) + return [np.zeros(800, dtype=np.float32)], 24000 + finally: + self.active -= 1 + + def generate_custom_voice(self, **kwargs): + return self._record("custom", kwargs) + + def generate_voice_design(self, **kwargs): + return self._record("design", kwargs) + + def generate_voice_clone(self, **kwargs): + return self._record("clone", kwargs) + + +class FakeWhisper: + def __init__(self) -> None: + self.transcribe_calls = [] + + def transcribe(self, path, **kwargs): + self.transcribe_calls.append((path, kwargs)) + return {"text": "hello world", "language": "en", "segments": [{"id": 0}]} + + +class FakeStreamingGenerator: + def __init__(self, model, chunk_count=2): + self.model = model + self.chunk_count = chunk_count + self.closed = False + + def __iter__(self): + return self + + def __next__(self): + if self.model.emitted >= self.chunk_count: + raise StopIteration + self.model.emitted += 1 + self.model.emitted_peak = max(self.model.emitted_peak, self.model.emitted) + # First frame proves the wire starts before synthesis is complete. + if self.model.emitted == 1: + return np.zeros(4, dtype=np.float32), 24000, {"frame": 1} + assert self.model.release.wait(timeout=2), "streaming fixture timed out" + if self.model.stop.is_set(): + raise StopIteration + return np.full(4, self.model.emitted, dtype=np.float32), 24000, {"frame": self.model.emitted} + + def close(self): + self.closed = True + + +class FakeStreamingTTS: + def __init__(self, chunk_count=16): + self.calls = [] + self.emitted = 0 + self.emitted_peak = 0 + self.chunk_count = chunk_count + self.release = threading.Event() + self.stop = threading.Event() + + def generate_voice_clone_streaming(self, **kwargs): + self.calls.append(kwargs) + self.emitted = 0 + self.emitted_peak = 0 + return FakeStreamingGenerator(self, self.chunk_count) + + +def _holder(kind, model_id, model, model_path="/tmp/model"): + config = mod.ServerConfig(kind=kind, model_id=model_id, model_path=model_path) + holder = mod.ModelHolder(config) + holder.model = model + holder.ready = True + return holder + + +def _speech_request(holder, **overrides): + base = dict( + model_id=holder.model_id, + text="hi", + language=mod.DEFAULT_LANGUAGE, + instruct="warm", + speaker="serena", + ref_audio=None, + ref_text=None, + x_vector_only_mode=False, + max_new_tokens=64, + stream=False, + ) + base.update(overrides) + return mod.SpeechRequest(**base) + + +# --------------------------------------------------------------------------- +# Mode inference / fixed id / config +# --------------------------------------------------------------------------- + + +def test_mode_inference(): + assert mod._infer_tts_mode("Qwen3-TTS-12Hz-1.7B-CustomVoice") == "custom_voice" + assert mod._infer_tts_mode("Qwen3-TTS-12Hz-1.7B-VoiceDesign") == "voice_design" + assert mod._infer_tts_mode("Qwen3-TTS-12Hz-0.6B-Base") == "voice_clone" + + +def test_fixed_model_id_rejects_other_ids(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + assert holder.resolve_model_id("Qwen3-TTS-12Hz-0.6B-CustomVoice") == holder.model_id + assert holder.resolve_model_id(None) == holder.model_id + assert holder.resolve_model_id("") == holder.model_id + with pytest.raises(mod.ClientError) as err: + holder.resolve_model_id("gpt-4o-audio-preview") + assert err.value.status_code == 404 + + +def test_stt_gateway_id_alias_accepted_and_others_rejected(): + holder = _holder( + "stt", "openai/whisper-large-v3-turbo", FakeWhisper(), model_path="/tmp/whisper" + ) + assert holder.resolve_model_id("openai/whisper-large-v3-turbo") == holder.model_id + assert holder.resolve_model_id("whisper-1") == holder.model_id + with pytest.raises(mod.ClientError): + holder.resolve_model_id("gpt-4o-transcribe") + + +def test_config_rejects_bad_kind_and_empty_id(): + with pytest.raises(ValueError): + mod.ServerConfig(kind="video", model_id="x") + with pytest.raises(ValueError): + mod.ServerConfig(kind="tts", model_id=" ") + + +# --------------------------------------------------------------------------- +# Speech validation +# --------------------------------------------------------------------------- + + +def test_custom_voice_instructions_optional_but_speaker_validated(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + # Instructions are optional: a bare preset-speaker request is valid. + req = mod._build_speech_request(holder, {"input": "hi"}) + assert req.speaker == "serena" + assert req.instruct is None + with pytest.raises(mod.ClientError) as err: + mod._build_speech_request(holder, {"input": "hi", "speaker": "ghost"}) + assert err.value.status_code == 404 + + +def test_voice_design_requires_instruct(): + holder = _holder("tts", "Qwen3-TTS-12Hz-1.7B-VoiceDesign", FakeTTSModel()) + with pytest.raises(mod.ClientError): + mod._build_speech_request(holder, {"input": "hi"}) + req = mod._build_speech_request(holder, {"input": "hi", "instruct": "deep narrator"}) + assert req.instruct == "deep narrator" + + +def test_language_defaults_to_auto(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + req = mod._build_speech_request(holder, {"input": "hi"}) + assert req.language == "Auto" + blank = mod._build_speech_request(holder, {"input": "hi", "language": " "}) + assert blank.language == "Auto" + explicit = mod._build_speech_request(holder, {"input": "hi", "language": "en"}) + assert explicit.language == "en" + + +def test_base_clone_requires_ref_audio_and_ref_text(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-Base", FakeTTSModel()) + with pytest.raises(mod.ClientError): + mod._build_speech_request(holder, {"input": "hi"}) + + wav = mod._to_wav_bytes(np.zeros(24000, dtype=np.float32), 24000) + with pytest.raises(mod.ClientError): + mod._build_speech_request(holder, {"input": "hi", "ref_audio": wav}) + + req = mod._build_speech_request( + holder, {"input": "hi", "ref_audio": wav, "ref_text": "hello"} + ) + assert req.ref_audio is not None + assert req.ref_text == "hello" + + +def test_base_clone_rejects_caller_paths_and_urls(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-Base", FakeTTSModel()) + for value in ("/etc/passwd", "file:///etc/passwd", "https://example.com/a.wav"): + with pytest.raises(mod.ClientError): + mod._build_speech_request( + holder, {"input": "hi", "ref_audio": value, "ref_text": "x"} + ) + + +def test_base_clone_accepts_data_url_base64(monkeypatch): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-Base", FakeTTSModel()) + wav = mod._to_wav_bytes(np.zeros(24000, dtype=np.float32), 24000) + monkeypatch.setattr(mod, "_decode_ref_audio", lambda raw: (np.zeros(4, np.float32), 16000)) + b64 = base64.b64encode(wav).decode() + req = mod._build_speech_request( + holder, + {"input": "hi", "ref_audio": f"data:audio/wav;base64,{b64}", "ref_text": "x"}, + ) + assert req.ref_audio is not None + + +def test_x_vector_only_mode_allows_missing_ref_text(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-Base", FakeTTSModel()) + wav = mod._to_wav_bytes(np.zeros(24000, dtype=np.float32), 24000) + req = mod._build_speech_request( + holder, {"input": "hi", "ref_audio": wav, "x_vector_only_mode": "true"} + ) + assert req.x_vector_only_mode is True + + +def test_oversized_text_is_rejected_and_tokens_capped(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + with pytest.raises(mod.ClientError) as err: + mod._build_speech_request( + holder, {"input": "x" * (mod.MAX_TEXT_CHARS + 1), "instructions": "w"} + ) + assert err.value.status_code == 413 + + req = mod._build_speech_request( + holder, {"input": "hi", "instructions": "w", "max_new_tokens": 10_000_000} + ) + assert req.max_new_tokens == mod.MAX_NEW_TOKENS + + +def test_oversized_ref_audio_rejected(): + big = b"\x00" * (mod.MAX_UPLOAD_BYTES + 1) + with pytest.raises(mod.ClientError) as err: + mod._decode_ref_audio(big) + assert err.value.status_code == 413 + + +def test_parameter_aliases(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + req = mod._build_speech_request( + holder, {"text": "hi", "instruct": "calm", "voice": "Ryan"} + ) + assert req.text == "hi" + assert req.instruct == "calm" + assert req.speaker == "ryan" + + +# --------------------------------------------------------------------------- +# Token propagation (no call may silently drop max_new_tokens) +# --------------------------------------------------------------------------- + + +def test_max_new_tokens_reaches_every_generate_method(): + cases = [ + ("custom_voice", "Qwen3-TTS-12Hz-0.6B-CustomVoice", "generate_custom_voice"), + ("voice_design", "Qwen3-TTS-12Hz-1.7B-VoiceDesign", "generate_voice_design"), + ("voice_clone", "Qwen3-TTS-12Hz-0.6B-Base", "generate_voice_clone"), + ] + for _mode, model_id, expected in cases: + model = FakeTTSModel() + holder = _holder("tts", model_id, model) + body = {"input": "hi", "max_new_tokens": 123} + if expected == "generate_voice_design": + body["instructions"] = "deep" + if expected == "generate_voice_clone": + body["ref_audio"] = mod._to_wav_bytes(np.zeros(24000, np.float32), 24000) + body["ref_text"] = "ref" + req = mod._build_speech_request(holder, body) + assert req.max_new_tokens == 123 + # Decoding needs soundfile; only the token kwarg is under test here. + try: + anyio.run(mod._run_tts, holder, req) + except mod.ClientError: + pass + assert model.calls, f"{expected} was not called" + name, kwargs = model.calls[-1] + assert kwargs["max_new_tokens"] == 123 + + +def test_custom_voice_dispatch_uses_speaker_and_optional_instruct(): + model = FakeTTSModel() + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", model) + req = mod._build_speech_request(holder, {"input": "hi"}) + response = anyio.run(mod._run_tts, holder, req) + assert model.calls[0][0] == "custom" + assert model.calls[0][1]["speaker"] == "serena" + assert model.calls[0][1]["instruct"] is None + assert model.calls[0][1]["language"] == "Auto" + assert response.media_type == "audio/wav" + assert response.body[:4] == b"RIFF" + + +def test_voice_clone_dispatches_ref_audio_tuple(): + model = FakeTTSModel() + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-Base", model) + wav = mod._to_wav_bytes(np.zeros(24000, dtype=np.float32), 24000) + try: + req = mod._build_speech_request( + holder, {"input": "hi", "ref_audio": wav, "ref_text": "hello"} + ) + except mod.ClientError: + pytest.skip("soundfile not available to decode the reference clip") + anyio.run(mod._run_tts, holder, req) + # A failed clone call must fail the test, not be swallowed. + assert model.calls, "generate_voice_clone was never invoked" + ref = model.calls[0][1]["ref_audio"] + assert isinstance(ref, tuple) and len(ref) == 2 + + +# --------------------------------------------------------------------------- +# Concurrency: serialization, 429, health responsiveness, cleanup on cancel +# --------------------------------------------------------------------------- + + +def test_gpu_calls_are_serialized(): + model = FakeTTSModel() + model.delay = 0.1 + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", model) + reqs = [_speech_request(holder) for _ in range(5)] + + async def main(): + async with anyio.create_task_group() as tg: + for r in reqs: + tg.start_soon(mod._run_tts, holder, r) + + anyio.run(main) + assert model.max_active == 1, "GPU inference must be serialized" + assert len(model.calls) == 5 + + +def _client(holder): + return TestClient(mod.create_app(holder)) + + +def test_health_stays_responsive_during_slow_inference(): + """The ASGI event loop must answer /health while a slow generation runs.""" + model = FakeTTSModel() + model.delay = 0.4 + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", model) + client = _client(holder) + + result = {} + + def slow_call(): + result["resp"] = client.post( + "/v1/audio/speech", + json={"input": "hi", "model": holder.model_id}, + ) + + worker = threading.Thread(target=slow_call) + worker.start() + time.sleep(0.05) # let the request enter the slow inference + + t0 = time.time() + health = client.get("/health") + elapsed = time.time() - t0 + assert health.status_code == 200 + assert health.json()["busy"] is True + assert elapsed < 0.3, f"/health was blocked for {elapsed:.2f}s by inference" + + worker.join(timeout=5) + assert result["resp"].status_code == 200 + assert model.max_active == 1 + + +def test_second_generation_is_rejected_with_429(): + model = FakeTTSModel() + model.delay = 0.4 + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", model) + client = _client(holder) + + results = {} + + def slow_call(): + results["resp"] = client.post( + "/v1/audio/speech", json={"input": "hi", "model": holder.model_id} + ) + + worker = threading.Thread(target=slow_call) + worker.start() + time.sleep(0.05) + + second = client.post("/v1/audio/speech", json={"input": "hi", "model": holder.model_id}) + assert second.status_code == 429 + assert second.json()["error"]["type"] == "rate_limit_error" + + worker.join(timeout=5) + assert results["resp"].status_code == 200 + # The busy flag must be free again once the first generation finished. + assert holder.busy is False + assert client.post( + "/v1/audio/speech", json={"input": "hi", "model": holder.model_id} + ).status_code == 200 + + +def test_busy_is_held_until_worker_completes_on_cancel(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + holder.ensure_idle() + finished = threading.Event() + def slow_inference(): + time.sleep(0.1) + assert holder.busy + finished.set() + async def scenario(): + with anyio.move_on_after(0.03) as scope: + await holder.run_in_thread(slow_inference) + assert scope.cancel_called + assert finished.is_set() + assert not holder.busy + anyio.run(scenario) + + +def test_stt_cleanup_only_after_worker_completes(monkeypatch): + model = FakeWhisper() + holder = _holder("stt", "turbo", model, model_path="/tmp/whisper") + holder.ensure_idle() + seen_paths = [] + def decode(path): + seen_paths.append(path) + assert os.path.exists(path) + return np.zeros(16000, dtype=np.float32) + def slow_transcribe(audio, **kwargs): + assert isinstance(audio, np.ndarray) + time.sleep(0.1) + assert holder.busy + assert os.path.exists(seen_paths[0]) + return {"text":"done"} + monkeypatch.setattr(model,"transcribe",slow_transcribe) + import sys + monkeypatch.setitem(sys.modules,"whisper",type("W",(),{"load_audio":staticmethod(decode),"audio":type("A",(),{"SAMPLE_RATE":16000})})) + async def scenario(): + with anyio.move_on_after(0.03) as scope: + await mod._run_stt(holder,b"RIFFdata",".wav",None,None) + assert scope.cancel_called + anyio.run(scenario) + assert seen_paths + assert not os.path.exists(seen_paths[0]) + assert not holder.busy + + +# --------------------------------------------------------------------------- +# STT: model path, formats, limits +# --------------------------------------------------------------------------- + + +def test_stt_loads_local_checkpoint_path_not_alias(monkeypatch, tmp_path): + checkpoint = tmp_path / "whisper-large-v3-turbo" + checkpoint.write_bytes(b"synthetic checkpoint") + + recorded = {} + + def fake_load_model(name, device=None, **kwargs): + recorded["name"] = name + recorded["device"] = device + return FakeWhisper() + + fake_whisper = type("W", (), {"load_model": staticmethod(fake_load_model)}) + import sys + + monkeypatch.setitem(sys.modules, "whisper", fake_whisper) + + holder = mod.ModelHolder( + mod.ServerConfig(kind="stt", model_id="turbo", model_path=str(checkpoint)) + ) + holder.load() + assert recorded["name"] == str(checkpoint) + assert recorded["device"] == "cuda" + assert holder.ready is True + + +def test_stt_requires_existing_local_checkpoint(): + holder = mod.ModelHolder( + mod.ServerConfig(kind="stt", model_id="turbo", model_path="/nope/missing") + ) + with pytest.raises(Exception): + holder.load() + assert holder.fatal is True + + +def test_transcription_passes_language_and_prompt(): + model = FakeWhisper() + holder = _holder("stt", "turbo", model, model_path="/tmp/whisper") + fake_whisper = type( + "W", + (), + { + "load_audio": staticmethod(lambda p: np.zeros(16000, dtype=np.float32)), + "audio": type("A", (), {"SAMPLE_RATE": 16000}), + }, + ) + import sys + + sys.modules["whisper"] = fake_whisper + try: + anyio.run(mod._run_stt, holder, b"x", ".wav", "es", "context") + finally: + del sys.modules["whisper"] + _path, kwargs = model.transcribe_calls[0] + assert kwargs == {"language": "es", "initial_prompt": "context"} + + +def test_transcription_verbose_json_and_text(): + result = {"text": "hello world", "language": "en", "segments": [{"id": 0}]} + verbose = mod._format_transcription(result, "verbose_json") + assert b"segments" in verbose.body + text = mod._format_transcription(result, "text") + assert text.body == b"hello world" + plain = mod._format_transcription(result, "json") + assert b"hello world" in plain.body + + +def test_transcription_transcribes_numpy_array_and_caps_duration(monkeypatch): + """Long decoded audio is rejected before transcribe; arrays are forwarded.""" + model = FakeWhisper() + holder = _holder("stt", "turbo", model, model_path="/tmp/whisper") + sample_rate = 16000 + too_long = np.zeros(sample_rate * int(mod.MAX_STT_SECONDS) + sample_rate, dtype=np.float32) + fake_whisper = type( + "W", + (), + { + "load_audio": staticmethod(lambda p: too_long), + "audio": type("A", (), {"SAMPLE_RATE": sample_rate}), + }, + ) + import sys + + monkeypatch.setitem(sys.modules, "whisper", fake_whisper) + with pytest.raises(mod.ClientError) as err: + anyio.run(mod._run_stt, holder, b"x", ".mp3", None, None) + assert err.value.status_code == 413 + assert model.transcribe_calls == [] + + +def test_transcription_rejects_unknown_extension_and_upload_size(monkeypatch): + holder = _holder("stt", "turbo", FakeWhisper(), model_path="/tmp/whisper") + fake_whisper = type( + "W", + (), + { + "load_audio": staticmethod(lambda p: np.zeros(16, dtype=np.float32)), + "audio": type("A", (), {"SAMPLE_RATE": 16000}), + }, + ) + import sys + + monkeypatch.setitem(sys.modules, "whisper", fake_whisper) + with pytest.raises(mod.ClientError): + anyio.run(mod._run_stt, holder, b"x", ".xyz", None, None) + + client = _client(holder) + oversized = b"\x00" * (mod.MAX_UPLOAD_BYTES + 1) + resp = client.post( + "/v1/audio/transcriptions", + files={"file": ("a.wav", oversized, "audio/wav")}, + ) + assert resp.status_code == 413 + + +def test_safe_unlink_ignores_missing(tmp_path): + target = tmp_path / "gone.wav" + mod._safe_unlink(str(target)) + target.write_bytes(b"x") + mod._safe_unlink(str(target)) + assert not target.exists() + + +# --------------------------------------------------------------------------- +# Compressed reference duration (decoded frames, not a WAV header) +# --------------------------------------------------------------------------- + + +def _flac_bytes(seconds: float, sample_rate: int = 16000) -> bytes: + import soundfile as sf + + data = np.zeros(int(seconds * sample_rate), dtype=np.float32) + buffer = io.BytesIO() + sf.write(buffer, data, sample_rate, format="FLAC") + return buffer.getvalue() + + +def test_compressed_reference_duration_uses_decoded_frames(): + raw = _flac_bytes(0.25) + # A WAV-header-only probe cannot see a duration for a compressed container. + assert mod._decode_wav_duration(raw) is None + array, rate = mod._decode_ref_audio(raw) + assert rate == 16000 + assert abs(mod._decoded_duration_seconds(array, rate) - 0.25) < 0.02 + + +def test_compressed_reference_over_limit_is_rejected(): + raw = _flac_bytes(mod.MAX_REF_SECONDS + 2.0) + assert mod._decode_wav_duration(raw) is None + with pytest.raises(mod.ClientError) as err: + mod._decode_ref_audio(raw) + assert err.value.status_code == 413 + + +def test_compressed_reference_too_short_is_rejected(): + raw = _flac_bytes(0.005) + with pytest.raises(mod.ClientError): + mod._decode_ref_audio(raw) + + +def test_decoded_duration_helpers(): + assert mod._decoded_duration_seconds(np.zeros(16000, np.float32), 16000) == 1.0 + with pytest.raises(mod.ClientError): + mod._decoded_duration_seconds(np.zeros(16000, np.float32), 0) + wav = mod._to_wav_bytes(np.zeros(24000, dtype=np.float32), 24000) + dur = mod._decode_wav_duration(wav) + assert dur is not None and abs(dur - 1.0) < 0.05 + + +# --------------------------------------------------------------------------- +# ASGI-level body caps and speech format rejection +# --------------------------------------------------------------------------- + + +def test_json_body_cap_enforced_before_parsing(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + client = _client(holder) + # No Content-Length (chunked) and a body far past the cap. + sent = {"done": False} + oversized = b"x" * (mod.MAX_BODY_BYTES + 1024) + + async def receive(): + if sent["done"]: + return {"type": "http.request", "body": b"", "more_body": False} + sent["done"] = True + return {"type": "http.request", "body": oversized, "more_body": False} + + scope = { + "type": "http", + "asgi": {"version": "3.0"}, + "http_version": "1.1", + "method": "POST", + "scheme": "http", + "path": "/v1/audio/speech", + "raw_path": b"/v1/audio/speech", + "query_string": b"", + "root_path": "", + "headers": [(b"content-type", b"application/json")], + "client": ("127.0.0.1", 1234), + "server": ("testserver", 80), + } + + async def run(): + response = await httpx.ASGITransport(app=client.app).handle_async_request( + httpx.Request("POST", "http://testserver/v1/audio/speech", content=b"") + ) + return response + + resp = anyio.run(run) + assert resp.status_code in (400, 413) + + +def test_declared_content_length_over_cap_is_rejected(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + client = _client(holder) + resp = client.post( + "/v1/audio/speech", + content=b"{}", + headers={ + "content-type": "application/json", + "content-length": str(mod.MAX_BODY_BYTES + 1), + }, + ) + assert resp.status_code == 413 + + +def test_multipart_body_cap_enforced_before_parsing(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-Base", FakeTTSModel()) + client = _client(holder) + payload = b"\x00" * (mod.MAX_BODY_BYTES + 1) + resp = client.post( + "/v1/audio/speech", + files={"ref_audio": ("ref.wav", payload, "audio/wav")}, + data={"input": "hi", "ref_text": "x"}, + ) + assert resp.status_code in (400, 413) + + +def test_speech_rejects_unsupported_response_format(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + client = _client(holder) + for fmt in ("mp3", "opus", "aac", "flac"): + resp = client.post( + "/v1/audio/speech", + json={"input": "hi", "model": holder.model_id, "response_format": fmt}, + ) + assert resp.status_code == 400, fmt + assert "wav" in resp.json()["error"]["message"] + ok = client.post( + "/v1/audio/speech", + json={"input": "hi", "model": holder.model_id, "response_format": "wav"}, + ) + assert ok.status_code == 200 + + +def test_speech_rejects_unknown_model_id(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + client = _client(holder) + resp = client.post( + "/v1/audio/speech", json={"input": "hi", "model": "gpt-4o-audio-preview"} + ) + assert resp.status_code == 404 + + +# --------------------------------------------------------------------------- +# Routes / lifecycle +# --------------------------------------------------------------------------- + + +def test_health_reports_ready_state_and_routes_present(): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-CustomVoice", FakeTTSModel()) + app = mod.create_app(holder) + assert app.state.holder is holder + routes = {r.path for r in app.routes} + assert {"/health", "/v1/models", "/v1/audio/speech", "/v1/audio/transcriptions"} <= routes + client = _client(holder) + assert client.get("/health").status_code == 200 + assert client.get("/v1/models").json()["data"][0]["id"] == holder.model_id + + +def test_load_failure_marks_fatal(monkeypatch): + holder = mod.ModelHolder( + mod.ServerConfig(kind="tts", model_id="Qwen3-TTS-12Hz-0.6B-CustomVoice") + ) + + def boom(): + raise RuntimeError("cuda oom") + + monkeypatch.setattr(holder, "_load_tts", boom) + with pytest.raises(RuntimeError): + holder.load() + assert holder.fatal is True + assert holder.ready is False + assert holder.load_error == "RuntimeError" + + +def test_transcription_rejects_unsupported_format_field(): + holder = _holder("stt", "turbo", FakeWhisper(), model_path="/tmp/whisper") + client = _client(holder) + resp = client.post( + "/v1/audio/transcriptions", + files={"file": ("a.wav", b"RIFFdata", "audio/wav")}, + data={"response_format": "srt"}, + ) + assert resp.status_code == 400 + + +def test_wav_output_is_riff(): + data = mod._to_wav_bytes([0.0, 0.5, -0.5, 1.0], 16000) + assert data[:4] == b"RIFF" + assert data[8:12] == b"WAVE" + with wave.open(io.BytesIO(data), "rb") as handle: + assert handle.getframerate() == 16000 + assert handle.getnframes() == 4 + + +# --------------------------------------------------------------------------- +# Optional FasterQwen3TTS streaming +# --------------------------------------------------------------------------- + + +def _streaming_holder(model): + holder = _holder("tts", "Qwen3-TTS-12Hz-0.6B-Base", model) + holder.uses_faster_engine = True + return holder + + +def _clone_request(holder, **overrides): + wav = mod._to_wav_bytes(np.zeros(2400, dtype=np.float32), 24000) + return mod._build_speech_request( + holder, + { + "input": "hi", + "ref_audio": wav, + "ref_text": "reference", + "stream": "true", + "response_format": "pcm", + **overrides, + }, + ) + + +def test_faster_stream_first_chunk_is_pcm_le_before_completion(): + model = FakeStreamingTTS() + holder = _streaming_holder(model) + request = _clone_request(holder) + + async def scenario(): + holder.ensure_idle() + response = await mod._run_tts(holder, request) + iterator = response.body_iterator + first = await iterator.__anext__() + assert first == b"\x00\x00\x00\x00\x00\x00\x00\x00" + assert model.calls[0]["xvec_only"] is False + model.release.set() + second = await iterator.__anext__() + assert second == b"\xff\x7f" * 4 + # The process-lifetime, content-addressed path is reused across calls. + path = model.calls[0]["ref_audio"] + assert path in holder._reference_paths + assert mod._stable_reference_path(holder, request) == path + + anyio.run(scenario) + for _ in range(100): + if not holder.busy: + break + time.sleep(0.01) + assert not holder.busy + + +def test_faster_stream_close_signals_producer_and_releases_gpu(): + model = FakeStreamingTTS() + holder = _streaming_holder(model) + request = _clone_request(holder) + + async def scenario(): + holder.ensure_idle() + response = await mod._run_tts(holder, request) + iterator = response.body_iterator + await iterator.__anext__() + assert holder.busy + await iterator.aclose() + model.release.set() + + anyio.run(scenario) + for _ in range(200): + if model.calls and model.emitted_peak <= mod.MAX_STREAM_QUEUE_CHUNKS and not holder.busy: + break + time.sleep(0.01) + assert holder._stream_stop.is_set() + assert not holder.busy + + +def test_faster_stream_queue_is_bounded_and_lock_outlives_disconnect(): + model = FakeStreamingTTS(chunk_count=32) + holder = _streaming_holder(model) + request = _clone_request(holder) + + async def scenario(): + holder.ensure_idle() + await mod._run_tts(holder, request) + # Do not consume. The producer must block at the bounded queue, not + # turn the whole utterance into an unbounded in-memory queue. + for _ in range(20): + if model.emitted_peak >= mod.MAX_STREAM_QUEUE_CHUNKS: + break + await asyncio.sleep(0.01) + assert holder.busy + holder._stream_stop.set() + model.release.set() + + anyio.run(scenario) + for _ in range(200): + if not holder.busy: + break + time.sleep(0.01) + assert model.emitted_peak <= mod.MAX_STREAM_QUEUE_CHUNKS + assert not holder.busy + + +def test_malformed_faster_stream_mode_is_rejected_before_gpu(): + model = FakeStreamingTTS() + holder = _streaming_holder(model) + client = _client(holder) + wav = mod._to_wav_bytes(np.zeros(2400, dtype=np.float32), 24000) + resp = client.post( + "/v1/audio/speech", + json={ + "model": holder.model_id, + "input": "hi", + "ref_audio": "data:audio/wav;base64,"+base64.b64encode(wav).decode(), + "ref_text": "reference", + "stream": True, + "response_format": "wav", + }, + ) + assert resp.status_code == 400 + assert resp.json()["error"]["message"] == "stream=true requires response_format=pcm" + assert model.calls == [] + + +def test_faster_engine_load_requires_voice_clone(monkeypatch): + holder = mod.ModelHolder( + mod.ServerConfig( + kind="tts", + model_id="Qwen3-TTS-12Hz-0.6B-CustomVoice", + model_path="/tmp/model", + engine="faster", + ) + ) + fake_torch = type("T", (), {}) + monkeypatch.setitem(sys.modules, "torch", fake_torch) + with pytest.raises(ValueError, match="voice_clone only"): + holder.load() + assert holder.fatal + + +def test_stream_route_acquires_busy_once_and_emits_pcm(): + model = FakeStreamingTTS(chunk_count=2) + model.release.set() + holder = _streaming_holder(model) + wav = mod._to_wav_bytes(np.zeros(2400, dtype=np.float32), 24000) + response = _client(holder).post('/v1/audio/speech', json={ + 'model': holder.model_id, 'input': 'hello', 'ref_text': 'reference', + 'ref_audio': 'data:audio/wav;base64,'+base64.b64encode(wav).decode(), + 'stream': True, 'response_format': 'pcm'}) + assert response.status_code == 200 + assert response.headers['x-audio-format'] == 'pcm_s16le' + assert response.content == b'\x00\x00'*4+b'\xff\x7f'*4 + + +def test_stream_fault_is_not_successful_eof(): + class FailingModel: + def generate_voice_clone_streaming(self, **kw): + yield np.zeros(4), 24000, {} + raise RuntimeError('private model detail') + holder = _streaming_holder(FailingModel()) + async def scenario(): + holder.ensure_idle() + response = await mod._run_tts(holder, _clone_request(holder)) + chunks = response.body_iterator + assert await chunks.__anext__() == b'\x00\x00'*4 + with pytest.raises(RuntimeError, match='speech generation failed'): + await chunks.__anext__() + await chunks.aclose() + anyio.run(scenario) + + +def test_old_lease_cannot_clear_new_request_busy_flag(): + holder = _streaming_holder(FakeStreamingTTS()) + old = holder.ensure_idle() + holder.clear_busy(old) + new = holder.ensure_idle() + holder.clear_busy(old) + assert holder.busy + holder.clear_busy(new) + assert not holder.busy + + +def test_stock_pcm_format_contains_no_wav_header(): + holder = _holder('tts', 'Qwen3-TTS-12Hz-0.6B-CustomVoice', FakeTTSModel()) + response = _client(holder).post('/v1/audio/speech', json={'input':'hello','response_format':'pcm'}) + assert response.status_code == 200 + assert response.headers['x-audio-format'] == 'pcm_s16le' + assert not response.content.startswith(b'RIFF') + + +def test_clone_sampling_preserves_profile_settings(): + model = FakeStreamingTTS(chunk_count=2) + model.release.set() + holder = _streaming_holder(model) + wav = mod._to_wav_bytes(np.zeros(2400, dtype=np.float32), 24000) + settings = {'temperature': 0.85, 'top_p': 0.9, 'top_k': 50, 'repetition_penalty': 1.05} + response = _client(holder).post('/v1/audio/speech', json={ + 'model': holder.model_id, 'input': 'hello', 'ref_text': 'reference', + 'ref_audio': 'data:audio/wav;base64,'+base64.b64encode(wav).decode(), + 'stream': True, 'response_format': 'pcm', **settings}) + assert response.status_code == 200 + for key, value in settings.items(): + assert model.calls[-1][key] == value + + +def test_invalid_profile_sampling_rejected_before_generation(): + holder = _streaming_holder(FakeStreamingTTS()) + for key, value in [('temperature', -1), ('top_p', 2), ('top_k', 1.5)]: + response = _client(holder).post('/v1/audio/speech', json={ + 'model': holder.model_id, 'input': 'hello', key: value}) + assert response.status_code == 400 + assert not holder.model.calls diff --git a/package.json b/package.json index ea65903..167b111 100644 --- a/package.json +++ b/package.json @@ -65,7 +65,7 @@ "check:cluster": "node --check src/cluster.mjs && node --check test/cluster.test.mjs", "check:entity-stagger": "node --check scripts/entity-stagger-benchmark.mjs && node --check test/entity-stagger-benchmark.test.mjs", "check:js": "node --check bin/lloom.mjs && node --check bin/lloom-host.mjs && node --check src/backend-catalog.mjs && node --check src/benchmarks.mjs && node --check src/bootstrap.mjs && node --check src/client-integrations.mjs && node --check src/community-client.mjs && node --check src/community-site.mjs && node --check src/config.mjs && node --check src/dashboard.mjs && node --check src/doctor.mjs && node --check src/host-memory.mjs && node --check src/host-server.mjs && node --check src/init.mjs && node --check src/installer.mjs && node --check src/interchange.mjs && node --check src/machine-profile.mjs && node --check src/managed-environment.mjs && node --check src/model-acquisition.mjs && node --check src/model-intake.mjs && node --check src/model-removal.mjs && node --check src/onboarding.mjs && node --check src/recipe-index.mjs && node --check src/recipe-pack-export.mjs && node --check src/recipe-pack.mjs && node --check src/registry.mjs && node --check src/route-control.mjs && node --check src/recipes.mjs && node --check src/resource-fit.mjs && node --check src/runtime-manager.mjs && node --check src/runtime-policy.mjs && node --check src/server.mjs && node --check src/setup.mjs && node --check src/setup-status.mjs && node --check src/process-control.mjs && node --check src/security.mjs && node --check src/tts-catalog.mjs && node --check src/voice-profiles.mjs && node --check src/protocol/text.mjs && node --check src/protocol/reasoning-normalize.mjs && node --check src/protocol/reasoning-effort.mjs && node --check src/protocol/responses.mjs && node --check src/protocol/anthropic.mjs && node --check src/protocol/sse.mjs && node --check src/protocol/stream-anthropic.mjs && node --check src/protocol/stream-responses.mjs && node --check src/protocol/index.mjs && node --check scripts/apply-spark-route-catalog.mjs && node --check scripts/capture-live-metrics.mjs && node --check scripts/merge-live-metrics.mjs && node --check scripts/run-qualitative-chat-benchmark.mjs && node --check scripts/run-local-decision-benchmark.mjs && node --check scripts/check-community-deploy.mjs && node --check scripts/generate-client-configs.mjs && node --check scripts/check-interchange.mjs && node --check scripts/check-package.mjs && node --check scripts/release-lib.mjs && node --check scripts/build-release.mjs && node --check scripts/deploy-spark.mjs && node --check test/audio-generations.test.mjs && node --check test/community-host.test.mjs && node --check test/dashboard-status.test.mjs && node --check test/ds4fv-recipe.test.mjs && node --check test/dspark-hotfixes.test.mjs && node --check test/glm53-exl3-recipe.test.mjs && node --check test/host-memory.test.mjs && node --check test/metrics-persistence.test.mjs && node --check test/model-acquisition.test.mjs && node --check test/model-failover.test.mjs && node --check test/qwen38-vllm-recipe.test.mjs && node --check test/reasoning-normalize.test.mjs && node --check test/reasoning-effort.test.mjs && node --check test/resource-fit.test.mjs && node --check test/route-control.test.mjs && node --check test/spark-route-catalog.test.mjs && node --check test/runtime-watchdog.test.mjs && node --check test/tts-catalog.test.mjs && node --check test/voice-profiles.test.mjs && node --check test/chatterbox-recipe.test.mjs && node --check test/hear-recipe.test.mjs && node --check test/smoke.mjs && node --check src/runtime-supervisor.mjs && node --check test/runtime-supervisor.test.mjs && node --check test/comfyui-media.test.mjs && node --check src/runtime-residency.mjs && node --check src/dashboard-scene.mjs && node --check src/dashboard-presence.mjs && node --check src/dashboard-presence-client.mjs && node --check src/first-run.mjs && node --check src/first-run-page.mjs && node --check src/browser-setup.mjs && node --check src/runtime-preferences.mjs && node --check src/installation-jobs.mjs && node --check src/model-installation.mjs && node --check src/dashboard-installation.mjs && node --check src/runtime-memory-safety.mjs && node --check src/dashboard-memory.mjs && node --check src/runtime-memory-usage.mjs && node --check src/config-profiles.mjs && node --check test/config-profiles.test.mjs", - "check:python": "python3 -m py_compile backends/dspark-vllm/apply-patch-pack.py backends/dspark-vllm/packs/miaai-dsv4flash-d1b76251-defaults/patches/*.py backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/patches/*.py backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/patches/vision_exp/*.py backends/qwen38-vllm/apply-ple-fp8-patch.py backends/qwen38-sglang/*.py backends/glm53-exl3/*.py backends/mlx-audio/lloom_audio_server.py backends/hear/lloom_hear_server.py backends/chatterbox/lloom_chatterbox_server.py patches/apply_mtplx_longctx_fix.py backends/comfyui-media/bridge/*.py backends/comfyui-media/graphs/*.py backends/comfyui-media/build/*.py backends/comfyui-media/install.py backends/qwen-image-diffusers/*.py backends/hear/test/test_hear_security.py src/darwin-memory-usage.py", + "check:python": "python3 -m py_compile backends/dspark-vllm/apply-patch-pack.py backends/dspark-vllm/packs/miaai-dsv4flash-d1b76251-defaults/patches/*.py backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/patches/*.py backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/patches/vision_exp/*.py backends/qwen38-vllm/apply-ple-fp8-patch.py backends/qwen38-sglang/*.py backends/glm53-exl3/*.py backends/mlx-audio/lloom_audio_server.py backends/hear/lloom_hear_server.py backends/chatterbox/lloom_chatterbox_server.py patches/apply_mtplx_longctx_fix.py backends/comfyui-media/bridge/*.py backends/comfyui-media/graphs/*.py backends/comfyui-media/build/*.py backends/comfyui-media/install.py backends/qwen-image-diffusers/*.py backends/hear/test/test_hear_security.py backends/spark-audio/lloom_audio_cuda_server.py backends/spark-audio/test_lloom_audio_cuda_server.py src/darwin-memory-usage.py", "check:shell": "bash -n backends/dspark-vllm/entrypoint.sh backends/dspark-vllm/runtime/build-hardened-image.sh backends/dspark-vllm/packs/miaai-dsv4flash-d1b76251-defaults/patches/*.sh backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/apply-runtime.sh backends/dspark-vllm/packs/miaai-ds4fv-f5665e8/patches/*.sh backends/qwen38-vllm/entrypoint.sh backends/qwen38-sglang/entrypoint.sh backends/glm53-exl3/entrypoint.sh backends/mlx-audio/install.sh backends/hear/install.sh backends/chatterbox/install.sh clients/examples/lloom-claude clients/examples/lloom-codex clients/examples/lloom-hermes clients/examples/lloom-opencode clients/examples/lloom-zero scripts/remote-install-spark.sh scripts/remote-stage-spark-worker.sh", "check:spark-deploy": "node --check test/spark-deploy.test.mjs && node test/spark-deploy.test.mjs", "generate:clients": "node scripts/generate-client-configs.mjs", @@ -73,7 +73,7 @@ "community:check": "node scripts/check-community-deploy.mjs", "package:check": "node scripts/check-package.mjs", "test:adoption": "node test/resource-fit.test.mjs && node test/model-acquisition.test.mjs && node test/dashboard-status.test.mjs", - "test:unit": "node test/runtime-health-ownership.test.mjs && node test/long-prefill-transport.test.mjs && node test/sparkglm-managed.test.mjs && node test/protocol.test.mjs && node test/protocol-stream.test.mjs && node test/stream-metrics.test.mjs && node test/responses-codex.test.mjs && node test/security.test.mjs && node test/community-host.test.mjs && node test/openrouter-provider.test.mjs && node test/ds4fv-recipe.test.mjs && node test/dspark-hotfixes.test.mjs && node test/glm53-exl3-recipe.test.mjs && node test/host-memory.test.mjs && node test/config-profiles.test.mjs && node test/route-control.test.mjs && node test/runtime-policy.test.mjs && node --test test/rate-limit.test.mjs test/runtime-queue.test.mjs test/federated-priority.test.mjs test/runtime-supervisor.test.mjs && node test/interchange.test.mjs && node test/server-resilience.test.mjs && node test/metrics-persistence.test.mjs && node test/model-failover.test.mjs && node test/qwen38-vllm-recipe.test.mjs && node test/reasoning-normalize.test.mjs && node test/reasoning-effort.test.mjs && node test/route-control.test.mjs && node test/spark-route-catalog.test.mjs && node test/chat-lane-canary.test.mjs && node test/runtime-watchdog.test.mjs && node test/tts-catalog.test.mjs && node test/voice-profiles.test.mjs && node test/chatterbox-recipe.test.mjs && node test/hear-recipe.test.mjs && npm run test:adoption && node --test test/performance-sampler.test.mjs && node --test test/audio-generations.test.mjs && node test/video-providers.test.mjs && node --test test/web-functions.test.mjs && node --test test/config-mutation.test.mjs && node --test test/model-maintenance.test.mjs && node --test test/model-maintenance-control.test.mjs && node --test test/route-maintenance-concurrency.test.mjs && node --test test/model-maintenance-integration.test.mjs && npm run test:workers && node test/comfyui-media.test.mjs && node test/qwen-image-diffusers.test.mjs && node test/runtime-residency.test.mjs && npm run test:ux && npm run test:memory-safety && node --test test/dashboard-memory.test.mjs test/runtime-memory-usage.test.mjs", + "test:unit": "node test/runtime-health-ownership.test.mjs && node test/long-prefill-transport.test.mjs && node test/sparkglm-managed.test.mjs && node test/protocol.test.mjs && node test/protocol-stream.test.mjs && node test/stream-metrics.test.mjs && node test/responses-codex.test.mjs && node test/security.test.mjs && node test/community-host.test.mjs && node test/openrouter-provider.test.mjs && node test/ds4fv-recipe.test.mjs && node test/dspark-hotfixes.test.mjs && node test/glm53-exl3-recipe.test.mjs && node test/host-memory.test.mjs && node test/config-profiles.test.mjs && node test/route-control.test.mjs && node test/runtime-policy.test.mjs && node --test test/rate-limit.test.mjs test/runtime-queue.test.mjs test/federated-priority.test.mjs test/runtime-supervisor.test.mjs && node test/interchange.test.mjs && node test/server-resilience.test.mjs && node test/metrics-persistence.test.mjs && node test/model-failover.test.mjs && node test/qwen38-vllm-recipe.test.mjs && node test/reasoning-normalize.test.mjs && node test/reasoning-effort.test.mjs && node test/route-control.test.mjs && node test/spark-route-catalog.test.mjs && node test/chat-lane-canary.test.mjs && node test/runtime-watchdog.test.mjs && node test/tts-catalog.test.mjs && node test/voice-profiles.test.mjs && node test/chatterbox-recipe.test.mjs && node test/hear-recipe.test.mjs && npm run test:adoption && node --test test/performance-sampler.test.mjs && node --test test/audio-generations.test.mjs && node test/video-providers.test.mjs && node --test test/web-functions.test.mjs && node --test test/config-mutation.test.mjs && node --test test/model-maintenance.test.mjs && node --test test/model-maintenance-control.test.mjs && node --test test/route-maintenance-concurrency.test.mjs && node --test test/model-maintenance-integration.test.mjs && npm run test:workers && node test/comfyui-media.test.mjs && node test/qwen-image-diffusers.test.mjs && node test/runtime-residency.test.mjs && npm run test:ux && npm run test:memory-safety && node --test test/dashboard-memory.test.mjs test/runtime-memory-usage.test.mjs && node --test test/speech-stream.test.mjs", "test:cluster": "node test/cluster.test.mjs", "test": "npm run test:cluster && npm run test:unit && npm run test:entity-stagger && npm run smoke && npm run community:check", "test:entity-stagger": "node test/entity-stagger-benchmark.test.mjs", diff --git a/src/server.mjs b/src/server.mjs index 5cac260..5dfb7b9 100644 --- a/src/server.mjs +++ b/src/server.mjs @@ -18,6 +18,8 @@ import { writeFileSync } from 'node:fs'; import path from 'node:path'; +import { open as openFile, realpath } from 'node:fs/promises'; +import { constants as fsConstants } from 'node:fs'; import os from 'node:os'; import { execFile } from 'node:child_process'; import { promisify } from 'node:util'; @@ -669,7 +671,11 @@ function copyResponseHeaders(upstream) { 'retry-after', 'x-lloom-provider', 'x-lloom-provider-job-id', - 'x-lloom-upstream-model' + 'x-lloom-upstream-model', + // Raw PCM speech streaming tells clients how to interpret the bytes. + 'x-audio-sample-rate', + 'x-audio-channels', + 'x-audio-format' ]) { const value = upstream.headers.get(name); if (value) headers[name] = value; @@ -2743,7 +2749,8 @@ export function createLloomServer( noteRuntimeRequestOutcome(resolved.model.runtime, outcome); clusterCoordinator.noteTargetOutcome(resolved, outcome); if (status === 499 || isClientClosedError(error)) { - endResponseWithError(res, error, { stream, config, status: 499 }); + if (binaryResponse && res.headersSent) res.destroy(); + else endResponseWithError(res, error, { stream: stream && !binaryResponse, config, status: 499 }); return { status: 499, stream, @@ -2767,7 +2774,7 @@ export function createLloomServer( } // Upstream death mid-stream (Metal abort, connection reset): finish SSE/JSON without // rethrowing into the outer handler (which would try writeHead again and crash Node). - if (res.headersSent || stream) { + if (res.headersSent || (stream && !binaryResponse)) { endResponseWithError(res, error, { stream: true, config, @@ -3250,16 +3257,45 @@ export function createLloomServer( }); // Keep named profile id in voice for logging; clone backends ignore unknown speakers. if (profile?.id) normalized.voice = profile.id; + // Only expand the trusted named-profile reference. Caller-supplied paths + // are never opened by the gateway. This also works across container mounts. + if (profile?.refAudioPath && normalized.ref_audio === profile.refAudioPath) { + const root = await realpath(voicesRoot()); + const reference = await realpath(profile.refAudioPath); + if (!reference.startsWith(root + path.sep)) throw new Error('voice reference is outside the voice registry'); + const file = await openFile(reference, fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW); + try { + const stat = await file.stat(); + if (!stat.isFile() || stat.size > 16 * 1024 * 1024) throw new Error('invalid voice reference size'); + const bytes = Buffer.alloc(stat.size + 1); + let bytesRead = 0; + while (bytesRead < bytes.length) { + const part = await file.read(bytes, bytesRead, bytes.length - bytesRead, bytesRead); + if (!part.bytesRead) break; + bytesRead += part.bytesRead; + } + if (bytesRead !== stat.size) throw new Error('voice reference changed while reading'); + normalized.ref_audio = `data:audio/wav;base64,${bytes.subarray(0, bytesRead).toString('base64')}`; + } finally { + await file.close(); + } + } + // Opt-in low-latency streaming: only real PCM payloads can be streamed as + // bytes (WAV needs a header with the final length). Any other combination + // falls back to the buffered compatibility path so normal WAV is unchanged. + const wantsStream = normalized.stream === true && String(normalized.response_format ?? '').toLowerCase() === 'pcm'; + normalized.stream = wantsStream; await recordModelRequest( { route: '/v1/audio/speech', resolved, - stream: false, + stream: wantsStream, + binaryResponse: wantsStream, req, res, voiceProfile: profile?.id ?? null }, - async ({ signal, timing, watchdog }) => { + async ({ signal, timing, progress, watchdog }) => { watchdog.arm(); const upstream = await fetchUpstream({ headers: inferenceGatewayHeaders(req, resolved), @@ -3268,6 +3304,19 @@ export function createLloomServer( signal, body: normalized }); + if (wantsStream) { + if (!upstream.ok) { + // Let upstream validation errors (400/404/…) surface as buffered JSON + // with their original status instead of a partial byte stream. + return proxyRawResponse(res, upstream, { signal, timing, corsConfig: config }); + } + return proxyRawResponseStreaming(res, upstream, { + signal, + timing, + progress, + corsConfig: config + }); + } return proxyRawResponse(res, upstream, { signal, timing, corsConfig: config }); } ); @@ -3296,7 +3345,8 @@ export function createLloomServer( min_p: multipartTextField(raw, type, 'min_p'), language_id: multipartTextField(raw, type, 'language_id'), audio_prompt_path: multipartTextField(raw, type, 'audio_prompt_path'), - response_format: multipartTextField(raw, type, 'response_format') + response_format: multipartTextField(raw, type, 'response_format'), + stream: multipartTextField(raw, type, 'stream') === 'true' }; const expanded = await expandSpeechBody(fields); // Named profile: expand to JSON clone request (ref on disk) so clients only send voice+input. @@ -3330,16 +3380,41 @@ export function createLloomServer( ); return; } + const multipartWantsStream = + String(fields.stream ?? '').toLowerCase() === 'true' && + String(fields.response_format ?? '').toLowerCase() === 'pcm'; await recordModelRequest( { route: '/v1/audio/speech', resolved, - stream: false, + stream: multipartWantsStream, + binaryResponse: multipartWantsStream, req, res }, - async ({ signal, timing, watchdog }) => { + async ({ signal, timing, progress, watchdog }) => { watchdog.arm(); + if (multipartWantsStream) { + const upstream = await fetchRawUpstream({ + backend: resolved.backend, + path: '/v1/audio/speech', + body: upstreamBody, + signal, + headers: { + ...inferenceGatewayHeaders(req, resolved), + 'content-type': type + } + }); + if (!upstream.ok) { + return proxyRawResponse(res, upstream, { signal, timing, corsConfig: config }); + } + return proxyRawResponseStreaming(res, upstream, { + signal, + timing, + progress, + corsConfig: config + }); + } const upstream = await fetchRawUpstream({ backend: resolved.backend, path: '/v1/audio/speech', diff --git a/test/speech-stream.test.mjs b/test/speech-stream.test.mjs new file mode 100644 index 0000000..5f417e3 --- /dev/null +++ b/test/speech-stream.test.mjs @@ -0,0 +1,273 @@ +import assert from 'node:assert/strict'; +import http from 'node:http'; +import test from 'node:test'; +import { once } from 'node:events'; +import { createLloomServer } from '../src/server.mjs'; + +// PCM16 LE signed frames: 0, +1, -1 as raw bytes (no container). Keep the +// values inside int16 range so the fixture is a realistic payload. +const pcmA = Buffer.from([0x00, 0x00, 0x01, 0x00, 0xff, 0xff]); +const pcmB = Buffer.from([0x10, 0x27, 0xf0, 0xd8]); +const wavBytes = Buffer.from('RIFFlloom-wav-body'); + +async function listen(server) { + server.listen(0, '127.0.0.1'); + await once(server, 'listening'); + return `http://127.0.0.1:${server.address().port}`; +} + +// The upstream stands in for the mlx-audio backend: for stream requests it +// emits signed PCM16 with the raw `audio/L16` content type and rate headers. +async function fixture(t, respond) { + const calls = []; + const backend = http.createServer(async (req, res) => { + const chunks = []; + for await (const chunk of req) chunks.push(chunk); + const rawBody = Buffer.concat(chunks); + const body = /multipart\/form-data/i.test(req.headers['content-type'] ?? '') + ? { multipart: true } + : JSON.parse(rawBody); + calls.push({ path: req.url, headers: req.headers, body }); + if (body.stream === true && String(body.response_format).toLowerCase() === 'pcm') { + res.setHeader('content-type', 'audio/pcm; rate=24000; channels=1'); + res.setHeader('x-audio-sample-rate', '24000'); + res.setHeader('x-audio-channels', '1'); + res.setHeader('x-audio-format', 'pcm_s16le'); + } else { + res.setHeader('content-type', 'audio/wav'); + } + respond(req, res, body); + }); + const baseUrl = await listen(backend); + const config = { + server: { host: '127.0.0.1', port: 0 }, + security: { allowMissingAuth: true, apiKeys: [] }, + logging: { metricsPersistence: false }, + telemetry: { performanceSampler: false }, + defaults: { speechModel: 'speech' }, + backends: { 'synthetic-speech': { type: 'openai', baseUrl: `${baseUrl}/v1`, apiKey: 'fixture-key' } }, + models: [ + { + id: 'speech', + backend: 'synthetic-speech', + upstreamModel: 'upstream-speech', + kind: 'audio_speech', + tts: { family: 'qwen3-tts', mode: 'custom_voice' } + }, + { id: 'chat', backend: 'synthetic-speech', upstreamModel: 'upstream-chat', kind: 'chat' } + ] + }; + const app = createLloomServer(config, { logger: { error() {}, warn() {} } }); + const gateway = await listen(app.server); + t.after(async () => { + app.server.closeAllConnections(); + backend.closeAllConnections(); + await app.close({ stopRuntimes: false }); + await new Promise((resolve) => backend.close(resolve)); + }); + return { + calls, + config, + metrics: app.metrics, + gateway, + request(body, options = {}) { + return fetch(gateway + '/v1/audio/speech', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: JSON.stringify(body), + ...options + }); + } + }; +} + +test('normal wav speech stays buffered and unchanged', async (t) => { + const f = await fixture(t, (_req, res) => res.end(wavBytes)); + const response = await f.request({ model: 'speech', input: 'hello', response_format: 'wav' }); + assert.equal(response.status, 200); + assert.equal(response.headers.get('content-type'), 'audio/wav'); + assert.deepEqual(Buffer.from(await response.arrayBuffer()), wavBytes); + assert.equal(f.calls.at(-1).body.stream ?? false, false); + assert.equal(f.calls.at(-1).body.response_format, 'wav'); +}); + +test('stream:true + response_format:pcm streams raw bytes before completion', { timeout: 5000 }, async (t) => { + let finish; + const f = await fixture(t, (_req, res) => { + res.writeHead(200); + res.write(pcmA); + finish = () => res.end(pcmB); + }); + const response = await f.request({ model: 'speech', input: 'hello', stream: true, response_format: 'pcm' }); + assert.equal(response.status, 200); + assert.equal(f.calls.at(-1).body.stream, true); + assert.equal(f.calls.at(-1).body.response_format, 'pcm'); + assert.equal(f.calls.at(-1).headers.authorization, 'Bearer fixture-key'); + // First chunk is readable while the upstream is still open. + const reader = response.body.getReader(); + const first = await reader.read(); + assert.deepEqual(Buffer.from(first.value), pcmA); + assert.equal(first.done, false); + finish(); + const second = await reader.read(); + assert.deepEqual(Buffer.from(second.value), pcmB); + assert.equal((await reader.read()).done, true); +}); + +test('stream pcm forwards declared sample-rate and format headers', async (t) => { + const f = await fixture(t, (_req, res) => res.end(pcmA)); + const response = await f.request({ model: 'speech', input: 'hi', stream: true, response_format: 'pcm' }); + assert.equal(response.status, 200); + assert.match(response.headers.get('content-type'), /audio\/pcm/i); + assert.equal(response.headers.get('x-audio-sample-rate'), '24000'); + assert.equal(response.headers.get('x-audio-format'), 'pcm_s16le'); + assert.equal(response.headers.get('x-audio-channels'), '1'); + assert.deepEqual(Buffer.from(await response.arrayBuffer()), pcmA); +}); + +test('stream request for a non-pcm format falls back to buffered wav', { timeout: 5000 }, async (t) => { + const f = await fixture(t, (_req, res) => res.end(wavBytes)); + const response = await f.request({ model: 'speech', input: 'hi', stream: true, response_format: 'wav' }); + assert.equal(response.status, 200); + assert.equal(response.headers.get('content-type'), 'audio/wav'); + assert.deepEqual(Buffer.from(await response.arrayBuffer()), wavBytes); + // Not forwarded as a stream request to the backend. + assert.notEqual(f.calls.at(-1).body.stream, true); +}); + +test('stream auth errors and wrong kinds are preserved as JSON', async (t) => { + const f = await fixture(t, (_req, res) => res.end(pcmA)); + const wrongKind = await f.request({ model: 'chat', input: 'hi', stream: true, response_format: 'pcm' }); + assert.equal(wrongKind.status, 400); + assert.equal((await wrongKind.json()).error.code, 'wrong_model_kind'); + const unknown = await f.request({ model: 'nope', input: 'hi', stream: true, response_format: 'pcm' }); + assert.equal(unknown.status, 404); + assert.equal((await unknown.json()).error.code, 'unknown_model'); + assert.equal(f.calls.length, 0); +}); + +test('upstream validation error surfaces as buffered JSON, not a truncated stream', async (t) => { + const f = await fixture(t, (_req, res) => { + res.writeHead(400, { 'content-type': 'application/json' }); + res.end( + JSON.stringify({ + error: { message: 'stream=true requires response_format=pcm', code: 'unsupported_stream_format' } + }) + ); + }); + const response = await f.request({ model: 'speech', input: 'hi', stream: true, response_format: 'pcm' }); + assert.equal(response.status, 400); + assert.match(response.headers.get('content-type'), /application\/json/); + assert.equal((await response.json()).error.code, 'unsupported_stream_format'); +}); + +test('client cancellation aborts the upstream stream', { timeout: 5000 }, async (t) => { + let closed; + const upstreamClosed = new Promise((resolve) => { + closed = resolve; + }); + const f = await fixture(t, (_req, res) => { + res.on('close', closed); + res.write(pcmA); + }); + const controller = new AbortController(); + const response = await f.request( + { model: 'speech', input: 'hi', stream: true, response_format: 'pcm' }, + { signal: controller.signal } + ); + const reader = response.body.getReader(); + await reader.read(); + controller.abort(); + await upstreamClosed; +}); + +test('malformed stream JSON is rejected before reaching the backend', async (t) => { + const f = await fixture(t, (_req, res) => res.end(pcmA)); + const response = await fetch(f.gateway + '/v1/audio/speech', { + method: 'POST', + headers: { 'content-type': 'application/json' }, + body: '{ not json' + }); + assert.ok(response.status >= 400); + await response.arrayBuffer(); + assert.equal(f.calls.length, 0); +}); + +test('multipart pcm speech streams before completion', { timeout: 5000 }, async (t) => { + let finish; + const f = await fixture(t, (_req, res) => { + res.writeHead(200, { + 'content-type': 'audio/pcm; rate=24000; channels=1', + 'x-audio-format': 'pcm_s16le' + }); + res.write(pcmA); + finish = () => res.end(pcmB); + }); + const form = new FormData(); + form.set('model', 'speech'); + form.set('input', 'hello'); + form.set('response_format', 'pcm'); + form.set('stream', 'true'); + const response = await fetch(f.gateway + '/v1/audio/speech', { method: 'POST', body: form }); + const reader = response.body.getReader(); + const first = await reader.read(); + assert.deepEqual(Buffer.from(first.value), pcmA); + finish(); + assert.deepEqual(Buffer.from((await reader.read()).value), pcmB); +}); + +test('multipart wav speech stays buffered', async (t) => { + const f = await fixture(t, (_req, res) => res.end(wavBytes)); + const form = new FormData(); + form.set('model', 'speech'); + form.set('input', 'hello'); + form.set('response_format', 'wav'); + form.set('stream', 'true'); + const response = await fetch(f.gateway + '/v1/audio/speech', { method: 'POST', body: form }); + assert.equal(response.headers.get('content-type'), 'audio/wav'); + assert.deepEqual(Buffer.from(await response.arrayBuffer()), wavBytes); +}); + +test('named profile embeds only its bounded server-owned reference for container backends', async (t) => { + const fs = await import('node:fs/promises'); + const os = await import('node:os'); + const path = await import('node:path'); + const root = await fs.mkdtemp(path.join(os.tmpdir(), 'speech-profile-')); + t.after(() => fs.rm(root, { recursive: true, force: true })); + const directory = path.join(root, 'synthetic'); + await fs.mkdir(directory); + const reference = Buffer.from('synthetic-reference-bytes'); + await fs.writeFile(path.join(directory, 'reference.wav'), reference); + await fs.writeFile( + path.join(directory, 'profile.json'), + JSON.stringify({ + id: 'synthetic', + name: 'Synthetic', + kind: 'voice_clone', + model: 'speech', + refAudio: 'reference.wav', + refText: 'Synthetic reference' + }) + ); + const f = await fixture(t, (_req, res) => res.end(pcmA)); + f.config.paths = { voicesRoot: root }; + const response = await f.request({ voice: 'synthetic', input: 'hello', stream: true, response_format: 'pcm' }); + assert.equal(response.status, 200); + await response.arrayBuffer(); + assert.equal(f.calls.at(-1).body.ref_audio, 'data:audio/wav;base64,' + reference.toString('base64')); + const supplied = '/untrusted/client/supplied.wav'; + const explicit = await f.request({ voice: 'synthetic', input: 'hello', ref_audio: supplied }); + await explicit.arrayBuffer(); + assert.equal(f.calls.at(-1).body.ref_audio, supplied, 'gateway must not open caller paths'); +}); + +test('partial binary upstream failure does not append SSE error bytes', async (t) => { + const f = await fixture(t, (_req, res) => { + res.write(pcmA); + setTimeout(() => res.destroy(), 30); + }); + const response = await f.request({ model: 'speech', input: 'hello', stream: true, response_format: 'pcm' }); + const reader = response.body.getReader(); + assert.deepEqual(Buffer.from((await reader.read()).value), pcmA); + await assert.rejects(reader.read()); +}); From feca31e0446786e9468de1e5c0973c3f1b981bd5 Mon Sep 17 00:00:00 2001 From: data-angel Date: Sat, 26 Sep 2026 16:15:38 -0700 Subject: [PATCH 2/3] Upload installed voice references for remote clone backends --- src/voice-profiles.mjs | 22 +++++++++++++++++++++- test/voice-profiles.test.mjs | 25 ++++++++++++++++++++++++- 2 files changed, 45 insertions(+), 2 deletions(-) diff --git a/src/voice-profiles.mjs b/src/voice-profiles.mjs index 655d0bf..0a5899a 100644 --- a/src/voice-profiles.mjs +++ b/src/voice-profiles.mjs @@ -379,5 +379,25 @@ export async function resolveSpeechVoice(body = {}, { voicesRoot = defaultVoices if (!profile) { return { body, profile: null, applied: false, unknownVoice: false }; } - return applyVoiceProfileToSpeechBody(body, profile); + const expanded = applyVoiceProfileToSpeechBody(body, profile); + // Only a trusted installed profile may cause a local file read. Caller-supplied + // references remain untouched for the backend's normal input validation. + if (body.ref_audio == null && body.refAudio == null) { + const handle = await fs.open(profile.refAudioPath, 'r'); + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.size <= 0 || stat.size > 16 * 1024 * 1024) { + throw new Error('voice reference must be a nonempty audio file at most 16 MiB'); + } + const audio = await handle.readFile(); + const mime = + { '.mp3': 'audio/mpeg', '.flac': 'audio/flac', '.ogg': 'audio/ogg', '.m4a': 'audio/mp4' }[ + path.extname(profile.refAudioPath).toLowerCase() + ] ?? 'audio/wav'; + expanded.body.ref_audio = `data:${mime};base64,${audio.toString('base64')}`; + } finally { + await handle.close(); + } + } + return expanded; } diff --git a/test/voice-profiles.test.mjs b/test/voice-profiles.test.mjs index 858f47f..3a2ce64 100644 --- a/test/voice-profiles.test.mjs +++ b/test/voice-profiles.test.mjs @@ -8,7 +8,8 @@ import { listVoiceProfiles, listVoicesDiscovery, loadVoiceProfileFromDir, - removeVoiceProfile + removeVoiceProfile, + resolveSpeechVoice } from '../src/voice-profiles.mjs'; const tmp = await fs.mkdtemp(path.join(os.tmpdir(), 'lloom-voices-')); @@ -63,6 +64,28 @@ assert.equal(retargeted.body.ref_audio, loaded.refAudioPath); assert.equal(retargeted.body.exaggeration, 0.75); assert.equal(retargeted.body.cfg_weight, 0.5); +// HTTP-bound profile resolution uploads reference bytes across process/host boundaries. +const portable = await resolveSpeechVoice({ voice: 'character-demo', input: 'Portable clone.' }, { voicesRoot: tmp }); +assert.equal(portable.body.ref_audio, 'data:audio/wav;base64,' + Buffer.from('RIFFxxxxWAVEfmt ').toString('base64')); +assert.equal(portable.body.ref_text, expanded.body.ref_text); +assert.equal(portable.body.temperature, 1.05); +assert.equal(portable.body.model, loaded.model); +const explicit = await resolveSpeechVoice( + { voice: 'character-demo', ref_audio: '/caller/path/must-not-be-read' }, + { voicesRoot: tmp } +); +assert.equal(explicit.body.ref_audio, '/caller/path/must-not-be-read'); +const inline = await resolveSpeechVoice( + { voice: 'character-demo', refAudio: 'data:audio/wav;base64,AQID' }, + { voicesRoot: tmp } +); +assert.equal(inline.body.ref_audio, 'data:audio/wav;base64,AQID'); +const unknown = await resolveSpeechVoice({ voice: 'not-installed', input: 'Hello.' }, { voicesRoot: tmp }); +assert.equal(unknown.applied, false); +await fs.writeFile(loaded.refAudioPath, Buffer.alloc(0)); +await assert.rejects(resolveSpeechVoice({ voice: 'character-demo' }, { voicesRoot: tmp }), /nonempty audio file/); +await fs.writeFile(loaded.refAudioPath, Buffer.from('RIFFxxxxWAVEfmt ')); + const discovery = listVoicesDiscovery({ profiles: listed, modelVoices: { From 69db91b2aa94222b2abecc90274b07a8212130a2 Mon Sep 17 00:00:00 2001 From: Jason McCartney Date: Sun, 27 Sep 2026 11:59:29 -0700 Subject: [PATCH 3/3] Consolidate trusted voice reference inlining in the profile resolver The gateway inlined installed profile references in two places. Keep the resolver's version, which also covers remote clone backends, and give it the registry containment, no-follow open and size checks the server copy had. Co-Authored-By: Claude Opus 5.5 --- src/server.mjs | 25 ---------------------- src/voice-profiles.mjs | 40 +++++++++++++++++++++--------------- test/voice-profiles.test.mjs | 12 +++++++++++ 3 files changed, 36 insertions(+), 41 deletions(-) diff --git a/src/server.mjs b/src/server.mjs index 5dfb7b9..a1e0823 100644 --- a/src/server.mjs +++ b/src/server.mjs @@ -18,8 +18,6 @@ import { writeFileSync } from 'node:fs'; import path from 'node:path'; -import { open as openFile, realpath } from 'node:fs/promises'; -import { constants as fsConstants } from 'node:fs'; import os from 'node:os'; import { execFile } from 'node:child_process'; import { promisify } from 'node:util'; @@ -3257,29 +3255,6 @@ export function createLloomServer( }); // Keep named profile id in voice for logging; clone backends ignore unknown speakers. if (profile?.id) normalized.voice = profile.id; - // Only expand the trusted named-profile reference. Caller-supplied paths - // are never opened by the gateway. This also works across container mounts. - if (profile?.refAudioPath && normalized.ref_audio === profile.refAudioPath) { - const root = await realpath(voicesRoot()); - const reference = await realpath(profile.refAudioPath); - if (!reference.startsWith(root + path.sep)) throw new Error('voice reference is outside the voice registry'); - const file = await openFile(reference, fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW); - try { - const stat = await file.stat(); - if (!stat.isFile() || stat.size > 16 * 1024 * 1024) throw new Error('invalid voice reference size'); - const bytes = Buffer.alloc(stat.size + 1); - let bytesRead = 0; - while (bytesRead < bytes.length) { - const part = await file.read(bytes, bytesRead, bytes.length - bytesRead, bytesRead); - if (!part.bytesRead) break; - bytesRead += part.bytesRead; - } - if (bytesRead !== stat.size) throw new Error('voice reference changed while reading'); - normalized.ref_audio = `data:audio/wav;base64,${bytes.subarray(0, bytesRead).toString('base64')}`; - } finally { - await file.close(); - } - } // Opt-in low-latency streaming: only real PCM payloads can be streamed as // bytes (WAV needs a header with the final length). Any other combination // falls back to the buffered compatibility path so normal WAV is unchanged. diff --git a/src/voice-profiles.mjs b/src/voice-profiles.mjs index 0a5899a..1738150 100644 --- a/src/voice-profiles.mjs +++ b/src/voice-profiles.mjs @@ -8,7 +8,7 @@ */ import fs from 'node:fs/promises'; import path from 'node:path'; -import { createReadStream, existsSync } from 'node:fs'; +import { constants as fsConstants, createReadStream, existsSync } from 'node:fs'; import { pipeline } from 'node:stream/promises'; import { createWriteStream } from 'node:fs'; import { defaultLloomHome } from './config.mjs'; @@ -367,6 +367,28 @@ export async function removeVoiceProfile(voiceId, { voicesRoot = defaultVoicesRo return { removed: true, voiceId: id, directory }; } +const MAX_VOICE_REFERENCE_BYTES = 16 * 1024 * 1024; +const VOICE_REFERENCE_MIME = { '.mp3': 'audio/mpeg', '.flac': 'audio/flac', '.ogg': 'audio/ogg', '.m4a': 'audio/mp4' }; + +/** Inline a profile's reference audio. The bytes can leave the host, so it must stay in the registry. */ +async function readVoiceReference(refAudioPath, voicesRoot) { + const [root, reference] = await Promise.all([fs.realpath(voicesRoot), fs.realpath(refAudioPath)]); + if (!reference.startsWith(root + path.sep)) throw new Error('voice reference is outside the voice registry'); + const handle = await fs.open(reference, fsConstants.O_RDONLY | fsConstants.O_NOFOLLOW); + try { + const stat = await handle.stat(); + if (!stat.isFile() || stat.size <= 0 || stat.size > MAX_VOICE_REFERENCE_BYTES) { + throw new Error('voice reference must be a nonempty audio file at most 16 MiB'); + } + const audio = await handle.readFile(); + if (audio.length !== stat.size) throw new Error('voice reference changed while reading'); + const mime = VOICE_REFERENCE_MIME[path.extname(reference).toLowerCase()] ?? 'audio/wav'; + return `data:${mime};base64,${audio.toString('base64')}`; + } finally { + await handle.close(); + } +} + /** * Resolve speech request voice field against installed profiles. */ @@ -383,21 +405,7 @@ export async function resolveSpeechVoice(body = {}, { voicesRoot = defaultVoices // Only a trusted installed profile may cause a local file read. Caller-supplied // references remain untouched for the backend's normal input validation. if (body.ref_audio == null && body.refAudio == null) { - const handle = await fs.open(profile.refAudioPath, 'r'); - try { - const stat = await handle.stat(); - if (!stat.isFile() || stat.size <= 0 || stat.size > 16 * 1024 * 1024) { - throw new Error('voice reference must be a nonempty audio file at most 16 MiB'); - } - const audio = await handle.readFile(); - const mime = - { '.mp3': 'audio/mpeg', '.flac': 'audio/flac', '.ogg': 'audio/ogg', '.m4a': 'audio/mp4' }[ - path.extname(profile.refAudioPath).toLowerCase() - ] ?? 'audio/wav'; - expanded.body.ref_audio = `data:${mime};base64,${audio.toString('base64')}`; - } finally { - await handle.close(); - } + expanded.body.ref_audio = await readVoiceReference(profile.refAudioPath, voicesRoot); } return expanded; } diff --git a/test/voice-profiles.test.mjs b/test/voice-profiles.test.mjs index 3a2ce64..de9ab49 100644 --- a/test/voice-profiles.test.mjs +++ b/test/voice-profiles.test.mjs @@ -85,6 +85,18 @@ assert.equal(unknown.applied, false); await fs.writeFile(loaded.refAudioPath, Buffer.alloc(0)); await assert.rejects(resolveSpeechVoice({ voice: 'character-demo' }, { voicesRoot: tmp }), /nonempty audio file/); await fs.writeFile(loaded.refAudioPath, Buffer.from('RIFFxxxxWAVEfmt ')); +// Uploaded bytes may leave the host, so the reference must stay inside the registry. +const outside = await fs.mkdtemp(path.join(os.tmpdir(), 'lloom-voice-outside-')); +await fs.writeFile(path.join(outside, 'secret.wav'), Buffer.from('not a voice')); +await fs.rename(loaded.refAudioPath, `${loaded.refAudioPath}.bak`); +await fs.symlink(path.join(outside, 'secret.wav'), loaded.refAudioPath); +await assert.rejects( + resolveSpeechVoice({ voice: 'character-demo' }, { voicesRoot: tmp }), + /outside the voice registry/ +); +await fs.rm(loaded.refAudioPath); +await fs.rename(`${loaded.refAudioPath}.bak`, loaded.refAudioPath); +await fs.rm(outside, { recursive: true, force: true }); const discovery = listVoicesDiscovery({ profiles: listed,