tts: run coqui XTTS from a venv of its own

coqui-tts imports a transformers 4.x symbol that 5.x removed. Unlike parler and
the OCR engines it had never been isolated locally: _CoquiBackend did a plain
'from TTS.api import TTS' and gave up when it was missing. On a pod that meant
an image with a verified, working TTS venv answered 'pip install coqui-tts' —
the venv was there and nothing knew to use it.

xtts_worker finds the venv the way the others do (explicit env var, the baked
/opt/coderai/venvs/TTS, then per-user) and drives tools/xtts_service.py in it
over stdin/stdout. XttsSubprocess wears the TTS object's shape — speakers,
tts(**kwargs), synthesizer.output_sample_rate, and the TypeError the backend
relies on to retry without speed — so synthesize() runs unchanged whether the
library is in this process or next door.

Verified inside the pushed image: the venv is found and the service answers
from it. The fall-through is unit-tested with a fake worker, and the error when
no venv exists names the places it looked.
Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_0182DhR1eFNPQyGmDrpHbedY

codai/api/tts_backends.py

@@ -199,15 +199,27 @@ class _CoquiBackend:
     default_voice = ""
 
     def __init__(self, model_name: str, config: dict):
+        self._cfg = config or {}
         try:
-            from TTS.api import TTS  # coqui-tts
+            from TTS.api import TTS  # coqui-tts, in this venv
         except ImportError as e:
+            # Not here — but coqui-tts cannot share a transformers-5 venv, so
+            # "not here" is the normal case. Look for it in a venv of its own,
+            # the way parler and the OCR engines are found. A pod image with a
+            # 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
+            if xtts_worker.available():
+                print(f"[tts] coqui-tts from {xtts_worker.venv_dir()}", flush=True)
+                self._tts = xtts_worker.XttsSubprocess(model_name)
+                return
             raise MissingEngineError(
-                "Coqui/XTTS models need the coqui-tts package: "
-                "pip install coqui-tts"
+                "Coqui/XTTS models need the coqui-tts package, and no venv "
+                "containing it was found (CODERAI_XTTS_VENV, /opt/coderai/venvs/TTS, "
+                "~/.coderai/xtts_venv). It cannot be installed beside "
+                "transformers 5.x: build it in its own venv."
             ) from e
         import torch
-        self._cfg = config or {}
         self._tts = TTS(model_name).to("cuda" if torch.cuda.is_available() else "cpu")
 
     def synthesize(self, text: str, voice: str, speed: float, lang: str,

codai/api/xtts_worker.py

+# 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.
+
+"""Drive coqui XTTS from a venv of its own.
+
+coqui-tts imports isin_mps_friendly from transformers.pytorch_utils, which 5.x
+removed. The server runs 5.x, and pinning 4.x there would break every other
+capability — so XTTS gets its own interpreter, like parler-tts and the OCR
+engines. This module speaks to tools/xtts_service.py running in that venv.
+
+Unlike those, XTTS had never been isolated locally: _CoquiBackend did a plain
+`from TTS.api import TTS` and gave up when it was missing. On a pod that meant
+an image with a verified, working TTS venv answered "pip install coqui-tts" —
+the venv was there and nothing knew to use it.
+
+The venv is found the way the other workers find theirs: an explicit env var,
+then the location the pod images bake it to, then a per-user directory.
+"""
+
+import base64
+import io
+import json
+import os
+import subprocess
+import threading
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+_SERVICE = _REPO_ROOT / "tools" / "xtts_service.py"
+
+
+def venv_dir() -> Path:
+    explicit = os.environ.get("CODERAI_XTTS_VENV", "").strip()
+    if explicit:
+        return Path(explicit)
+    baked = Path("/opt/coderai/venvs/TTS")
+    if baked.exists():
+        return baked
+    return Path(os.path.expanduser("~/.coderai/xtts_venv"))
+
+
+def venv_python() -> Path:
+    return venv_dir() / ("Scripts" if os.name == "nt" else "bin") / (
+        "python.exe" if os.name == "nt" else "python")
+
+
+def available() -> bool:
+    """True when there is a venv with coqui-tts in it to talk to."""
+    py = venv_python()
+    if not py.is_file() or not _SERVICE.is_file():
+        return False
+    try:
+        return subprocess.run([str(py), "-c", "import TTS"],
+                              capture_output=True, timeout=120).returncode == 0
+    except Exception:
+        return False
+
+
+class XttsSubprocess:
+    """A coqui TTS object living in another interpreter.
+
+    Exposes what _CoquiBackend needs — `speakers`, `tts(**kwargs)` returning
+    a float32 array, and the output sample rate — so the backend's synthesize()
+    logic (speaker vs speaker_wav selection, speed fallback) runs unchanged.
+    """
+
+    def __init__(self, model_name: str):
+        self._lock = threading.RLock()
+        self._proc = subprocess.Popen(
+            [str(venv_python()), str(_SERVICE)],
+            stdin=subprocess.PIPE, stdout=subprocess.PIPE, text=True, bufsize=1,
+            env={**os.environ, "PYTHONUNBUFFERED": "1"})
+        reply = self._call({"op": "load", "model": model_name})
+        if not reply.get("ok"):
+            self.close()
+            raise RuntimeError(f"XTTS venv could not load {model_name!r}: "
+                               f"{reply.get('error')}")
+        self.speakers = list(reply.get("speakers") or [])
+        self.sample_rate = int(reply.get("sample_rate") or 24000)
+        # _CoquiBackend reads `self._tts.synthesizer.output_sample_rate`, the
+        # way it would off a real TTS object. Give it the same shape.
+        self.synthesizer = type("_Synth", (), {})()
+        self.synthesizer.output_sample_rate = self.sample_rate
+
+    def tts(self, **kwargs):
+        import numpy as np
+        reply = self._call({"op": "tts", **kwargs})
+        if not reply.get("ok"):
+            err = str(reply.get("error") or "")
+            # The in-process backend relies on TypeError to retry without
+            # `speed`; reproduce that so its fallback still works across the wire.
+            if err.startswith("TypeError"):
+                raise TypeError(err)
+            raise RuntimeError(f"XTTS synthesis failed: {err}")
+        import soundfile as sf
+        wav, sr = sf.read(io.BytesIO(base64.b64decode(reply["b64_wav"])), dtype="float32")
+        self.sample_rate = int(sr)
+        self.synthesizer.output_sample_rate = self.sample_rate
+        return np.asarray(wav, dtype=np.float32)
+
+    def _call(self, request: dict) -> dict:
+        with self._lock:
+            if self._proc.poll() is not None:
+                return {"ok": False, "error": "the XTTS worker has exited"}
+            self._proc.stdin.write(json.dumps(request) + "\n")
+            self._proc.stdin.flush()
+            line = self._proc.stdout.readline()
+        if not line:
+            return {"ok": False, "error": "the XTTS worker returned nothing"}
+        try:
+            return json.loads(line)
+        except Exception as exc:
+            return {"ok": False, "error": f"unparseable worker reply: {exc}"}
+
+    def close(self) -> None:
+        try:
+            if self._proc.poll() is None:
+                self._proc.terminate()
+                self._proc.wait(timeout=10)
+        except Exception:
+            pass
+
+    def __del__(self):
+        self.close()

tests/test_remote_execution.py

@@ -1733,3 +1733,41 @@ def test_asking_for_audiocraft_when_absent_is_an_error_not_a_substitution():
             raise AssertionError("expected a RuntimeError")
     finally:
         builtins.__import__ = real
+
+
+def test_xtts_falls_through_to_its_own_venv_when_not_importable(monkeypatch):
+    """coqui-tts cannot share a transformers-5 venv, so 'not importable here' is
+    the normal case. A pod image with a verified TTS venv used to answer 'pip
+    install coqui-tts' — the venv was there and nothing knew to use it."""
+    import builtins, numpy as np
+    from codai.api import tts_backends, xtts_worker
+
+    real = builtins.__import__
+    def _no_tts(name, *a, **k):
+        if name == "TTS.api" or name == "TTS":
+            raise ImportError("not in this venv")
+        return real(name, *a, **k)
+    monkeypatch.setattr(builtins, "__import__", _no_tts)
+
+    class _FakeSub:
+        def __init__(self, model):
+            self.speakers = ["Ana Florence"]
+            self.synthesizer = type("S", (), {"output_sample_rate": 24000})()
+        def tts(self, **kw):
+            assert kw["speaker"] == "Ana Florence" and kw["language"] == "en"
+            return np.zeros(2400, dtype=np.float32)
+    monkeypatch.setattr(xtts_worker, "available", lambda: True)
+    monkeypatch.setattr(xtts_worker, "XttsSubprocess", _FakeSub)
+
+    b = tts_backends._CoquiBackend("coqui/XTTS-v2", {})
+    wav, sr = b.synthesize("hello", voice="", speed=1.0, lang="en")
+    assert sr == 24000 and len(wav) == 2400
+
+    # and with no venv either, the error says where it looked
+    monkeypatch.setattr(xtts_worker, "available", lambda: False)
+    try:
+        tts_backends._CoquiBackend("coqui/XTTS-v2", {})
+    except tts_backends.MissingEngineError as e:
+        assert "/opt/coderai/venvs/TTS" in str(e)
+    else:
+        raise AssertionError("expected MissingEngineError")

tools/xtts_service.py

+#!/usr/bin/env python3
+# 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.
+
+"""coqui XTTS in its own venv, spoken to over stdin/stdout.
+
+coqui-tts imports a transformers 4.x symbol (isin_mps_friendly) that 5.x removed,
+and the server runs 5.x — so it cannot share the main venv. This script is the
+far side of the split: one JSON object per line in, one per line out, the same
+shape audiocraft_service.py and pyannote_service.py use.
+
+  {"op": "load", "model": "coqui/XTTS-v2"}                       -> {"ok": true, "speakers": [...], "sample_rate": 24000}
+  {"op": "tts", "text": "...", "language": "en", "speaker": "..",
+   "speaker_wav": "/path/or/absent", "speed": 1.0}                -> {"ok": true, "b64_wav": "...", "sample_rate": 24000}
+  {"op": "ping"}                                                   -> {"ok": true}
+
+Audio returns as base64 WAV: the caller may be a pod, and a few seconds of
+speech is small enough that a file would only be a thing to clean up.
+"""
+
+import base64
+import io
+import json
+import sys
+
+_state = {"tts": None, "sr": 24000}
+
+
+def _load(name: str) -> dict:
+    import torch
+    from TTS.api import TTS
+    tts = TTS(name).to("cuda" if torch.cuda.is_available() else "cpu")
+    _state["tts"] = tts
+    # XTTS reports its output rate on the synthesizer; fall back to its default.
+    try:
+        _state["sr"] = int(tts.synthesizer.output_sample_rate)
+    except Exception:
+        _state["sr"] = 24000
+    return {"ok": True, "speakers": list(getattr(tts, "speakers", None) or []),
+            "sample_rate": _state["sr"]}
+
+
+def _tts(req: dict) -> dict:
+    tts = _state["tts"]
+    if tts is None:
+        return {"ok": False, "error": "no model loaded"}
+    kwargs = {"text": req.get("text") or "", "language": (req.get("language") or "en")[:2]}
+    for key in ("speaker", "speaker_wav"):
+        if req.get(key):
+            kwargs[key] = req[key]
+    speed = req.get("speed")
+    try:
+        if speed:
+            kwargs["speed"] = float(speed)
+        wav = tts.tts(**kwargs)
+    except TypeError:
+        kwargs.pop("speed", None)          # some coqui models do not take speed
+        wav = tts.tts(**kwargs)
+    import numpy as np
+    import soundfile as sf
+    buf = io.BytesIO()
+    sf.write(buf, np.asarray(wav, dtype="float32"), _state["sr"], format="WAV")
+    return {"ok": True, "b64_wav": base64.b64encode(buf.getvalue()).decode(),
+            "sample_rate": _state["sr"]}
+
+
+def main() -> int:
+    for line in sys.stdin:
+        line = line.strip()
+        if not line:
+            continue
+        try:
+            req = json.loads(line)
+            op = req.get("op")
+            if op == "ping":
+                out = {"ok": True}
+            elif op == "load":
+                out = _load(str(req.get("model") or ""))
+            elif op == "tts":
+                out = _tts(req)
+            else:
+                out = {"ok": False, "error": f"unknown op {op!r}"}
+        except Exception as exc:
+            # One bad request must not take the loaded model down with it.
+            out = {"ok": False, "error": f"{type(exc).__name__}: {exc}"}
+        sys.stdout.write(json.dumps(out) + "\n")
+        sys.stdout.flush()
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())