make the isolated speech engines actually start, and list their voices

XTTS-v2 and Parler were advertised in /v1/models and failed on every request.
Three separate causes, none of them the one the report assumed.

A venv built by a DIFFERENT interpreter bricks the engine permanently. A
worker venv keeps bin/python as a symlink to the python that created it, so
one built on the host and used from the container (or built before the image
moved its interpreter) leaves a dangling link. The failure is not a clean
"missing": Path.exists() follows symlinks and says False, so the code decides
the venv is absent and runs `python -m venv` over the top, which trips on that
same dangling bin/python3:

    Error: [Errno 2] No such file or directory: '.../parler_venv/bin/python3'

reported for weeks as "the venv can't be created", with a fully populated venv
and 7 GB of packages sitting right there. Both workers now check that the
interpreter RUNS, not that the path exists, and clear a foreign venv before
recreating.

TRANSFORMERS_CACHE pointed at a second cache. It is deprecated but transformers
4.x still honours it, and honours it as the cache_dir handed to
hf_hub_download -- a hub-layout cache, not a different kind of store. Setting
it to <dir>/transformers therefore gave the machine TWO hub caches: measured,
<dir>/hub held 122 models and <dir>/transformers held 2. A worker in an
isolated venv read only the second, found nothing, and -- forced offline so a
missing model fails fast -- reported "We couldn't connect to
https://huggingface.co" for a model that was present the whole time. It is
also why workers re-downloaded models the server already had. All four
variables now name the hub directory.

XTTS never built its venv at all. parler bootstraps on first use; XTTS only
looked for a venv and raised a 501 telling the operator to go and make one.
Same situation, two behaviours. It bootstraps now, with
CODERAI_XTTS_AUTOINSTALL=0 for a host that manages its own venvs.

Also the voice listing the report asks for, but NOT at /v1/audio/speakers:
that path is already enrolled speaker voiceprints for recognition, and serving
TTS voices there would conflate two unrelated things. The voices join the other
"what can this model do" answers at /v1/audio/speech/capabilities, which
already takes the same model= parameter.

It never loads a model to answer -- an already-loaded backend is asked
directly, otherwise the list is whatever was remembered the last time that
model loaded, and voices_source says which. And it does not invent metadata:
Kokoro ids genuinely encode accent and gender (af_sarah -> en-us, female) so
those are derived, while an XTTS speaker name carries neither and reports
None. A wrong gender against a voice is worse than no gender.

voice_selection is reported too, because parler has no speaker names at all --
it is steered by a free-text description, and a client reading only an empty
voices list would conclude it cannot be directed.
Co-Authored-By: 's avatarClaude Opus 5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_014T3rZozi1Ehk96NG1WesgC
parent 1f42d08c
......@@ -44,6 +44,30 @@ def _venv_python() -> Path:
"python.exe" if os.name == "nt" else "python")
def _venv_runs(py: Path) -> bool:
"""Whether this venv's interpreter actually RUNS.
``Path.exists()`` is not enough and its failure mode is nasty. A venv's
``bin/python`` is a symlink to the interpreter that built it, so a venv
created by a DIFFERENT python -- on the host rather than in the container,
or before the image moved its interpreter -- leaves a dangling link.
``exists()`` follows symlinks and answers False, so the code concludes the
venv is missing and tries to create one over the top; ``python -m venv``
then trips over the same dangling ``bin/python3`` and dies:
Error: [Errno 2] No such file or directory: '.../parler_venv/bin/python3'
which surfaced as "Command '[... -m venv ...]' returned non-zero exit
status 1" on every request, for ever, with a fully populated venv sitting
right there. Measured 2026-10-28.
"""
try:
return subprocess.run([str(py), "-c", "import sys"],
capture_output=True, timeout=120).returncode == 0
except Exception:
return False
def _pip_ok(py: Path) -> bool:
try:
return subprocess.run([str(py), "-c", "import parler_tts, soundfile"],
......@@ -77,6 +101,13 @@ def _bootstrap_venv() -> Path:
import shutil
print("[parler] rebuilding venv as fully isolated …", flush=True)
shutil.rmtree(_VENV_DIR, ignore_errors=True)
# A venv built by another interpreter cannot be repaired in place -- see
# _venv_runs. Clear it out so the creation below starts from nothing.
if _VENV_DIR.exists() and not _venv_runs(_venv_python()):
import shutil
print(f"[parler] {_VENV_DIR} was built by a different interpreter "
f"(its python does not run here); rebuilding …", flush=True)
shutil.rmtree(_VENV_DIR, ignore_errors=True)
if not _venv_python().exists():
print(f"[parler] creating isolated venv at {_VENV_DIR} …", flush=True)
_VENV_DIR.parent.mkdir(parents=True, exist_ok=True)
......
......@@ -90,11 +90,32 @@ async def speech_capabilities(model: str = ""):
"""
from codai.api import tts_backends
name = (model or "").strip()
# The built-in voices belong with the other "what can this model do"
# answers rather than on a route of their own -- and NOT on
# /v1/audio/speakers, which already means something else entirely:
# enrolled speaker voiceprints for recognition.
#
# Answered without loading anything. An already-loaded model is asked
# directly; otherwise the list is whatever was remembered the last time it
# was loaded, and `voices_source` says which.
voices = {"voices": [], "voices_source": "none", "voice_selection": "builtin"}
try:
backend = None
try:
from codai.models.manager import multi_model_manager
backend = (multi_model_manager.models.get(f"tts:{name}")
or multi_model_manager.models.get(name))
except Exception:
backend = None
voices = tts_backends.family_voices(name, backend)
except Exception as exc:
print(f" [tts] could not list voices for {name!r}: {exc}", flush=True)
return {
"model": name,
"family": tts_backends._family(name),
"emotions": tts_backends.family_emotions(name),
"styles": tts_backends.family_styles(name),
**voices,
}
......
......@@ -209,6 +209,12 @@ class _CoquiBackend:
# verified TTS venv used to answer "pip install coqui-tts" at this
# exact line: the venv was there and nothing knew to use it.
from codai.api import xtts_worker
# Build the venv on first use, the way parler does. Opt out with
# CODERAI_XTTS_AUTOINSTALL=0 on a host that manages its own venvs
# or has no network.
if not xtts_worker.available() and \
os.environ.get("CODERAI_XTTS_AUTOINSTALL", "1") != "0":
xtts_worker.bootstrap()
if xtts_worker.available():
print(f"[tts] coqui-tts from {xtts_worker.venv_dir()}", flush=True)
self._tts = xtts_worker.XttsSubprocess(model_name)
......@@ -435,8 +441,155 @@ class _RemoteSpeechBackend:
pass
# --------------------------------------------------------------------------- #
# Built-in voices
# --------------------------------------------------------------------------- #
#
# Clients need to offer a voice picker BEFORE the model is loaded, and a listing
# endpoint must never load one to answer -- a cheap availability check that
# loads is how a catalogue route turns into a several-hundred-MB loader.
#
# So each backend's own `voices` property stays authoritative, and what it
# returns is remembered the first time that model is loaded. A later listing is
# served from that memory, and says which it was.
#: How a family picks a voice at all. Not every engine HAS a voice list: parler
#: is steered by a natural-language description, which is why its `voices` is
#: empty and a client that only looks at the list concludes, wrongly, that it
#: cannot be directed.
_FAMILY_VOICE_SELECTION: dict[str, str] = {
"kokoro": "builtin",
"coqui": "builtin+clone", # a named speaker, or speaker_wav to clone
"bark": "builtin",
"parler": "described", # free-text description, no speaker names
"melo": "builtin",
}
#: Kokoro encodes the accent and the gender in the id itself (``af_sarah`` =
#: American female). Derived, never guessed -- a family whose ids carry nothing
#: reports null rather than an invented gender.
_KOKORO_ACCENTS = {
"a": ("en-us", "American English"), "b": ("en-gb", "British English"),
"e": ("es", "Spanish"), "f": ("fr", "French"), "h": ("hi", "Hindi"),
"i": ("it", "Italian"), "j": ("ja", "Japanese"), "p": ("pt-br", "Portuguese"),
"z": ("zh", "Chinese"),
}
def _voice_meta(family: str, voice_id: str) -> dict:
"""What is genuinely known about one voice id.
Unknown fields are ``None``, not a plausible-looking default: a gender shown
against the wrong voice is worse than no gender at all.
"""
vid = str(voice_id or "")
out = {"id": vid, "name": vid, "gender": None, "languages": []}
if family == "kokoro" and len(vid) >= 2 and vid[1] in ("f", "m"):
acc = _KOKORO_ACCENTS.get(vid[0])
out["gender"] = "female" if vid[1] == "f" else "male"
if acc:
out["languages"] = [acc[0]]
if "_" in vid:
out["name"] = vid.split("_", 1)[1].replace("-", " ").title()
elif family == "bark":
# v2/en_speaker_3 -> the language is in the id, the speaker is a number.
tail = vid.split("/")[-1]
if "_speaker_" in tail:
out["languages"] = [tail.split("_speaker_")[0]]
out["name"] = tail.replace("_", " ")
return out
def _voices_cache_file() -> Path:
return _cache_dir() / "tts_voices.json"
def remember_voices(model_name: str, backend) -> None:
"""Record a loaded backend's voice list so it can be listed without loading.
Best effort in every direction: a backend that cannot enumerate, an
unwritable cache or a malformed existing file must not affect synthesis,
which is what the caller is actually doing.
"""
try:
voices = [str(v) for v in (getattr(backend, "voices", None) or [])]
except Exception:
return
if not voices:
return
try:
import json
f = _voices_cache_file()
try:
cur = json.loads(f.read_text()) if f.is_file() else {}
if not isinstance(cur, dict):
cur = {}
except Exception:
cur = {}
if cur.get(model_name) == voices:
return
cur[model_name] = voices
f.parent.mkdir(parents=True, exist_ok=True)
tmp = f.with_suffix(".json.tmp")
tmp.write_text(json.dumps(cur, indent=2))
tmp.replace(f)
except Exception:
pass
def _remembered_voices(model_name: str) -> list:
try:
import json
f = _voices_cache_file()
if not f.is_file():
return []
cur = json.loads(f.read_text())
return [str(v) for v in (cur.get(model_name) or [])]
except Exception:
return []
def family_voices(model_name: str, backend=None) -> dict:
"""The model's built-in voices, and where the list came from.
``source`` is part of the answer on purpose: "none" means nothing is known
yet, which a caller should show differently from an engine that genuinely
has no voice list (``selection: "described"``).
"""
fam = _family(model_name)
ids, source = [], "none"
if backend is not None:
try:
ids = [str(v) for v in (getattr(backend, "voices", None) or [])]
if ids:
source = "loaded"
except Exception:
ids = []
if not ids:
ids = _remembered_voices(model_name)
if ids:
source = "remembered"
if not ids and fam == "bark":
# Deterministic ids, so they can be named without loading anything.
ids = [f"v2/en_speaker_{i}" for i in range(10)]
source = "static"
return {
"voices": [_voice_meta(fam, v) for v in ids],
"voices_source": source,
"voice_selection": _FAMILY_VOICE_SELECTION.get(fam, "builtin"),
}
def load_backend(model_name: str, model_path: Optional[str], config: Optional[dict]):
"""Instantiate the TTS backend for ``model_name`` (cached by the caller)."""
backend = _load_backend(model_name, model_path, config)
# Remember what it can say in, so a voice picker can be offered later
# without loading the model just to ask.
remember_voices(model_name, backend)
return backend
def _load_backend(model_name: str, model_path: Optional[str], config: Optional[dict]):
fam = _family(model_name)
config = config or {}
if fam == "kokoro":
......
......@@ -27,6 +27,7 @@ import io
import json
import os
import subprocess
import sys
import threading
from pathlib import Path
......@@ -61,6 +62,84 @@ def available() -> bool:
return False
_bootstrap_lock = threading.Lock()
_bootstrapped = False
def _venv_runs(py: Path) -> bool:
"""Whether this venv's interpreter actually RUNS (see parler_worker)."""
try:
return subprocess.run([str(py), "-c", "import sys"],
capture_output=True, timeout=120).returncode == 0
except Exception:
return False
def bootstrap(timeout: int = 3600) -> bool:
"""Create the venv and install coqui-tts into it (idempotent).
parler-tts builds its own venv on first use and XTTS did not, so the same
"needs an isolated venv" situation produced a self-healing engine in one
case and a 501 telling the operator to go and build one in the other. This
closes that gap.
Fully isolated for the reason in the module docstring: coqui-tts needs a
transformers the server cannot have, so sharing site-packages would either
break coqui or break the server.
Returns True when coqui-tts is importable afterwards. Never raises -- the
caller's fallback is the same 501 it would have raised anyway, and a
failure here must not take down a synthesis request for another engine.
"""
global _bootstrapped
if available():
_bootstrapped = True
return True
with _bootstrap_lock:
if available(): # another thread got there first
_bootstrapped = True
return True
d = venv_dir()
try:
py = venv_python()
# A venv built by a different interpreter leaves a dangling
# bin/python: is_file() says no, and `python -m venv` over the top
# then dies on that same dangling link, so the engine is stuck for
# good with a populated venv sitting right there. Clear it first.
if d.exists() and not _venv_runs(py):
import shutil
print(f"[xtts] {d} was built by a different interpreter "
f"(its python does not run here); rebuilding …",
flush=True)
shutil.rmtree(d, ignore_errors=True)
if not py.is_file():
print(f"[xtts] creating isolated venv at {d} …", flush=True)
d.parent.mkdir(parents=True, exist_ok=True)
subprocess.run([sys.executable, "-m", "venv", str(d)],
check=True, timeout=600)
py = venv_python()
print("[xtts] installing coqui-tts into the isolated venv "
"(first run, downloads several GB) …", flush=True)
subprocess.run([str(py), "-m", "pip", "install", "--no-input",
"coqui-tts", "soundfile", "requests"],
check=True, timeout=timeout)
except Exception as exc:
# A partially built venv is worse than none: `available()` stays
# False, and the next attempt resumes rather than reporting a venv
# that cannot import. Measured on a sibling worker: an install
# interrupted by a full disk left 42 packages and no torch, and
# the engine reported "venv can't be created" ever after.
print(f"[xtts] bootstrap failed ({type(exc).__name__}: "
f"{str(exc)[:200]}); XTTS stays unavailable", flush=True)
return False
ok = available()
_bootstrapped = ok
if not ok:
print("[xtts] coqui-tts installed but still not importable",
flush=True)
return ok
class XttsSubprocess:
"""A coqui TTS object living in another interpreter.
......
......@@ -79,14 +79,31 @@ def configured_hf_cache_dir(config_dir: Optional[str] = None) -> Optional[str]:
def hf_cache_env(hf_cache_dir: str) -> Dict[str, str]:
"""The four variables a child process needs to use ``hf_cache_dir``."""
"""The four variables a child process needs to use ``hf_cache_dir``.
All four name the SAME hub directory. ``TRANSFORMERS_CACHE`` is deprecated
but transformers 4.x still honours it, and it honours it as the cache_dir
handed to ``hf_hub_download`` -- i.e. as a hub-layout cache, not as a
second kind of store. Pointing it at ``<dir>/transformers`` therefore does
not give transformers its own corner of one cache; it gives the machine
TWO hub caches, and which one a process uses depends on the transformers
version in its venv.
Measured 2026-10-28: ``<dir>/hub`` held 122 models and
``<dir>/transformers`` held 2, both in ``models--org--name`` layout. An
isolated worker running transformers 4.x looked only in the second, found
nothing, and -- being forced offline so a missing model fails fast --
reported "We couldn't connect to https://huggingface.co" for a model that
was sitting in the other directory all along. The same split is what makes
a worker re-download a model the server already has.
"""
hf_cache_dir = os.path.expanduser(str(hf_cache_dir))
hub = os.path.join(hf_cache_dir, "hub")
return {
"HF_HOME": hf_cache_dir,
"HF_HUB_CACHE": hub,
"HUGGINGFACE_HUB_CACHE": hub,
"TRANSFORMERS_CACHE": os.path.join(hf_cache_dir, "transformers"),
"TRANSFORMERS_CACHE": hub,
}
......
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""A venv built by a different interpreter has to be rebuilt, not reused.
An isolated-worker venv keeps ``bin/python`` as a symlink to the interpreter
that created it. Create it with one python and run with another -- built on the
host and used from the container, or built before the image moved its
interpreter -- and that link dangles.
The failure is not a clean "missing": ``Path.exists()`` follows symlinks and
answers False, so the code decides the venv is absent and runs ``python -m
venv`` over the top, which trips on the very same dangling ``bin/python3``:
Error: [Errno 2] No such file or directory: '.../parler_venv/bin/python3'
Reported from production as "the venv can't be created", on a host where the
venv was fully populated and 7 GB of packages were sitting in it.
"""
import os
import subprocess
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import pytest
def _make_foreign_venv(root):
"""A venv whose interpreter symlink points at a path that is not there."""
(root / "bin").mkdir(parents=True)
(root / "lib" / "python3.13" / "site-packages").mkdir(parents=True)
os.symlink("python3.13", root / "bin" / "python")
os.symlink("/nonexistent/python3.13", root / "bin" / "python3.13")
(root / "pyvenv.cfg").write_text(
"home = /nonexistent/bin\ninclude-system-site-packages = false\n")
return root
def test_the_dangling_interpreter_is_not_mistaken_for_a_working_one(tmp_path):
from codai.api import parler_worker as P
v = _make_foreign_venv(tmp_path / "parler_venv")
py = v / "bin" / "python"
assert py.is_symlink()
assert not py.exists(), "precondition: the link dangles"
assert P._venv_runs(py) is False
def test_a_real_interpreter_is_accepted():
from codai.api import parler_worker as P
assert P._venv_runs(sys.executable) is True
def test_a_foreign_venv_is_cleared_before_recreating(tmp_path, monkeypatch):
"""Without the clear, `python -m venv` dies on the dangling link and the
engine reports "venv can't be created" on every request, for ever."""
from codai.api import parler_worker as P
v = _make_foreign_venv(tmp_path / "parler_venv")
monkeypatch.setattr(P, "_VENV_DIR", v)
monkeypatch.setattr(P, "_bootstrapped", False)
monkeypatch.setattr(P, "_pip_ok", lambda py: True)
created = []
def _fake_run(cmd, **kw):
cmd = [str(c) for c in cmd]
# The dangling interpreter cannot run anything -- without this the
# stub would answer "it works" and the clear would never be reached.
if cmd[0].endswith(("bin/python", "bin/python3.13")) \
and not os.path.exists(os.path.realpath(cmd[0])):
return subprocess.CompletedProcess(cmd, 1)
if "venv" in cmd:
# Mimic the real failure when the target still holds a dangling
# bin/python3, and success once it has been cleared.
if (v / "bin").exists():
raise subprocess.CalledProcessError(1, cmd)
created.append(cmd)
(v / "bin").mkdir(parents=True, exist_ok=True)
(v / "bin" / "python").write_text("#!/bin/sh\n")
return subprocess.CompletedProcess(cmd, 0)
monkeypatch.setattr(P.subprocess, "run", _fake_run)
P._bootstrap_venv()
assert created, "the venv was never recreated after the foreign one"
def test_both_workers_check_that_the_interpreter_runs():
"""The trap is identical in each isolated worker, so neither may rely on
a bare exists()/is_file() to decide a venv is usable."""
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
for rel in ("codai/api/parler_worker.py", "codai/api/xtts_worker.py"):
src = open(os.path.join(root, rel), encoding="utf-8").read()
assert "_venv_runs(" in src, f"{rel} does not verify the interpreter runs"
assert "rmtree" in src, f"{rel} cannot clear a foreign venv"
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""One HuggingFace cache, not two.
``TRANSFORMERS_CACHE`` is deprecated, but transformers 4.x still honours it --
and honours it as the ``cache_dir`` passed to ``hf_hub_download``, i.e. as a
hub-layout cache. Pointing it at ``<dir>/transformers`` does not give
transformers a corner of one cache; it gives the machine two hub caches, and
which one a process reads depends on the transformers version in its venv.
Measured 2026-10-28 on a production host: ``<dir>/hub`` held 122 models and
``<dir>/transformers`` held 2, both ``models--org--name``. A worker in an
isolated venv (transformers 4.46) looked only in the second one, found nothing,
and -- forced offline so a missing model fails fast -- reported "We couldn't
connect to https://huggingface.co" for a model that was present the whole time.
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
from codai.hf_cache import hf_cache_env
def test_every_variable_names_the_same_hub_directory():
env = hf_cache_env("/AI/huggingface")
hub = "/AI/huggingface/hub"
assert env["HF_HUB_CACHE"] == hub
assert env["HUGGINGFACE_HUB_CACHE"] == hub
assert env["TRANSFORMERS_CACHE"] == hub, (
"transformers 4.x would read a different cache from huggingface_hub, "
"so a model downloaded by the server is invisible to the worker")
assert env["HF_HOME"] == "/AI/huggingface"
def test_no_variable_points_at_a_second_store():
env = hf_cache_env("/AI/huggingface")
caches = {v for k, v in env.items() if k != "HF_HOME"}
assert len(caches) == 1, f"more than one cache directory in play: {caches}"
def test_a_user_path_is_expanded():
env = hf_cache_env("~/hfcache")
assert env["HF_HOME"] == os.path.expanduser("~/hfcache")
assert not env["HF_HUB_CACHE"].startswith("~")
def test_a_worker_env_carries_the_hub_cache(monkeypatch):
"""child_env is what the isolated workers actually receive."""
from codai.hf_cache import child_env
monkeypatch.setenv("HF_HOME", "/AI/huggingface")
env = child_env({})
assert env["TRANSFORMERS_CACHE"] == env["HF_HUB_CACHE"] == "/AI/huggingface/hub"
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""Listing a TTS model's built-in voices.
A client offers a voice picker before anything is loaded, so this has to be
answerable without loading a model -- a cheap catalogue call that loads is how
a listing route turns into a several-hundred-MB loader.
It also must not invent metadata. Kokoro encodes the accent and gender in the
id (``af_sarah`` = American female) so those are derived; a family whose ids
carry nothing reports ``None`` rather than a plausible-looking guess.
"""
import os
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import pytest
from codai.api import tts_backends as T
class _Loaded:
def __init__(self, voices):
self.voices = voices
def test_a_loaded_model_is_asked_directly():
got = T.family_voices("hexgrad/Kokoro-82M", _Loaded(["af_sarah", "bm_george"]))
assert got["voices_source"] == "loaded"
ids = [v["id"] for v in got["voices"]]
assert ids == ["af_sarah", "bm_george"]
def test_kokoro_ids_give_up_their_accent_and_gender():
got = T.family_voices("hexgrad/Kokoro-82M", _Loaded(["af_sarah", "bm_george"]))
sarah, george = got["voices"]
assert sarah["gender"] == "female" and sarah["languages"] == ["en-us"]
assert sarah["name"] == "Sarah"
assert george["gender"] == "male" and george["languages"] == ["en-gb"]
def test_an_id_that_carries_nothing_reports_none_not_a_guess():
"""A wrong gender against a voice is worse than no gender."""
got = T.family_voices("coqui/XTTS-v2", _Loaded(["Claribel Dervla"]))
v = got["voices"][0]
assert v["id"] == "Claribel Dervla"
assert v["gender"] is None
assert v["languages"] == []
def test_bark_is_listed_without_loading_anything():
got = T.family_voices("suno/bark", None)
assert got["voices_source"] == "static"
assert len(got["voices"]) == 10
assert got["voices"][0]["languages"] == ["en"]
def test_parler_says_it_is_described_rather_than_chosen():
"""Parler has no speaker names -- it is steered by a description. A client
reading only an empty `voices` would wrongly conclude it cannot be
directed at all."""
got = T.family_voices("parler-tts/parler-tts-mini-multilingual", None)
assert got["voices"] == []
assert got["voice_selection"] == "described"
def test_xtts_offers_cloning_as_well_as_named_speakers():
got = T.family_voices("coqui/XTTS-v2", None)
assert "clone" in got["voice_selection"]
def test_an_unloaded_model_reports_none_rather_than_loading():
got = T.family_voices("coqui/XTTS-v2", None)
assert got["voices_source"] in ("none", "remembered")
def test_voices_are_remembered_across_a_load(tmp_path, monkeypatch):
"""So the next listing needs no load at all."""
monkeypatch.setattr(T, "_cache_dir", lambda: tmp_path)
T.remember_voices("coqui/XTTS-v2", _Loaded(["Ana Florence", "Damien Black"]))
got = T.family_voices("coqui/XTTS-v2", None)
assert got["voices_source"] == "remembered"
assert [v["id"] for v in got["voices"]] == ["Ana Florence", "Damien Black"]
def test_remembering_never_breaks_synthesis(tmp_path, monkeypatch):
"""It runs on the load path, so every failure mode is swallowed."""
monkeypatch.setattr(T, "_cache_dir", lambda: tmp_path)
class _Angry:
@property
def voices(self):
raise RuntimeError("cannot enumerate")
T.remember_voices("x", _Angry()) # must not raise
T.remember_voices("y", None) # must not raise
monkeypatch.setattr(T, "_cache_dir",
lambda: tmp_path / "nope" / "deeper")
T.remember_voices("z", _Loaded(["a"])) # unwritable: must not raise
def test_a_corrupt_cache_is_ignored(tmp_path, monkeypatch):
monkeypatch.setattr(T, "_cache_dir", lambda: tmp_path)
(tmp_path / "tts_voices.json").write_text("{not json")
assert T.family_voices("coqui/XTTS-v2", None)["voices"] == []
T.remember_voices("coqui/XTTS-v2", _Loaded(["Ana Florence"]))
assert T.family_voices("coqui/XTTS-v2", None)["voices_source"] == "remembered"
def test_the_capabilities_route_does_not_collide_with_voiceprints():
"""/v1/audio/speakers is enrolled speaker voiceprints (recognition). The
built-in TTS voices must NOT be served from that path."""
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
import re
tts = open(os.path.join(root, "codai/api/tts.py"), encoding="utf-8").read()
# Route DECLARATIONS only -- the path is named in a comment here on
# purpose, to say why it is not being used.
routes = re.findall(r'@router\.\w+\(\s*"([^"]+)"', tts)
assert not [r for r in routes if r.startswith("/v1/audio/speakers")], (
f"the TTS voice list is being served from the speaker-recognition "
f"path; routes declared here: {routes}")
assert "/v1/audio/speech/capabilities" in routes
# CoderAI - OpenAI-compatible API server
# Copyright (C) 2026 Stefy Lanza <stefy@nexlab.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with this program. If not, see <https://www.gnu.org/licenses/>.
"""XTTS builds its own venv, like parler does.
Both engines need an isolated interpreter for the same reason (a transformers
the server cannot have). parler bootstrapped its venv on first use; XTTS only
LOOKED for one and raised a 501 telling the operator to build it by hand. Same
situation, two different behaviours -- reported from production as "speech
engines listed but not installed".
Nothing here touches the network: the subprocess calls are stubbed, and what is
asserted is the decision-making around them.
"""
import os
import subprocess
import sys
sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
import pytest
from codai.api import xtts_worker as X
@pytest.fixture(autouse=True)
def _no_real_subprocess(monkeypatch):
calls = []
def _fake_run(cmd, **kw):
calls.append([str(c) for c in cmd])
return subprocess.CompletedProcess(cmd, 0)
monkeypatch.setattr(X.subprocess, "run", _fake_run)
monkeypatch.setattr(X, "_bootstrapped", False)
return calls
def test_an_existing_venv_is_not_rebuilt(monkeypatch, _no_real_subprocess):
monkeypatch.setattr(X, "available", lambda: True)
assert X.bootstrap() is True
assert _no_real_subprocess == [], "it reinstalled over a working venv"
def test_a_missing_venv_is_created_and_coqui_installed(monkeypatch, tmp_path,
_no_real_subprocess):
seen = {"n": 0}
def _avail():
# Unavailable on the way in, available once the install has run.
seen["n"] += 1
return seen["n"] > 1 and len(_no_real_subprocess) >= 1
monkeypatch.setattr(X, "available", _avail)
monkeypatch.setattr(X, "venv_dir", lambda: tmp_path / "xtts_venv")
monkeypatch.setattr(X, "venv_python",
lambda: tmp_path / "xtts_venv" / "bin" / "python")
assert X.bootstrap() is True
flat = " ".join(" ".join(c) for c in _no_real_subprocess)
assert "-m venv" in flat or "venv" in flat
assert "coqui-tts" in flat, "coqui-tts was never installed"
def test_a_failed_install_does_not_raise(monkeypatch, tmp_path):
"""The caller's fallback is the same 501 it would have raised anyway, and
one engine's bad install must not break a request for another."""
def _boom(cmd, **kw):
raise subprocess.CalledProcessError(1, cmd)
monkeypatch.setattr(X.subprocess, "run", _boom)
monkeypatch.setattr(X, "available", lambda: False)
monkeypatch.setattr(X, "venv_dir", lambda: tmp_path / "v")
monkeypatch.setattr(X, "venv_python", lambda: tmp_path / "v" / "bin" / "python")
assert X.bootstrap() is False # must not raise
def test_an_install_that_yields_no_import_reports_failure(monkeypatch, tmp_path,
_no_real_subprocess):
"""A partially built venv is worse than none: an interrupted install on a
sibling worker left 42 packages and no torch, and the engine reported
"venv can't be created" from then on."""
monkeypatch.setattr(X, "available", lambda: False)
monkeypatch.setattr(X, "venv_dir", lambda: tmp_path / "v")
monkeypatch.setattr(X, "venv_python", lambda: tmp_path / "v" / "bin" / "python")
assert X.bootstrap() is False
def test_the_backend_asks_for_a_bootstrap_and_can_be_opted_out():
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
src = open(os.path.join(root, "codai/api/tts_backends.py"),
encoding="utf-8").read()
assert "xtts_worker.bootstrap()" in src, (
"the coqui backend no longer builds the venv on first use")
assert "CODERAI_XTTS_AUTOINSTALL" in src, (
"a host with no network cannot opt out of the auto-install")
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment