aboutsummaryrefslogtreecommitdiff
path: root/app
diff options
context:
space:
mode:
Diffstat (limited to 'app')
-rw-r--r--app/backends/sglomni/__init__.py3
-rw-r--r--app/backends/sglomni/catalog.py13
-rw-r--r--app/backends/sglomni/constants.py3
-rw-r--r--app/backends/sglomni/models.py87
-rw-r--r--app/backends/sglomni/pythonenv.py3
-rw-r--r--app/backends/sglomni/status.py16
-rw-r--r--app/backends/sglomni/wizard.py103
-rw-r--r--app/converter/clients/sglomni.py147
-rw-r--r--app/tests/test_backends_sglomni.py291
-rw-r--r--app/tests/test_tts_sglomni.py99
10 files changed, 591 insertions, 174 deletions
diff --git a/app/backends/sglomni/__init__.py b/app/backends/sglomni/__init__.py
index be94a0e..ed7d4ed 100644
--- a/app/backends/sglomni/__init__.py
+++ b/app/backends/sglomni/__init__.py
@@ -22,7 +22,6 @@ from .constants import (
SERVER_NAME,
SERVER_START_TIMEOUT,
SGLOMNI_PIP_PKG,
- UV_PIP_PKG,
)
from .gpu import (
compute_capability,
@@ -79,7 +78,7 @@ from .wizard import (
__all__ = [
# constants
"CONFIGS_DIR", "DEFAULT_PORT", "PYTHON_SPEC", "PYTHON_VERSIONS",
- "SERVER_NAME", "SERVER_START_TIMEOUT", "SGLOMNI_PIP_PKG", "UV_PIP_PKG",
+ "SERVER_NAME", "SERVER_START_TIMEOUT", "SGLOMNI_PIP_PKG",
# gpu
"compute_capability", "describe",
# catalog
diff --git a/app/backends/sglomni/catalog.py b/app/backends/sglomni/catalog.py
index f20509f..8bfb1fa 100644
--- a/app/backends/sglomni/catalog.py
+++ b/app/backends/sglomni/catalog.py
@@ -52,6 +52,11 @@ class ModelEntry:
system_hint: Optional[str] = None # remediation when the binary is absent
speakers: Optional[Tuple[str, ...]] = None # preset voices (speaker)
supports_seed: bool = False # request-scoped seed accepted (Qwen3-TTS Base)
+ # NOTE(unverified upstream): only the two Base entries are known to
+ # accept a request-scoped seed (Voxtral rejects one outright); qwen's
+ # demo client does send seeds to the CustomVoice/VoiceDesign models,
+ # so those pipelines may accept one too — verify before flipping the
+ # flag (CONSTANT_SEED currently no-ops for every other entry).
notes: str = "" # one-line description (documentation)
# The model's DEFAULT pipeline dynamically quantizes its MoE experts to
# FP8 at load time (sglang-omni's zonos2 config hardcodes it) — a Triton
@@ -103,6 +108,12 @@ _QWEN_EXTRAS: Tuple[Extra, ...] = (
("sox", True), ("einops", True), ("qwen-tts==0.1.1", True))
_SOX_HINT = ("install the sox system package (e.g. sudo pacman -S sox, "
"sudo apt install sox, brew install sox)")
+# The Fish Audio and ZONOS2 pipelines decode their codec assets through
+# the ffmpeg binary (and the client concatenates multi-part chunks with
+# it); the weights download fine without it, the server just fails to
+# synthesize — so the install flow warns, like it does for sox.
+_FFMPEG_HINT = ("install the ffmpeg system package (e.g. sudo pacman -S "
+ "ffmpeg, sudo apt install ffmpeg, brew install ffmpeg)")
# The Fish Audio and ZONOS2 pipelines use the Descript DAC codec, which
# upstream installs WITH dependencies — but descript-audiotools carries a
# vestigial 2021-era pin, protobuf<3.20 (its code never imports protobuf),
@@ -255,6 +266,7 @@ ENTRIES: Tuple[ModelEntry, ...] = (
capability=CAPABILITY_CLONE,
requires_reference=False,
extras=_DAC_EXTRAS,
+ system_dep="ffmpeg", system_hint=_FFMPEG_HINT,
notes="zero-shot narration or cloning from a reference clip",
),
ModelEntry(
@@ -265,6 +277,7 @@ ENTRIES: Tuple[ModelEntry, ...] = (
capability=CAPABILITY_CLONE,
requires_reference=True,
extras=_DAC_EXTRAS,
+ system_dep="ffmpeg", system_hint=_FFMPEG_HINT,
notes="voice cloning, 44.1 kHz DAC vocoder",
fp8_moe=True,
fp8_min_compute_capability=(8, 9),
diff --git a/app/backends/sglomni/constants.py b/app/backends/sglomni/constants.py
index 94a07bd..945ce8a 100644
--- a/app/backends/sglomni/constants.py
+++ b/app/backends/sglomni/constants.py
@@ -6,9 +6,6 @@ from pathlib import Path
# (like qwen-tts and faster-qwen3-tts) so the update action can move with
# upstream releases; the stack this code was verified against is 0.1.4.
SGLOMNI_PIP_PKG = "sglang-omni"
-# uv is provisioned into the app env only when a Python the sglang-omni
-# stack accepts (>=3.10,<3.13) is not already available.
-UV_PIP_PKG = "uv"
# The dedicated venv is backends.envs.SGLOMNI_ENV_DIR (imported from there
# by the modules that need it) — its interpreter may differ from the
diff --git a/app/backends/sglomni/models.py b/app/backends/sglomni/models.py
index 607e36a..990673a 100644
--- a/app/backends/sglomni/models.py
+++ b/app/backends/sglomni/models.py
@@ -16,31 +16,25 @@ VoiceDesign exist in the qwen backend too) is downloaded once and its
deletion affects both — the same convention every backend here accepts.
"""
-import os
import shutil
from pathlib import Path
from typing import List, Optional
from backends import common, envs
-from backends.sglomni.catalog import ModelEntry, entry_by_key, \
- entry_by_repo, extra_import_name
+from backends.sglomni.catalog import ENTRIES, Extra, ModelEntry, \
+ entry_by_key, entry_by_repo, extra_import_name
from backends.sglomni.constants import SERVER_NAME, SGLOMNI_PIP_PKG
from backends.sglomni.pythonenv import SGLOMNI_ENV, prepare_env
-# The cache directory HF keeps repos in (models--<org>--<name> folders).
-# Resolution mirrors huggingface_hub.constants: HF_HUB_CACHE beats
-# HUGGINGFACE_HUB_CACHE beats HF_HOME/hub beats ~/.cache/huggingface/hub.
+# The HF-cache primitives live in backends.common (shared with the qwen
+# backend, which fetches into the same cache); the module-level wrappers
+# below keep the names internal callers (and the tests) reference.
def _hf_cache_dir() -> Path:
- override = os.environ.get("HF_HUB_CACHE") or os.environ.get(
- "HUGGINGFACE_HUB_CACHE")
- if override:
- return Path(override)
- home = os.environ.get("HF_HOME")
- if home:
- return Path(home) / "hub"
- return Path.home() / ".cache" / "huggingface" / "hub"
+ """The cache directory HF keeps repos in (models--<org>--<name>
+ folders); common.hf_cache_dir resolves the environment overrides."""
+ return common.hf_cache_dir()
def repo_dir(repo_id: str) -> Path:
@@ -55,17 +49,7 @@ def model_repo_dir(entry: ModelEntry) -> Path:
def _tree_has_file(path: Path) -> bool:
"""True when any file or symlink exists under PATH (recursively)."""
- try:
- for item in path.iterdir():
- # Snapshot files are symlinks into blobs/; count them even when
- # temporarily broken (presence is what the loader checks).
- if item.is_symlink() or item.is_file():
- return True
- if item.is_dir() and _tree_has_file(item):
- return True
- except OSError:
- return False
- return False
+ return common.hf_tree_has_file(path)
def model_installed(entry: ModelEntry) -> bool:
@@ -83,7 +67,7 @@ def model_installed(entry: ModelEntry) -> bool:
def installed_entries() -> List[ModelEntry]:
"""The catalog entries whose weights are already on disk."""
- return [entry for entry in _all_entries() if model_installed(entry)]
+ return [entry for entry in ENTRIES if model_installed(entry)]
def installed_keys() -> List[str]:
@@ -115,11 +99,6 @@ def preset_voices(entry: ModelEntry) -> List[str]:
return []
-def _all_entries() -> List[ModelEntry]:
- from backends.sglomni.catalog import ENTRIES
- return list(ENTRIES)
-
-
def system_dep_missing(entry: ModelEntry) -> Optional[str]:
"""Remediation text when ENTRY's system binary is absent (None = ok)."""
if entry.system_dep and not shutil.which(entry.system_dep):
@@ -147,15 +126,20 @@ def missing_companions(entry: ModelEntry) -> List[Extra]:
SGLOMNI_ENV)]
-def install_companions(entry: ModelEntry, *, emit=None, cancel=None) -> int:
- """pip-install ENTRY's missing companion packages into the venv.
+def install_companions(entry: ModelEntry, *, emit=None, cancel=None,
+ force: bool = False) -> int:
+ """pip-install ENTRY's companion packages into the venv.
The same recipe ``install_model`` runs (the catalog's ``--no-deps``
flags preserved — the Qwen3-TTS companions must not replace the pinned
Transformers 5 stack), limited to what the import probe found absent,
so a start-time heal touches as little of the pinned environment as
- possible. Returns the first failing exit code, 0 when all present."""
- for spec, no_deps in missing_companions(entry):
+ possible. With FORCE every extra's pip spec re-runs instead — a
+ satisfied pin is a pip no-op, so the update flow uses that to heal
+ version drift the import probe cannot see (the protobuf re-pin
+ especially). Returns the first failing exit code, 0 when all present."""
+ wanted = list(entry.extras) if force else missing_companions(entry)
+ for spec, no_deps in wanted:
args = ["--no-deps"] if no_deps else None
rc = common.pip_install([spec], emit=emit, cancel=cancel,
env_dir=SGLOMNI_ENV, extra_args=args)
@@ -237,36 +221,19 @@ def uninstall_model(key: str, *, emit=None, cancel=None) -> int:
def delete_model_weights(entries: Optional[List[ModelEntry]] = None) -> int:
"""Delete the cached HF weight dirs of ENTRIES (every model by default).
- Best-effort rmtree of each ``models--<org>--<name>`` directory; returns
- how many were present and removed. Only those directories are ever
- touched — the rest of the HF cache may be shared with unrelated tools.
+ Delegates to common.hf_delete_model_weights: best-effort rmtree of
+ each ``models--<org>--<name>`` directory; only those directories are
+ ever touched — the rest of the HF cache may be shared with unrelated
+ tools. Returns how many were present and removed.
"""
if entries is None:
- entries = _all_entries()
- removed = 0
- for entry in entries:
- directory = model_repo_dir(entry)
- if not directory.is_dir():
- continue
- print(f"[INFO] Removing cached {entry.repo} weights...")
- shutil.rmtree(directory, ignore_errors=True)
- if directory.exists():
- print(f"[WARNING] Could not fully remove {directory}")
- continue
- removed += 1
- if removed:
- print(f"[OK] Deleted cached weights for {removed} "
- f"{'model' if removed == 1 else 'models'}.")
- return removed
+ entries = list(ENTRIES)
+ return common.hf_delete_model_weights([entry.repo for entry in entries])
def _hf_download_prefix() -> Optional[List[str]]:
"""The sglang-omni venv's hf CLI argv prefix (None when absent)."""
- for name in ("hf", "huggingface-cli"):
- candidate = envs.env_script(name, SGLOMNI_ENV)
- if candidate.is_file():
- return [str(candidate)]
- return None
+ return common.hf_download_prefix(SGLOMNI_ENV)
def _managed_running_repo() -> Optional[str]:
@@ -292,7 +259,7 @@ def resolve_model(key: Optional[str]) -> ModelEntry:
if key is not None:
entry = entry_by_key(key)
if entry is None:
- known = ", ".join(e.key for e in _all_entries())
+ known = ", ".join(e.key for e in ENTRIES)
raise RuntimeError(
f"Unknown sglang-omni model {key!r} (installed models are "
f"picked by catalog key; known keys: {known})")
diff --git a/app/backends/sglomni/pythonenv.py b/app/backends/sglomni/pythonenv.py
index 11bbaa6..cb19710 100644
--- a/app/backends/sglomni/pythonenv.py
+++ b/app/backends/sglomni/pythonenv.py
@@ -21,10 +21,9 @@ the setup that follows anyway.
import shutil
import sys
-from pathlib import Path
from typing import Optional, Tuple
-from backends import common, envs
+from backends import envs
from backends.sglomni.constants import PYTHON_SPEC, PYTHON_VERSIONS
# The dedicated venv (its interpreter may differ from the launching one).
diff --git a/app/backends/sglomni/status.py b/app/backends/sglomni/status.py
index 2b77307..6a47cff 100644
--- a/app/backends/sglomni/status.py
+++ b/app/backends/sglomni/status.py
@@ -3,8 +3,8 @@
from pathlib import Path
from typing import List, Optional
-from backends import BackendStatus, ServerSpec, envs, format_launch_hint, \
- probe, servers
+from backends import BackendStatus, ServerSpec, common, envs, \
+ format_launch_hint, probe, servers
from backends.sglomni import gpu
from backends.sglomni.catalog import ModelEntry, entry_by_repo, \
fallback_config_path, config_path
@@ -101,15 +101,9 @@ def gpu_fallback_note(entry: ModelEntry) -> Optional[str]:
def _port() -> int:
- return _port_of(config.SGLOMNI_API_URL)
-
-
-def _port_of(url: str) -> int:
- import urllib.parse
- try:
- return urllib.parse.urlsplit(url).port or DEFAULT_PORT
- except ValueError:
- return DEFAULT_PORT
+ """The managed server's port (the configured URL's explicit port,
+ else the backend default)."""
+ return common.port_of(config.SGLOMNI_API_URL, DEFAULT_PORT)
def _managed_running_entry() -> Optional[ModelEntry]:
diff --git a/app/backends/sglomni/wizard.py b/app/backends/sglomni/wizard.py
index 96679d6..f39e754 100644
--- a/app/backends/sglomni/wizard.py
+++ b/app/backends/sglomni/wizard.py
@@ -13,7 +13,7 @@ upstream recipes + HuggingFace weight pre-download).
It is driven by ``audiobook.py``'s hub but can also be run directly:
Usage:
- python -m backends.sglomni [--models KEY[,KEY...]] [--all]
+ python -m backends.sglomni [KEY ...] [--models KEY[,KEY...]] [--all]
[--skip-install] [--skip-python]
The Configure screen (``models_screen``) manages models after the fact —
@@ -32,9 +32,9 @@ sys.path.insert(0, str(Path(__file__).resolve().parent.parent.parent))
from backends import common, envs, servers, setup
from backends.sglomni import catalog as sg_catalog
+from backends.sglomni import gpu as sg_gpu
from backends.sglomni import models as sg_models
-from backends.sglomni.constants import SERVER_NAME, SGLOMNI_PIP_PKG, \
- UV_PIP_PKG
+from backends.sglomni.constants import SERVER_NAME, SGLOMNI_PIP_PKG
from backends.sglomni.pythonenv import SGLOMNI_ENV, prepare_env
from backends.sglomni.status import _is_installed
from ui import taskview, tui
@@ -43,10 +43,11 @@ _GO_BACK = object()
def _nvidia_gpu_present() -> bool:
- """True when an NVIDIA driver answers nvidia-smi (best effort)."""
- proc = common.run_console_subprocess_quiet(
- ["nvidia-smi", "-L"], timeout=10)
- return proc is not None and proc.returncode == 0
+ """True when an NVIDIA driver answers nvidia-smi (best effort).
+
+ The same probe the launch decisions use (sglomni.gpu): nvidia-smi
+ names GPU 0, or there is no usable answer."""
+ return sg_gpu.describe() is not None
def _preflight() -> List[str]:
@@ -240,6 +241,10 @@ def run_tui(args: Optional[argparse.Namespace] = None) -> int:
settings = curses.wrapper(lambda scr: _wizard(scr, args))
except tui.WizardCancelled:
return 1
+ try:
+ curses.curs_set(1) # restore the text cursor hidden by the TUI
+ except curses.error:
+ pass
if settings is None:
return 1
return _execute(settings)
@@ -258,21 +263,27 @@ def _collect_from_flags(args: argparse.Namespace,
print(f"[WARNING] {warning}")
if args.all:
keys = [entry.key for entry in sg_catalog.ENTRIES]
- elif args.models:
- keys = []
- for part in args.models.split(","):
+ else:
+ # Positional keys and --models both feed the same list (the
+ # positional form is the --models shorthand's space-separated
+ # twin); duplicates collapse, unknown keys stop the run with the
+ # known set.
+ keys: List[str] = []
+ for part in list(args.models_pos or []) + \
+ (args.models or "").split(","):
key = part.strip()
if not key:
continue
if sg_catalog.entry_by_key(key) is None:
known = ", ".join(e.key for e in sg_catalog.ENTRIES)
parser.error(f"unknown model key {key!r} (known: {known})")
- keys.append(key)
- else:
- keys = []
- print("[INFO] No --models given: installing the package only "
- "(use --models KEY[,KEY...] or --all to add models, or the "
- "TUI's Configure screen).")
+ if key not in keys:
+ keys.append(key)
+ if not keys:
+ print("[INFO] No models given: installing the package only "
+ "(pass model keys — positional or --models KEY[,KEY…] — "
+ "or --all to add models, or use the TUI's Configure "
+ "screen).")
return {
"keys": keys,
"do_python": not args.skip_python,
@@ -357,33 +368,20 @@ def models_screen(stdscr) -> int:
def uninstall(*, emit=None, cancel=None) -> int:
"""Remove the SGLang-Omni backend entirely.
- Phases: stop the managed server, pip-uninstall sglang-omni and every
- catalog model's companion packages, delete every cached weight
- snapshot, then remove the tool-owned venv (app/envs/sglomni — the
- heavyweight CUDA stack is the install, so unlike the lighter backends
- the whole environment goes) and the uv-managed interpreters under
- app/envs/pythons. CANCEL is honored between phases only. Returns the
- exit code (130 when cancelled before a remaining phase).
+ Phases: stop the managed server, delete every catalog model's cached
+ weight snapshot, then remove the tool-owned venv (app/envs/sglomni —
+ the heavyweight CUDA stack is the install, so unlike the lighter
+ backends the whole environment goes) and the uv-managed interpreters
+ under app/envs/pythons. No pip-uninstall phase: the venv removal IS
+ the cleanup, and pip-ing the package plus every companion out of an
+ environment that is about to be deleted is minutes of pure wait time.
+ CANCEL is honored between phases only. Returns the exit code (130
+ when cancelled before a remaining phase).
"""
if servers.pid_for(SERVER_NAME) is not None:
servers.stop(SERVER_NAME)
if common.cancel_requested(cancel):
return 130
- packages = [SGLOMNI_PIP_PKG]
- for entry in sg_catalog.ENTRIES:
- for spec, _no_deps in entry.extras:
- name = spec.split("=")[0].split("<")[0].split(">")[0].strip()
- if name and name not in packages:
- packages.append(name)
- if envs.env_exists(SGLOMNI_ENV):
- rc = common.pip_uninstall(packages, emit=emit, env_dir=SGLOMNI_ENV)
- if rc != 0:
- print(f"[WARNING] pip uninstall failed (exit {rc}); the venv "
- "is removed below anyway")
- else:
- rc = 0
- if common.cancel_requested(cancel):
- return 130
sg_models.delete_model_weights()
if common.cancel_requested(cancel):
return 130
@@ -395,17 +393,23 @@ def uninstall(*, emit=None, cancel=None) -> int:
print(f"[WARNING] Could not fully remove {directory}")
else:
print(f"[OK] {directory} removed.")
- return rc
+ return 0
def update(*, emit=None, cancel=None) -> int:
"""Update the sglang-omni backend: pip install -U in its venv.
A managed server that is running is stopped first (best-effort): it
- imports the very package being upgraded. CANCEL is honored between
- phases only. Model weights are untouched (they live in the shared
- HuggingFace cache and survive package upgrades). When the venv does
- not exist there is nothing to update. Returns the exit code.
+ imports the very package being upgraded. The upgrade is followed by a
+ companion refresh — every installed model's extras re-run (a pin
+ already satisfied is a pip no-op, so this is cheap when nothing
+ drifted) — so a newer sglang-omni's companion requirements are met
+ the way a fresh install would meet them; a failing extra warns and
+ leaves the update successful (the import probe re-heals it at the
+ next model install or server start). Model weights are untouched
+ (they live in the shared HuggingFace cache and survive package
+ upgrades). When the venv does not exist there is nothing to update.
+ CANCEL is honored between phases only. Returns the exit code.
"""
if servers.pid_for(SERVER_NAME) is not None:
servers.stop(SERVER_NAME)
@@ -420,9 +424,16 @@ def update(*, emit=None, cancel=None) -> int:
if rc != 0:
print(f"[WARNING] pip install -U failed (exit {rc}); update "
f"{SGLOMNI_PIP_PKG} manually")
- else:
- print(f"[OK] {SGLOMNI_PIP_PKG} is up to date (or just upgraded).")
- return rc
+ return rc
+ print(f"[OK] {SGLOMNI_PIP_PKG} is up to date (or just upgraded).")
+ for entry in sg_models.installed_entries():
+ crc = sg_models.install_companions(entry, emit=emit, cancel=cancel,
+ force=True)
+ if crc != 0:
+ print(f"[WARNING] {entry.label}'s companion packages could "
+ "not all be refreshed; the next install or server start "
+ "retries what the import probe finds missing.")
+ return 0
def main() -> int:
diff --git a/app/converter/clients/sglomni.py b/app/converter/clients/sglomni.py
index 14c9990..6b8493b 100644
--- a/app/converter/clients/sglomni.py
+++ b/app/converter/clients/sglomni.py
@@ -20,7 +20,8 @@ catalog``):
Clone-capable models without a reference synthesize their built-in
default voice ("default") unless the catalog marks a reference as
-mandatory (Qwen3-TTS Base, dots.tts, ZONOS2 — those refuse at connect).
+mandatory (Qwen3-TTS Base, MOSS-TTS, dots.tts, ZONOS2 — those refuse at
+connect).
"""
import base64
@@ -59,12 +60,6 @@ _MIME_BY_SUFFIX = {
".webm": "audio/webm", ".mp4": "audio/mp4",
}
-# Error-envelope types the server returns for deterministic request
-# problems (bad voice, missing reference, unknown model): the identical
-# request fails on every retry, so the chunk loop gives up immediately.
-_NON_RETRYABLE_TYPES = ("BadRequestError", "InvalidRequestError",
- "NotFoundError", "PermissionDeniedError")
-
# The scheduler's KV-window admission error ("Request requires more tokens
# than the thinker KV cache can hold (input_tokens=684, max_new_tokens=
# 12288, required_tokens=12972, kv_capacity=4095)..."): the server names
@@ -130,8 +125,6 @@ class SgOmniTTSClient(BaseTTSClient):
# The catalog entry this run targets (the backend package validates
# the key; only its repo id and capability are client business).
from backends.sglomni.catalog import entry_by_key
- from backends.sglomni.constants import DEFAULT_PORT, SERVER_NAME
- from backends.common import port_of
self.entry = entry_by_key((model or "").strip())
if self.entry is None:
raise RuntimeError(
@@ -139,7 +132,6 @@ class SgOmniTTSClient(BaseTTSClient):
"(see Configure Backends → SGLang-Omni or the backend docs).")
self.api_url = ((api_url or config.SGLOMNI_API_URL).strip()
.rstrip("/"))
- self.port = port_of(self.api_url, DEFAULT_PORT)
self.voice = (voice or "").strip() or None
self.ref_audio = (ref_audio or "").strip() or None
self.ref_text = (ref_text or "").strip()
@@ -152,13 +144,18 @@ class SgOmniTTSClient(BaseTTSClient):
# rejection (None = none learned): later requests keep their
# max_new_tokens under it. See _kv_admission_fit.
self._kv_fit = None
+ # The ref_audio request value, computed on first use (see
+ # _ref_audio_value); None = not computed yet.
+ self._ref_audio_cached = None
# Seed sent with every request: config.SEED as-is, or (with
# CONSTANT_SEED and SEED < 0) one random value drawn per run and
# reused for every chunk so the voice stays consistent across
# chunk boundaries. Only sent to models that accept a
# request-scoped seed (Voxtral rejects it outright), and only
# when a concrete seed is in play (a negative one means "re-sample
- # every generation", so there is nothing to send).
+ # every generation", so there is nothing to send). NOTE(unverified
+ # upstream): whether the other pipelines accept a seed too — see
+ # the catalog's supports_seed note.
seed = resolve_request_seed() if self.entry.supports_seed else None
self._seed = seed if (seed is not None and seed >= 0) else None
if language is None:
@@ -188,15 +185,27 @@ class SgOmniTTSClient(BaseTTSClient):
self._report(f"[WARNING] --clone is ignored with {entry.label}: "
"it voices text with its built-in presets.")
self.ref_audio = None
+ elif entry.capability == "design" and self.ref_audio:
+ self._report(f"[WARNING] --clone is ignored with {entry.label}: "
+ "it designs the voice from instructions.")
+ self.ref_audio = None
elif self.ref_audio and not Path(self.ref_audio).is_file():
raise RuntimeError(
f"Reference audio not found: {self.ref_audio}")
- if entry.speakers and self.voice \
- and self.voice not in entry.speakers:
+ presets = self._preset_voices()
+ if presets and self.voice and self.voice not in presets:
self._report(
f"[WARNING] Voice {self.voice!r} is not one of "
- f"{entry.label}'s presets ({', '.join(entry.speakers)}); "
+ f"{entry.label}'s presets ({', '.join(presets)}); "
"the server will reject it if it does not know the name.")
+
+ def _preset_voices(self) -> List[str]:
+ """The preset voice names ENTRY can speak with — the same list the
+ hub's voice menu offers (catalog-declared speakers, or the
+ checkpoint's own voice_embedding presets, e.g. Voxtral's)."""
+ from backends.sglomni.models import preset_voices
+ return preset_voices(self.entry)
+
def _connect(self) -> None:
"""Verify the server is up, healthy, and hosting the expected model.
@@ -209,6 +218,24 @@ class SgOmniTTSClient(BaseTTSClient):
entry, url = self.entry, self.api_url
try:
payload = self._fetch_json("/health", timeout=10)
+ except urllib.error.HTTPError as exc:
+ # A booting server answers 503 with an "unhealthy" body —
+ # urlopen turns that into an HTTPError before the healthy
+ # check below can see it. Tell the user to wait for the
+ # server that is already starting, not to start another.
+ detail = _http_error_detail(exc)
+ if exc.code == 503:
+ raise RuntimeError(
+ f"The SGLang-Omni server at {url} is not healthy yet "
+ f"(HTTP 503: {detail[:300] or 'no body'}). Wait for it "
+ "to finish booting and retry.") from exc
+ raise RuntimeError(
+ f"The SGLang-Omni server at {url} answered HTTP {exc.code} "
+ f"on /health ({detail[:300] or 'no body'}). Is this an "
+ "sgl-omni server? Start the sgl-omni server first (the "
+ "CLI and the hub start the managed instance automatically "
+ "when the backend is installed), or point --api-url at a "
+ "running server.") from exc
except Exception as exc:
raise RuntimeError(
f"SGLang-Omni server not reachable at {url}: {exc}. Start "
@@ -253,7 +280,10 @@ class SgOmniTTSClient(BaseTTSClient):
local Whisper transcription."""
if self.entry.capability != "clone" or not self.ref_audio:
return
- if not self.ref_text and not self.skip_transcription:
+ if not self.ref_text and self.skip_transcription:
+ self._report("[INFO] Skipping reference audio transcription "
+ "(--no-transcription).")
+ elif not self.ref_text:
self._report("[INFO] Transcribing reference audio for voice "
"cloning...")
from .transcribe import transcribe_reference_audio
@@ -291,23 +321,40 @@ class SgOmniTTSClient(BaseTTSClient):
def _ref_audio_value(self) -> str:
"""The ref_audio request value: a local path on a loopback server
- (the server reads the file directly), else a base64 data URL."""
- path = Path(self.ref_audio)
- if not path.is_file():
- raise RuntimeError(
- f"Reference audio not found: {self.ref_audio}")
- if _is_loopback(self.api_url):
- return str(path.resolve())
- return _data_url(path)
+ (the server reads the file directly), else a base64 data URL.
+
+ Computed once per run and cached: the clip is validated at connect
+ and cannot change mid-run, and re-encoding its bytes for every
+ sub-request would ship the same payload over and over."""
+ cached = self._ref_audio_cached
+ if cached is None:
+ path = Path(self.ref_audio)
+ if not path.is_file():
+ raise RuntimeError(
+ f"Reference audio not found: {self.ref_audio}")
+ if _is_loopback(self.api_url):
+ cached = str(path.resolve())
+ else:
+ cached = _data_url(path)
+ self._ref_audio_cached = cached
+ return cached
def _request_payload(self, text: str) -> dict:
"""The /v1/audio/speech JSON body for one sub-chunk."""
entry = self.entry
payload = {
"model": entry.repo,
+ # NOTE(unverified upstream): "voice" is sent even when nothing
+ # was picked (the "default" sentinel) and to design runs,
+ # which have no voice — audio.cpp omits the field there.
+ # Verify the server tolerates it for every pipeline.
"voice": self.voice or DEFAULT_VOICE,
"input": text,
"response_format": RESPONSE_FORMAT,
+ # NOTE(unverified upstream): Qwen-style display names
+ # ("English", "Auto") go to every model; audio.cpp maps per
+ # family. Verify each pipeline accepts them (or wants ISO
+ # codes / the field omitted).
"language": self.language,
}
if self._seed is not None:
@@ -352,9 +399,6 @@ class SgOmniTTSClient(BaseTTSClient):
exc = retry_exc
detail = _http_error_detail(exc)
raise self._request_error(exc.code, detail) from exc
- except urllib.error.URLError as exc:
- raise RuntimeError(
- f"SGLang-Omni request failed: {exc.reason}") from exc
def _post_speech(self, payload: dict) -> bytes:
"""POST PAYLOAD to /v1/audio/speech; HTTPErrors propagate raw."""
@@ -373,8 +417,12 @@ class SgOmniTTSClient(BaseTTSClient):
except urllib.error.URLError as exc:
raise RuntimeError(
f"SGLang-Omni request failed: {exc.reason}") from exc
- if not wav:
- raise RuntimeError("SGLang-Omni server returned empty audio")
+ if len(wav) < 12 or wav[:4] != b"RIFF" or wav[8:12] != b"WAVE":
+ # A JSON error body handed back with HTTP 200 would otherwise
+ # be written as chunk bytes and fail later, confusingly, in
+ # the concat step.
+ raise RuntimeError(
+ "SGLang-Omni server returned audio that is not a WAV file")
return wav
def _kv_admission_fit(self, detail: str) -> Optional[int]:
@@ -417,17 +465,17 @@ class SgOmniTTSClient(BaseTTSClient):
the server's message; anything else stays retryable.
"""
message = detail[:500] or f"HTTP {status}"
- kind = None
try:
envelope = json.loads(detail)
error = envelope.get("error")
if isinstance(error, dict):
message = str(error.get("message") or message)
- kind = error.get("type")
except ValueError:
pass
- if 400 <= status < 500 and (kind is None
- or kind in _NON_RETRYABLE_TYPES):
+ # Every 4xx envelope is deterministic — the identical request
+ # fails identically on every attempt (this is a single-user local
+ # server: it queues work rather than answering 429-style limits).
+ if 400 <= status < 500:
return NonRetryableTTSError(
f"SGLang-Omni rejected the request (HTTP {status}): "
f"{message}")
@@ -455,31 +503,36 @@ class SgOmniTTSClient(BaseTTSClient):
if not sub_chunks:
raise RuntimeError("No text to synthesize")
- with self._chunk_heartbeat(chunk_num):
- wav_parts: List[bytes] = [
- self._request_wav(sub_text) for sub_text in sub_chunks]
-
output_path = self._chunk_path(chunk_num, ".wav")
- if len(wav_parts) == 1:
- output_path.write_bytes(wav_parts[0])
- else:
- # Several sub-request WAVs: concatenate through the shared
- # ffmpeg path (each part is a complete file with headers).
- with tempfile.TemporaryDirectory(
- prefix="sglomni_parts_") as parts_dir:
- part_paths: List[Path] = []
- for index, wav in enumerate(wav_parts, 1):
+ with tempfile.TemporaryDirectory(
+ prefix="sglomni_parts_") as parts_dir:
+ # One part per sub-request, spooled to disk as it arrives
+ # (like the other clients) instead of buffering every
+ # response in memory until the chunk is complete.
+ part_paths: List[Path] = []
+ with self._chunk_heartbeat(chunk_num):
+ for index, sub_text in enumerate(sub_chunks, 1):
part = Path(parts_dir) / f"part_{index:02d}.wav"
- part.write_bytes(wav)
+ part.write_bytes(self._request_wav(sub_text))
part_paths.append(part)
+ if len(part_paths) == 1:
+ output_path.write_bytes(part_paths[0].read_bytes())
+ else:
+ # Several sub-request WAVs: concatenate through the
+ # shared ffmpeg path (each part is a complete file
+ # with headers).
concat_audio_files(part_paths, output_path)
logger.debug("Chunk %d generated (%d sub-request(s))",
- chunk_num, len(wav_parts))
+ chunk_num, len(part_paths))
return str(output_path)
except ConversionCancelled:
raise
+ except NonRetryableTTSError:
+ # Propagate past the generic handler so the retry loop skips
+ # its remaining attempts for deterministic server errors.
+ raise
except Exception as exc:
logger.error("SGLang-Omni chunk processing failed for chunk "
"%d: %s", chunk_num, exc)
diff --git a/app/tests/test_backends_sglomni.py b/app/tests/test_backends_sglomni.py
index 54659d6..d4ec12d 100644
--- a/app/tests/test_backends_sglomni.py
+++ b/app/tests/test_backends_sglomni.py
@@ -2,6 +2,7 @@
import io
import json
+import shutil
import urllib.error
import sys
import tempfile
@@ -105,7 +106,8 @@ class ModelInstallStateTests(unittest.TestCase):
def setUp(self):
self._tmp = tempfile.TemporaryDirectory()
self.cache = Path(self._tmp.name)
- patcher = patch.object(models, "_hf_cache_dir", return_value=self.cache)
+ patcher = patch("backends.common.hf_cache_dir",
+ return_value=self.cache)
patcher.start()
self.addCleanup(patcher.stop)
self.addCleanup(self._tmp.cleanup)
@@ -850,8 +852,8 @@ class ModelsScreenTests(unittest.TestCase):
rc, _trees, _confirms, _flashes, _runs = self._screen(
[[], wizard._GO_BACK], installed=(entry.key,),
extra=[
- patch.object(models, "_hf_cache_dir",
- return_value=Path(td)),
+ patch("backends.common.hf_cache_dir",
+ return_value=Path(td)),
patch.object(models, "_managed_running_repo",
return_value=None),
])
@@ -1028,5 +1030,288 @@ class SetupWizardTests(unittest.TestCase):
execute.assert_called_once_with(settings)
+class SystemDepTests(unittest.TestCase):
+ """Docs-required system binaries are checked at install time."""
+
+ def test_dac_models_require_ffmpeg(self):
+ # The Fish Audio and ZONOS2 pipelines shell out to ffmpeg; the
+ # install flow must warn like it does for the Qwen entries' sox.
+ for key in ("fish_s2_pro", "zonos2"):
+ with self.subTest(key=key):
+ entry = entry_by_key(key)
+ self.assertEqual(entry.system_dep, "ffmpeg")
+ self.assertIn("ffmpeg", entry.system_hint)
+
+ def test_missing_system_dep_warns(self):
+ with patch("shutil.which", return_value=None):
+ message = models.system_dep_missing(entry_by_key("fish_s2_pro"))
+ self.assertIsNotNone(message)
+ self.assertIn("ffmpeg", message)
+
+ def test_present_system_dep_is_silent(self):
+ with patch("shutil.which", return_value="/usr/bin/ffmpeg"):
+ self.assertIsNone(
+ models.system_dep_missing(entry_by_key("fish_s2_pro")))
+
+ def test_missing_system_dep_warns_during_install(self):
+ out = io.StringIO()
+ with patch("shutil.which", return_value=None), \
+ patch("backends.sglomni.models.prepare_env", return_value=0), \
+ patch("backends.common.pip_install", return_value=0), \
+ patch("backends.sglomni.models._hf_download_prefix",
+ return_value=["hf"]), \
+ patch("backends.common.run_console_subprocess",
+ return_value=0), \
+ redirect_stdout(out):
+ rc = models.install_model("fish_s2_pro")
+ self.assertEqual(rc, 0)
+ self.assertIn("ffmpeg", out.getvalue())
+
+
+class UninstallModelServerStopTests(unittest.TestCase):
+ """uninstall_model stops a managed server hosting the model."""
+
+ def test_stops_a_server_hosting_the_model(self):
+ entry = ENTRIES[0]
+ with patch.object(models, "_managed_running_repo",
+ return_value=entry.repo), \
+ patch.object(models, "delete_model_weights"), \
+ patch.object(servers, "stop") as stop:
+ rc = models.uninstall_model(entry.key)
+ self.assertEqual(rc, 0)
+ stop.assert_called_once_with(constants.SERVER_NAME)
+
+ def test_leaves_a_server_hosting_something_else_alone(self):
+ entry = ENTRIES[0]
+ with patch.object(models, "_managed_running_repo",
+ return_value=ENTRIES[1].repo), \
+ patch.object(servers, "stop") as stop, \
+ patch.object(models, "delete_model_weights") as weights:
+ rc = models.uninstall_model(entry.key)
+ self.assertEqual(rc, 0)
+ stop.assert_not_called()
+ weights.assert_called_once_with([entry])
+
+
+class WizardUninstallTests(unittest.TestCase):
+ """wizard.uninstall: stop the server, delete weights, remove the venv.
+
+ The venv removal IS the cleanup (the heavyweight CUDA stack is the
+ install): there is deliberately no pip-uninstall phase ahead of it.
+ """
+
+ def _run(self, *, pid=None, cancel_after=None, venv=True, pythons=True,
+ rmtree_removes=True):
+ """Run uninstall with a temp venv tree; return (rc, output,
+ stop_calls, weights, pip_uninstall).
+
+ CANCEL_AFTER N lets the first N cancel checks pass (None = none
+ of them do).
+ """
+ from backends.sglomni import wizard
+ out = io.StringIO()
+ with tempfile.TemporaryDirectory() as td:
+ venv_dir = Path(td) / "sglomni"
+ pythons_dir = Path(td) / "pythons"
+ for make, directory in ((venv, venv_dir),
+ (pythons, pythons_dir)):
+ if make:
+ directory.mkdir()
+ state = {"stops": [], "cancel": 0}
+
+ def fake_stop(name):
+ state["stops"].append(name)
+
+ def fake_cancel(_cancel):
+ state["cancel"] += 1
+ return cancel_after is not None \
+ and state["cancel"] > cancel_after
+
+ real_rmtree = shutil.rmtree
+
+ def fake_rmtree(path, ignore_errors=False):
+ if rmtree_removes:
+ real_rmtree(path, ignore_errors=True)
+
+ with patch.object(servers, "pid_for", return_value=pid), \
+ patch.object(servers, "stop", side_effect=fake_stop), \
+ patch.object(models, "delete_model_weights") as weights, \
+ patch.object(common, "pip_uninstall") as pip_uninstall, \
+ patch.object(wizard, "SGLOMNI_ENV", venv_dir), \
+ patch.object(envs, "PYTHON_INSTALL_DIR", pythons_dir), \
+ patch.object(common, "cancel_requested",
+ side_effect=fake_cancel), \
+ patch("shutil.rmtree", side_effect=fake_rmtree), \
+ redirect_stdout(out):
+ rc = wizard.uninstall()
+ existed = venv_dir.exists() or pythons_dir.exists()
+ return rc, out.getvalue(), state["stops"], weights, \
+ pip_uninstall, existed
+
+ def test_removes_the_venv_tree_without_a_pip_phase(self):
+ rc, out, _stops, weights, pip_uninstall, existed = self._run()
+ self.assertEqual(rc, 0)
+ weights.assert_called_once_with()
+ pip_uninstall.assert_not_called()
+ self.assertFalse(existed)
+
+ def test_stops_only_a_running_managed_server(self):
+ _rc, _out, stops, _w, _p, _e = self._run(pid=123)
+ self.assertEqual(stops, ["sglomni"])
+ _rc, _out, stops, _w, _p, _e = self._run(pid=None)
+ self.assertEqual(stops, [])
+
+ def test_cancel_before_anything_removes_nothing(self):
+ rc, _out, _stops, weights, _p, _e = self._run(cancel_after=0)
+ self.assertEqual(rc, 130)
+ weights.assert_not_called()
+
+ def test_cancel_before_the_venv_removal_keeps_the_dirs(self):
+ rc, _out, _stops, weights, _p, existed = self._run(cancel_after=1)
+ self.assertEqual(rc, 130)
+ weights.assert_called_once_with()
+ self.assertTrue(existed)
+
+ def test_a_stuck_venv_warns_but_the_uninstall_succeeds(self):
+ rc, out, _stops, _w, _p, existed = self._run(rmtree_removes=False)
+ self.assertEqual(rc, 0)
+ self.assertTrue(existed)
+ self.assertIn("Could not fully remove", out)
+
+ def test_missing_venv_dirs_are_fine(self):
+ rc, _out, _stops, _w, _p, _e = self._run(venv=False, pythons=False)
+ self.assertEqual(rc, 0)
+
+
+class WizardUpdateTests(unittest.TestCase):
+ """wizard.update: package upgrade, then a companion refresh.
+
+ The refresh re-runs every installed model's extras (a satisfied pin
+ is a pip no-op), healing version drift the import probe cannot see.
+ """
+
+ def _run(self, *, pid=None, venv=True, pip_rc=0, companions_rc=0,
+ installed=(0, 1), cancel_after=None):
+ from backends.sglomni import wizard
+ out = io.StringIO()
+ entries = [ENTRIES[i] for i in installed]
+ state = {"pip": [], "companions": [], "stops": [], "cancel": 0}
+
+ def fake_stop(name):
+ state["stops"].append(name)
+
+ def fake_cancel(_cancel):
+ state["cancel"] += 1
+ return cancel_after is not None \
+ and state["cancel"] > cancel_after
+
+ def fake_pip(specs, **kwargs):
+ state["pip"].append((list(specs), kwargs))
+ return pip_rc
+
+ def fake_companions(entry, **kwargs):
+ state["companions"].append((entry.key, kwargs.get("force")))
+ return companions_rc
+
+ with patch.object(servers, "pid_for", return_value=pid), \
+ patch.object(servers, "stop", side_effect=fake_stop), \
+ patch.object(envs, "env_exists", return_value=venv), \
+ patch.object(common, "pip_install", side_effect=fake_pip), \
+ patch.object(models, "installed_entries",
+ return_value=entries), \
+ patch.object(models, "install_companions",
+ side_effect=fake_companions), \
+ patch.object(common, "cancel_requested",
+ side_effect=fake_cancel), \
+ redirect_stdout(out):
+ rc = wizard.update()
+ return rc, out.getvalue(), state
+
+ def test_no_venv_is_a_no_op(self):
+ rc, _out, state = self._run(venv=False)
+ self.assertEqual(rc, 0)
+ self.assertEqual(state["pip"], [])
+ self.assertEqual(state["companions"], [])
+
+ def test_stops_the_server_then_upgrades_the_package(self):
+ rc, _out, state = self._run(pid=123)
+ self.assertEqual(rc, 0)
+ self.assertEqual(state["stops"], ["sglomni"])
+ specs, kwargs = state["pip"][0]
+ self.assertEqual(specs, ["sglang-omni"])
+ self.assertTrue(kwargs["upgrade"])
+ self.assertEqual(kwargs["extra_args"], ["--pre"])
+
+ def test_upgrade_failure_skips_the_companion_refresh(self):
+ rc, _out, state = self._run(pip_rc=23)
+ self.assertEqual(rc, 23)
+ self.assertEqual(state["companions"], [])
+
+ def test_refreshes_every_installed_models_companions(self):
+ rc, _out, state = self._run(installed=(0, 4))
+ self.assertEqual(rc, 0)
+ self.assertEqual(state["companions"],
+ [(ENTRIES[0].key, True), (ENTRIES[4].key, True)])
+
+ def test_companion_failure_warns_but_the_update_succeeds(self):
+ rc, out, _state = self._run(companions_rc=23)
+ self.assertEqual(rc, 0)
+ self.assertIn("companion packages", out)
+
+ def test_cancel_before_pip_removes_nothing(self):
+ rc, _out, state = self._run(cancel_after=0)
+ self.assertEqual(rc, 130)
+ self.assertEqual(state["pip"], [])
+
+
+class NonInteractiveCliTests(unittest.TestCase):
+ """_collect_from_flags: the flag-driven (non-TUI) setup path.
+
+ Positional keys are the space-separated twin of --models: both feed
+ the same deduplicated install list.
+ """
+
+ def _collect(self, argv, *, installed=True):
+ from backends.sglomni import wizard
+ parser = wizard.build_parser()
+ args = parser.parse_args(argv)
+ with patch.object(wizard, "_preflight", return_value=[]), \
+ patch.object(wizard, "_gpu_warning", return_value=None), \
+ patch.object(wizard, "_is_installed",
+ return_value=installed):
+ return wizard._collect_from_flags(args, parser)
+
+ def test_positional_keys_are_installed(self):
+ settings = self._collect(["higgs_audio_v3_tts"])
+ self.assertEqual(settings["keys"], ["higgs_audio_v3_tts"])
+ self.assertTrue(settings["do_python"])
+ self.assertFalse(settings["do_install"])
+
+ def test_positional_and_models_flags_merge_in_order(self):
+ settings = self._collect(["moss_tts", "--models",
+ "higgs_audio_v3_tts,moss_tts"])
+ self.assertEqual(settings["keys"],
+ ["moss_tts", "higgs_audio_v3_tts"])
+
+ def test_all_installs_every_catalog_model(self):
+ settings = self._collect(["--all"])
+ self.assertEqual(settings["keys"], [entry.key for entry in ENTRIES])
+
+ def test_unknown_key_stops_the_run(self):
+ with self.assertRaises(SystemExit):
+ self._collect(["nope"])
+
+ def test_no_keys_installs_the_package_only(self):
+ settings = self._collect([], installed=False)
+ self.assertEqual(settings["keys"], [])
+ self.assertTrue(settings["do_install"])
+
+ def test_skip_flags_are_honored(self):
+ settings = self._collect(["--skip-python", "--skip-install",
+ "higgs_audio_v3_tts"])
+ self.assertFalse(settings["do_python"])
+ self.assertFalse(settings["do_install"])
+
+
if __name__ == "__main__":
unittest.main()
diff --git a/app/tests/test_tts_sglomni.py b/app/tests/test_tts_sglomni.py
index 0e3a7de..c4cdfaf 100644
--- a/app/tests/test_tts_sglomni.py
+++ b/app/tests/test_tts_sglomni.py
@@ -146,6 +146,29 @@ class ConnectHealthTests(unittest.TestCase):
self.assertIn("not reachable", str(ctx.exception))
self.assertIn("sgl-omni", str(ctx.exception))
+ def test_booting_503_tells_the_user_to_wait(self):
+ # A booting sgl-omni answers /health with 503 + an "unhealthy"
+ # body (urlopen surfaces that as an HTTPError before any JSON
+ # could be inspected) — the message must say wait, not start.
+ with patch("converter.clients.sglomni.urllib.request.urlopen",
+ side_effect=_http_error(503, '{"status": "unhealthy"}')):
+ with self.assertRaises(RuntimeError) as ctx:
+ SgOmniTTSClient(_DUMMY_CHUNKS, model="higgs_audio_v3_tts")
+ message = str(ctx.exception)
+ self.assertIn("not healthy yet", message)
+ self.assertIn("HTTP 503", message)
+ self.assertIn("booting", message)
+ self.assertNotIn("not reachable", message)
+
+ def test_other_health_errors_name_the_code(self):
+ with patch("converter.clients.sglomni.urllib.request.urlopen",
+ side_effect=_http_error(404, "<html>nope</html>")):
+ with self.assertRaises(RuntimeError) as ctx:
+ SgOmniTTSClient(_DUMMY_CHUNKS, model="higgs_audio_v3_tts")
+ message = str(ctx.exception)
+ self.assertIn("HTTP 404", message)
+ self.assertIn("Is this an sgl-omni server?", message)
+
def test_booting_server_raises(self):
with self.assertRaises(RuntimeError) as ctx:
self._connect([{"status": "unhealthy"}])
@@ -192,6 +215,7 @@ class PayloadTests(unittest.TestCase):
client.language = "English"
client._seed = None
client._kv_fit = None
+ client._ref_audio_cached = None
return client
def test_speaker_payload_sends_the_preset_name(self):
@@ -237,6 +261,20 @@ class PayloadTests(unittest.TestCase):
b"abc")
self.assertNotIn("ref_text", payload)
+ def test_reference_audio_is_encoded_once_per_run(self):
+ # The clip cannot change mid-run: the data URL (or resolved path)
+ # is computed on the first sub-request and reused verbatim.
+ from converter.clients.sglomni import _data_url as real_data_url
+ client = self._make_client("higgs_audio_v3_tts",
+ ref_audio=str(self.ref))
+ client.api_url = "http://10.20.30.40:8100"
+ with patch("converter.clients.sglomni._data_url",
+ wraps=real_data_url) as encode:
+ first = client._request_payload("Hello.")
+ second = client._request_payload("Hello again.")
+ self.assertEqual(encode.call_count, 1)
+ self.assertEqual(first["ref_audio"], second["ref_audio"])
+
def test_clone_without_reference_sends_no_reference_fields(self):
client = self._make_client("higgs_audio_v3_tts")
payload = client._request_payload("Hello.")
@@ -423,6 +461,33 @@ class KvAdmissionTests(unittest.TestCase):
self.assertEqual(second["max_new_tokens"], 2531)
+class ErrorClassificationTests(unittest.TestCase):
+ """HTTP status → retry decision: every 4xx envelope is deterministic."""
+
+ def _request_error(self, status, detail):
+ client = SgOmniTTSClient.__new__(SgOmniTTSClient)
+ return client._request_error(status, detail)
+
+ def test_every_4xx_envelope_is_non_retryable(self):
+ # Including types outside the OpenAI-style names: the identical
+ # request fails identically on every attempt.
+ exception = self._request_error(
+ 401, json.dumps({"error": {"message": "bad key",
+ "type": "AuthenticationError"}}))
+ self.assertIsInstance(exception, NonRetryableTTSError)
+ self.assertIn("bad key", str(exception))
+
+ def test_non_json_4xx_bodies_are_non_retryable(self):
+ exception = self._request_error(400, "plain text refusal")
+ self.assertIsInstance(exception, NonRetryableTTSError)
+ self.assertIn("plain text refusal", str(exception))
+
+ def test_5xx_stays_retryable(self):
+ exception = self._request_error(500, "CUDA out of memory")
+ self.assertNotIsInstance(exception, NonRetryableTTSError)
+ self.assertIn("CUDA out of memory", str(exception))
+
+
class GenerateChunkTests(unittest.TestCase):
"""Chunk generation: WAV output, sub-chunking, bookkeeping."""
@@ -511,6 +576,40 @@ class GenerateChunkTests(unittest.TestCase):
self.assertIsNone(client.generate_chunk("Hello.", 1))
self.assertEqual(mock_wav.call_count, 1)
+ def test_non_retryable_errors_propagate(self):
+ # Deterministic server errors must reach the retry loop directly
+ # (which skips its remaining attempts and re-raises with the
+ # actionable message), not come back as a generic failed attempt.
+ client = self._make_client()
+ with patch.object(client, "_request_wav",
+ side_effect=NonRetryableTTSError(
+ "unknown voice")):
+ with self.assertRaises(NonRetryableTTSError):
+ client.generate_chunk("Hello.", 1)
+
+ def test_retry_loop_skips_remaining_attempts(self):
+ client = self._make_client()
+ with patch.object(client, "_request_wav",
+ side_effect=NonRetryableTTSError(
+ "unknown voice")) as mock_wav:
+ with self.assertRaises(NonRetryableTTSError):
+ client.process_chunk_with_retry(1, "Hello.")
+ self.assertEqual(mock_wav.call_count, 1)
+
+ def test_a_non_wav_200_body_fails_the_request(self):
+ # A JSON error body served with HTTP 200 must not be written as
+ # chunk bytes (it would only fail later, confusingly, in the
+ # concat step).
+ client = self._make_client()
+ response = MagicMock()
+ response.__enter__.return_value = response
+ response.read.return_value = b'{"error": {"message": "nope"}}'
+ with patch("converter.clients.sglomni.urllib.request.urlopen",
+ return_value=response):
+ with self.assertRaises(RuntimeError) as ctx:
+ client._request_wav("Hello.")
+ self.assertIn("not a WAV file", str(ctx.exception))
+
def test_stale_chunk_files_are_removed(self):
stale = Path(self._tmp.name) / "chunk_0001.mp3"
stale.write_bytes(b"old")
generated by cgit v1.2.3 (git 2.25.1) at 2026-09-08 09:43:54 +0000