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 1788363..daa3900 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 && node --check src/audio-providers.mjs && node --check test/audio-providers.test.mjs && node --check src/runtime-policy-config.mjs && node --check test/runtime-policy-config.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 && node --test test/audio-providers.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 && node --test test/audio-providers.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 7dff74a..18f7bc3 100644 --- a/src/server.mjs +++ b/src/server.mjs @@ -707,7 +707,11 @@ function copyResponseHeaders(upstream) { 'x-lloom-provider-job-id', 'x-lloom-provider-origin', 'x-lloom-audio-format', - '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; @@ -2781,7 +2785,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, @@ -2805,7 +2810,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, @@ -3366,16 +3371,22 @@ export function createLloomServer( }); // Keep named profile id in voice for logging; clone backends ignore unknown speakers. if (profile?.id) normalized.voice = profile.id; + // 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), @@ -3384,6 +3395,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 }); } ); @@ -3412,7 +3436,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. @@ -3446,16 +3471,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/src/voice-profiles.mjs b/src/voice-profiles.mjs index 655d0bf..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. */ @@ -379,5 +401,11 @@ 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) { + expanded.body.ref_audio = await readVoiceReference(profile.refAudioPath, voicesRoot); + } + return expanded; } 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()); +}); diff --git a/test/voice-profiles.test.mjs b/test/voice-profiles.test.mjs index 858f47f..de9ab49 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,40 @@ 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 ')); +// 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, modelVoices: {