feat(hermes-tts): multilingual Piper voice policy (en=amy, ru=irina, es=claude)

Supersedes draft PR #24 (hermes/tts-voice-hfc-female): Brad changed the
decision after that task landed, so this starts fresh from origin/main
instead of building on it.

Bakes three checksum-pinned Piper voices (en_US-amy-medium,
ru_RU-irina-medium, es_MX-claude-high) alongside the existing lessac set,
and adds deterministic, allow-listed language routing to
hermes-jetson-tts-server.py: an explicit request "language" field maps
through a fixed dict to one of the three baked voices, with unknown,
missing, or malformed input always falling back to English amy. A
client-supplied "voice" field is never read, so no client input can reach
a filesystem path. All three voices are eagerly preloaded at process
start (measured ~243MB RSS for three vs. ~88MB for one).

The WebUI has no signal for the language of the text it is about to
speak (verified: hermes-webui-atlas-voice.js sends only text and engine),
so no client- or server-side language detection is added; this gap is
documented in NOTES.md and the PR description rather than papered over.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>

dockerfiles/Dockerfile.hermes-jetson-tts

@@ -24,18 +24,41 @@ ADD --checksum=sha256:f7d01dde371555732c4c314111ac79672b1a5ce2fc19266ab42178fd8d
 ADD --checksum=sha256:45754dfdebb3b8661c3fc564713772deec6e064feeb5b4e9594857dc7305193a --chmod=0444 \
     https://huggingface.co/rhasspy/piper-voices/resolve/ea046e8458f6acd997706d6e6066a022b42f6fb1/en/en_US/lessac/low/en_US-lessac-low.onnx.json?download=true \
     /opt/models/piper/en_US-lessac-low.onnx.json
+
+# Multilingual chat voice policy: English -> amy, Russian -> irina, Spanish ->
+# claude (Mexican Spanish, the only "claude" voice rhasspy/piper-voices
+# publishes; there is no es_ES-claude).
+ADD --checksum=sha256:b3a6e47b57b8c7fbe6a0ce2518161a50f59a9cdd8a50835c02cb02bdd6206c18 --chmod=0444 \
+    https://huggingface.co/rhasspy/piper-voices/resolve/ea046e8458f6acd997706d6e6066a022b42f6fb1/en/en_US/amy/medium/en_US-amy-medium.onnx?download=true \
+    /opt/models/piper/en_US-amy-medium.onnx
+ADD --checksum=sha256:95a23eb4d42909d38df73bb9ac7f45f597dbfcde2d1bf9526fdeaf5466977d77 --chmod=0444 \
+    https://huggingface.co/rhasspy/piper-voices/resolve/ea046e8458f6acd997706d6e6066a022b42f6fb1/en/en_US/amy/medium/en_US-amy-medium.onnx.json?download=true \
+    /opt/models/piper/en_US-amy-medium.onnx.json
+ADD --checksum=sha256:8ff38212d23da300bbe3705c645e6e5b9475f0bfde01558eb17813e22acaaaaa --chmod=0444 \
+    https://huggingface.co/rhasspy/piper-voices/resolve/ea046e8458f6acd997706d6e6066a022b42f6fb1/ru/ru_RU/irina/medium/ru_RU-irina-medium.onnx?download=true \
+    /opt/models/piper/ru_RU-irina-medium.onnx
+ADD --checksum=sha256:c2ec28bb38e2b59e93b959b3e40348c1afebbd272f30fed5d41205d08e98a9d7 --chmod=0444 \
+    https://huggingface.co/rhasspy/piper-voices/resolve/ea046e8458f6acd997706d6e6066a022b42f6fb1/ru/ru_RU/irina/medium/ru_RU-irina-medium.onnx.json?download=true \
+    /opt/models/piper/ru_RU-irina-medium.onnx.json
+ADD --checksum=sha256:3ef40a71ea63852cd8ab7e6fa7d2ecdcfa67a0b47c9c48e3f10e02ee02083ea0 --chmod=0444 \
+    https://huggingface.co/rhasspy/piper-voices/resolve/ea046e8458f6acd997706d6e6066a022b42f6fb1/es/es_MX/claude/high/es_MX-claude-high.onnx?download=true \
+    /opt/models/piper/es_MX-claude-high.onnx
+ADD --checksum=sha256:1afc81f703c0e4cb3b4d7c0dca096b8b54a98806807f0170cf5eb5557723c12d --chmod=0444 \
+    https://huggingface.co/rhasspy/piper-voices/resolve/ea046e8458f6acd997706d6e6066a022b42f6fb1/es/es_MX/claude/high/es_MX-claude-high.onnx.json?download=true \
+    /opt/models/piper/es_MX-claude-high.onnx.json
 RUN chmod 0555 /opt/models /opt/models/piper
 
 COPY dockerfiles/hermes-jetson-tts-server.py /opt/atlas/hermes-jetson-tts-server.py
 RUN chmod 0555 /opt/atlas/hermes-jetson-tts-server.py
 
-# Load the actual pinned voice during the ARM64 build. This catches package or
+# Load every pinned voice during the ARM64 build, including the three baked
-# model-format drift before the image can reach Flux.
+# for the multilingual chat policy. This catches package or model-format
-RUN python -c "import stat; from pathlib import Path; from piper import PiperVoice; p=Path('/opt/models/piper'); models=[p/'en_US-lessac-high.onnx',p/'en_US-lessac-medium.onnx',p/'en_US-lessac-low.onnx']; assert stat.S_IMODE(p.stat().st_mode)==0o555; assert all(stat.S_IMODE(model.stat().st_mode)==0o444 for model in models); voices=[PiperVoice.load(model,Path(str(model)+'.json'),use_cuda=False,download_dir=p) for model in models]; assert all(voice.config.sample_rate>0 for voice in voices)"
+# drift before the image can reach Flux.
+RUN python -c "import stat; from pathlib import Path; from piper import PiperVoice; p=Path('/opt/models/piper'); models=[p/'en_US-lessac-high.onnx',p/'en_US-lessac-medium.onnx',p/'en_US-lessac-low.onnx',p/'en_US-amy-medium.onnx',p/'ru_RU-irina-medium.onnx',p/'es_MX-claude-high.onnx']; assert stat.S_IMODE(p.stat().st_mode)==0o555; assert all(stat.S_IMODE(model.stat().st_mode)==0o444 for model in models); voices=[PiperVoice.load(model,Path(str(model)+'.json'),use_cuda=False,download_dir=p) for model in models]; assert all(voice.config.sample_rate>0 for voice in voices)"
 
 ENV HERMES_TTS_HOST=0.0.0.0 \
     HERMES_TTS_PORT=9001 \
-    HERMES_TTS_VOICE=en_US-lessac-medium \
+    HERMES_TTS_VOICE=en_US-amy-medium \
     HERMES_TTS_CACHE=/opt/models/piper \
     OMP_NUM_THREADS=2 \
     PYTHONDONTWRITEBYTECODE=1 \

dockerfiles/hermes-jetson-tts-server.py

@@ -17,12 +17,50 @@ from piper import PiperConfig, PiperVoice, SynthesisConfig
 
 HOST = os.getenv("HERMES_TTS_HOST", "0.0.0.0")
 PORT = int(os.getenv("HERMES_TTS_PORT", "9001"))
-VOICE_NAME = os.getenv("HERMES_TTS_VOICE", "en_US-lessac-high")
 CACHE_DIR = Path(os.getenv("HERMES_TTS_CACHE", "/cache/piper"))
 MAX_TEXT_CHARS = 5000
 ONNX_THREADS = max(1, int(os.getenv("HERMES_TTS_ONNX_THREADS", "4")))
 VOICE_LOCK = threading.Lock()
 
+# Fixed, allow-listed language -> baked voice mapping. This is the ONLY path
+# from a client-supplied string to a model name: client input is looked up
+# here and never used to build a filesystem path directly. Both "-" and "_"
+# separators and any case are accepted; anything not present here falls back
+# to DEFAULT_VOICE_NAME (safe English default), never an error and never an
+# unbaked model.
+LANGUAGE_VOICE_MAP = {
+    "en": "en_US-amy-medium",
+    "en-us": "en_US-amy-medium",
+    "ru": "ru_RU-irina-medium",
+    "ru-ru": "ru_RU-irina-medium",
+    "es": "es_MX-claude-high",
+    "es-mx": "es_MX-claude-high",
+    "es-es": "es_MX-claude-high",
+}
+BAKED_VOICE_NAMES = frozenset(LANGUAGE_VOICE_MAP.values())
+DEFAULT_VOICE_NAME = os.getenv("HERMES_TTS_VOICE", "en_US-amy-medium")
+
+
+def normalize_language(value: object) -> str | None:
+    """Lowercase and fold "_"/"-" separators; reject non-string/blank input."""
+    if not isinstance(value, str):
+        return None
+    normalized = value.strip().lower().replace("_", "-")
+    return normalized or None
+
+
+def resolve_voice_name(language: object) -> str:
+    """Map a client-supplied language to one of the baked policy voices.
+
+    Unknown, missing, or malformed language always resolves to the safe
+    default rather than raising, and the result is always a member of
+    BAKED_VOICE_NAMES.
+    """
+    normalized = normalize_language(language)
+    if normalized is None:
+        return DEFAULT_VOICE_NAME
+    return LANGUAGE_VOICE_MAP.get(normalized, DEFAULT_VOICE_NAME)
+
 
 def _json(handler: BaseHTTPRequestHandler, status: int, payload: dict) -> None:
     body = json.dumps(payload).encode("utf-8")
@@ -46,7 +84,16 @@ class SpeechHandler(BaseHTTPRequestHandler):
         if self.path != "/health":
             _json(self, 404, {"error": "not found"})
             return
-        _json(self, 200, {"ok": True, "voice": VOICE_NAME, "device": "cpu"})
+        _json(
+            self,
+            200,
+            {
+                "ok": True,
+                "voices": sorted(self.server.voices),  # type: ignore[attr-defined]
+                "default_voice": self.server.default_voice_name,  # type: ignore[attr-defined]
+                "device": "cpu",
+            },
+        )
 
     def do_POST(self) -> None:
         if self.path != "/v1/audio/speech":
@@ -71,10 +118,16 @@ class SpeechHandler(BaseHTTPRequestHandler):
             return
         speed = min(2.0, max(0.5, speed))
 
+        # Policy is driven ONLY by "language". A client-supplied "voice"
+        # field is deliberately never read here; it cannot override the
+        # allow-listed mapping.
+        voice_name = resolve_voice_name(payload.get("language"))
+        voice = self.server.voices[voice_name]  # type: ignore[attr-defined]
+
         output = io.BytesIO()
         try:
             with VOICE_LOCK, wave.open(output, "wb") as wav_file:
-                self.server.voice.synthesize_wav(  # type: ignore[attr-defined]
+                voice.synthesize_wav(
                     text,
                     wav_file,
                     SynthesisConfig(length_scale=1.0 / speed),
@@ -84,6 +137,7 @@ class SpeechHandler(BaseHTTPRequestHandler):
             self.send_header("Content-Type", "audio/wav")
             self.send_header("Content-Length", str(len(audio)))
             self.send_header("Cache-Control", "no-store")
+            self.send_header("X-TTS-Voice", voice_name)
             self.end_headers()
             self.wfile.write(audio)
         except Exception as exc:
@@ -91,30 +145,54 @@ class SpeechHandler(BaseHTTPRequestHandler):
             _json(self, 500, {"error": "speech synthesis failed"})
 
 
-def main() -> None:
+def _load_voice(cache_dir: Path, voice_name: str, threads: int) -> PiperVoice:
-    """Load the checksum-pinned voice from the image and serve it on CPU."""
+    model_path = cache_dir / f"{voice_name}.onnx"
-    CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    config_path = cache_dir / f"{voice_name}.onnx.json"
-    model_path = CACHE_DIR / f"{VOICE_NAME}.onnx"
-    config_path = CACHE_DIR / f"{VOICE_NAME}.onnx.json"
     if not model_path.exists() or not config_path.exists():
-        raise RuntimeError(f"baked Piper voice is missing: {VOICE_NAME}")
+        raise RuntimeError(f"baked Piper voice is missing: {voice_name}")
     with config_path.open("r", encoding="utf-8") as config_file:
         config = PiperConfig.from_dict(json.load(config_file))
     session_options = onnxruntime.SessionOptions()
-    session_options.intra_op_num_threads = ONNX_THREADS
+    session_options.intra_op_num_threads = threads
     session_options.inter_op_num_threads = 1
     session = onnxruntime.InferenceSession(
         str(model_path),
         sess_options=session_options,
         providers=["CPUExecutionProvider"],
     )
-    voice = PiperVoice(session=session, config=config, download_dir=CACHE_DIR)
+    return PiperVoice(session=session, config=config, download_dir=cache_dir)
+
+
+def load_voices(cache_dir: Path, threads: int) -> dict[str, PiperVoice]:
+    """Eagerly load all three policy voices.
+
+    Preload (not lazy-load-on-first-use) was chosen deliberately: measured
+    RSS on this model set is ~88MB for one voice and ~243MB for all three
+    (~+155MB versus the previous single-voice baseline), which comfortably
+    fits the pod's memory budget on the CPU-only voice node. Preloading
+    avoids a slow, request-serializing first synthesis per language and
+    keeps the fail-closed missing-model check (below) at process start
+    rather than deferring a possible crash to a live user request.
+    """
+    return {name: _load_voice(cache_dir, name, threads) for name in sorted(BAKED_VOICE_NAMES)}
+
+
+def main() -> None:
+    """Load the checksum-pinned policy voices from the image and serve them on CPU."""
+    CACHE_DIR.mkdir(parents=True, exist_ok=True)
+    if DEFAULT_VOICE_NAME not in BAKED_VOICE_NAMES:
+        raise RuntimeError(
+            f"HERMES_TTS_VOICE must name one of the baked policy voices: {sorted(BAKED_VOICE_NAMES)}"
+        )
+    voices = load_voices(CACHE_DIR, ONNX_THREADS)
     print(
-        f"[tts] loaded Piper voice {VOICE_NAME} on CPU with {ONNX_THREADS} ONNX threads",
+        f"[tts] loaded {len(voices)} Piper voices on CPU with {ONNX_THREADS} ONNX threads each: "
+        + ", ".join(sorted(voices)),
         flush=True,
     )
     server = ThreadingHTTPServer((HOST, PORT), SpeechHandler)
-    server.voice = voice  # type: ignore[attr-defined]
+    server.voices = voices  # type: ignore[attr-defined]
+    server.default_voice_name = DEFAULT_VOICE_NAME  # type: ignore[attr-defined]
     print(f"[tts] ready on {HOST}:{PORT}", flush=True)
     server.serve_forever(poll_interval=0.25)
 

dockerfiles/hermes-webui-atlas-patch.py

@@ -60,10 +60,14 @@ atlas = '''    # ── Atlas private Jetson TTS ──────────
                 speed = max(0.5, min(2.0, 1.0 + (float(rate_str.rstrip("%")) / 100.0)))
             except ValueError:
                 speed = 1.0
+        # No "voice" or "language" field: the WebUI has no signal for the
+        # language of the text being spoken (see NOTES.md), so voice
+        # selection is left entirely to the TTS service's own allow-listed
+        # policy (English amy) rather than sending a value that would only
+        # be ignored server-side or a fabricated language guess.
         request_body = json.dumps({
             "model": "piper",
             "input": text,
-            "voice": "en_US-lessac-high",
             "speed": speed,
         }).encode("utf-8")
         request = Request(atlas_url, data=request_body, headers={

services/hermes/NOTES.md

@@ -40,6 +40,31 @@ or the WebUI. Browser chat remains available when `bot_token` is empty.
 The bot token and relay key must never be added to Git or a Kubernetes Secret.
 The router does not log prompt bodies, raw Telegram IDs, link codes, or tokens.
 
+## Private Jetson voice: multilingual TTS policy
+
+`hermes-tts` on `titan-21` bakes three checksum-pinned Piper voices and
+selects one per request from a fixed, allow-listed `language` field: `en`/
+`en-US` → `en_US-amy-medium`, `ru`/`ru-RU` → `ru_RU-irina-medium`, `es`/
+`es-MX`/`es-ES` → `es_MX-claude-high` (Piper's `claude` voice is Mexican
+Spanish; there is no Castilian `es_ES-claude`). Matching is case-insensitive
+and accepts both `_` and `-` separators. Any language that is missing,
+unrecognized, or malformed falls back to English amy rather than erroring.
+The mapping is a fixed dict from `language` to one of the three baked model
+names only — a client-supplied `voice` field is never read, so no client
+input can select or construct a model path. All three voices are preloaded
+at process start (see `dockerfiles/hermes-jetson-tts-server.py`).
+
+The private WebUI voice bridge (`dockerfiles/hermes-webui-atlas-voice.js`,
+patched into `api/routes.py` by `hermes-webui-atlas-patch.py`) has no signal
+for the language of the assistant reply it is about to speak — it sends only
+`text` and `engine`. Until the WebUI or gateway attaches an explicit
+`language` field to that request, every reply speaks in the safe English
+default regardless of its actual language. Closing that gap needs a language
+signal upstream of the TTS call (e.g. tagging the assistant turn with a
+detected/declared reply language and threading it through
+`hermes-webui-atlas-voice.js` → `api/routes.py` → the `language` field), not
+client- or server-side guessing bolted onto the TTS service itself.
+
 ## The one-sentence explanation
 
 Hermes is the persistent agent runtime and control surface; Codex or the local

services/hermes/voice-deployment.yaml

@@ -114,7 +114,7 @@ spec:
         app: hermes-tts
       annotations:
         ai.bstein.dev/role: private-chat-text-to-speech
-        ai.bstein.dev/model: piper-en-us-lessac-medium
+        ai.bstein.dev/model: piper-multilingual-en_US-amy-medium+ru_RU-irina-medium+es_MX-claude-high
         ai.bstein.dev/gpu: CPU-only beside Whisper on the voice node
     spec:
       automountServiceAccountToken: false
@@ -131,7 +131,7 @@ spec:
             - {name: HOME, value: /tmp}
             - {name: XDG_CACHE_HOME, value: /tmp/cache}
             - {name: HERMES_TTS_PORT, value: "9001"}
-            - {name: HERMES_TTS_VOICE, value: en_US-lessac-medium}
+            - {name: HERMES_TTS_VOICE, value: en_US-amy-medium}
             - {name: HERMES_TTS_CACHE, value: /opt/models/piper}
             - {name: HERMES_TTS_ONNX_THREADS, value: "2"}
           startupProbe:
@@ -160,7 +160,11 @@ spec:
           volumeMounts:
             - {name: tmp, mountPath: /tmp}
           resources:
-            requests: {cpu: "1", memory: 512Mi}
+            # Preloading all three policy voices measured ~243MB RSS versus
+            # ~88MB for one; bump the request from 512Mi to 768Mi to cover
+            # that ~155MB increase with headroom. The 2Gi limit already had
+            # ample slack and is unchanged.
+            requests: {cpu: "1", memory: 768Mi}
             limits: {cpu: "4", memory: 2Gi}
       volumes:
         - name: tmp

testing/tests/test_hermes_chat_quality.py

@@ -398,13 +398,27 @@ def test_voice_models_are_baked_and_runtime_has_no_public_egress():
     assert "HERMES_STT_CACHE=/opt/models/whisper" in stt_dockerfile
     assert "ADD --checksum=sha256:4cabf7c3" in tts_dockerfile
     assert "ADD --checksum=sha256:db42b97d" in tts_dockerfile
-    assert tts_dockerfile.count("--chmod=0444") == 6
+    assert "ADD --checksum=sha256:b3a6e47b57b8c7fbe6a0ce2518161a50f59a9cdd8a50835c02cb02bdd6206c18" in tts_dockerfile
+    assert "ADD --checksum=sha256:95a23eb4d42909d38df73bb9ac7f45f597dbfcde2d1bf9526fdeaf5466977d77" in tts_dockerfile
+    assert "ADD --checksum=sha256:8ff38212d23da300bbe3705c645e6e5b9475f0bfde01558eb17813e22acaaaaa" in tts_dockerfile
+    assert "ADD --checksum=sha256:c2ec28bb38e2b59e93b959b3e40348c1afebbd272f30fed5d41205d08e98a9d7" in tts_dockerfile
+    assert "ADD --checksum=sha256:3ef40a71ea63852cd8ab7e6fa7d2ecdcfa67a0b47c9c48e3f10e02ee02083ea0" in tts_dockerfile
+    assert "ADD --checksum=sha256:1afc81f703c0e4cb3b4d7c0dca096b8b54a98806807f0170cf5eb5557723c12d" in tts_dockerfile
+    assert tts_dockerfile.count("--chmod=0444") == 12
+    assert "/opt/models/piper/en_US-amy-medium.onnx" in tts_dockerfile
+    assert "/opt/models/piper/ru_RU-irina-medium.onnx" in tts_dockerfile
+    assert "/opt/models/piper/es_MX-claude-high.onnx" in tts_dockerfile
     assert "chmod 0555 /opt/models /opt/models/piper" in tts_dockerfile
     assert "HERMES_TTS_CACHE=/opt/models/piper" in tts_dockerfile
+    assert "HERMES_TTS_VOICE=en_US-amy-medium" in tts_dockerfile
     tts_server = (ROOT / "dockerfiles" / "hermes-jetson-tts-server.py").read_text()
     assert "download_voice" not in tts_server
     assert "baked Piper voice is missing" in tts_server
-    assert "session_options.intra_op_num_threads = ONNX_THREADS" in tts_server
+    assert "session_options.intra_op_num_threads = threads" in tts_server
+    assert 'LANGUAGE_VOICE_MAP = {' in tts_server
+    assert '"en": "en_US-amy-medium"' in tts_server
+    assert '"ru": "ru_RU-irina-medium"' in tts_server
+    assert '"es": "es_MX-claude-high"' in tts_server
 
     policies = _documents(HERMES / "networkpolicy.yaml")
     voice_policy = next(
@@ -449,8 +463,9 @@ def test_voice_workloads_have_deliberate_xavier_placement():
     tts_env = {
         item["name"]: item["value"] for item in tts["containers"][0]["env"]
     }
-    assert tts_env["HERMES_TTS_VOICE"] == "en_US-lessac-medium"
+    assert tts_env["HERMES_TTS_VOICE"] == "en_US-amy-medium"
     assert tts_env["HERMES_TTS_ONNX_THREADS"] == "2"
     assert tts["containers"][0]["resources"]["limits"]["cpu"] == "4"
+    assert tts["containers"][0]["resources"]["requests"]["memory"] == "768Mi"
     assert all("hostPath" not in volume for volume in stt["volumes"])
     assert all("hostPath" not in volume for volume in tts["volumes"])

testing/tests/test_hermes_tts_language_routing.py

@@ -0,0 +1,194 @@
+"""Language allow-list contracts for the private Hermes chat TTS voice policy."""
+
+from __future__ import annotations
+
+import importlib.util
+import sys
+from types import SimpleNamespace
+
+import pytest
+
+from testing.tests.test_hermes_chat_support import ROOT
+
+AMY = "en_US-amy-medium"
+IRINA = "ru_RU-irina-medium"
+CLAUDE = "es_MX-claude-high"
+
+
+def _load_tts_server(monkeypatch):
+    server_path = ROOT / "dockerfiles" / "hermes-jetson-tts-server.py"
+    spec = importlib.util.spec_from_file_location("hermes_jetson_tts_server", server_path)
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+
+    class _FakeSessionOptions:
+        def __init__(self) -> None:
+            self.intra_op_num_threads = None
+            self.inter_op_num_threads = None
+
+    fake_onnxruntime = SimpleNamespace(
+        SessionOptions=_FakeSessionOptions,
+        InferenceSession=lambda *a, **k: SimpleNamespace(),
+    )
+    fake_piper = SimpleNamespace(
+        PiperConfig=SimpleNamespace(from_dict=lambda d: d),
+        PiperVoice=lambda **kwargs: SimpleNamespace(**kwargs),
+        SynthesisConfig=lambda **kwargs: SimpleNamespace(**kwargs),
+    )
+    monkeypatch.setitem(sys.modules, "onnxruntime", fake_onnxruntime)
+    monkeypatch.setitem(sys.modules, "piper", fake_piper)
+    spec.loader.exec_module(module)
+    return module
+
+
+@pytest.fixture
+def tts(monkeypatch):
+    return _load_tts_server(monkeypatch)
+
+
+@pytest.mark.parametrize(
+    "language,expected",
+    [
+        ("en", AMY),
+        ("en-US", AMY),
+        ("en_US", AMY),
+        ("EN", AMY),
+        ("En-Us", AMY),
+        ("ru", IRINA),
+        ("ru-RU", IRINA),
+        ("ru_RU", IRINA),
+        ("RU", IRINA),
+        ("es", CLAUDE),
+        ("es-MX", CLAUDE),
+        ("es_MX", CLAUDE),
+        ("es-ES", CLAUDE),
+        ("es_ES", CLAUDE),
+        ("ES", CLAUDE),
+    ],
+)
+def test_allow_listed_languages_resolve_to_the_approved_voice(tts, language, expected):
+    assert tts.resolve_voice_name(language) == expected
+
+
+@pytest.mark.parametrize(
+    "language",
+    [
+        None,
+        "",
+        "  ",
+        "fr",
+        "fr-FR",
+        "de-DE",
+        "xx",
+        "en-GB",
+        "es-AR",
+        "english",
+        123,
+        1.5,
+        True,
+        [],
+        {},
+        {"lang": "ru"},
+        "../../etc/passwd",
+        "en_US-amy-medium/../../ru_RU-irina-medium",
+        "\x00ru",
+        "ru\x00",
+    ],
+)
+def test_unknown_missing_or_malformed_language_falls_back_to_amy(tts, language):
+    assert tts.resolve_voice_name(language) == AMY
+
+
+def test_default_voice_name_matches_the_dockerfile_env_default(tts):
+    assert tts.DEFAULT_VOICE_NAME == AMY
+
+
+def test_resolved_voice_is_always_one_of_the_three_baked_names(tts):
+    assert tts.BAKED_VOICE_NAMES == frozenset({AMY, IRINA, CLAUDE})
+    fuzz_inputs = [
+        "en", "ru", "es", "unknown", "", None, 42, "../../../etc/shadow",
+        "en_US-amy-medium\x00; rm -rf /", "RU-ru", "Es-Es", "en-us-extra",
+    ]
+    for value in fuzz_inputs:
+        assert tts.resolve_voice_name(value) in tts.BAKED_VOICE_NAMES
+
+
+def test_client_voice_field_cannot_override_the_language_policy(tts, monkeypatch):
+    """The POST handler must select the voice from "language" only.
+
+    A malicious or stale "voice" field in the request body (e.g. the legacy
+    hardcoded value the WebUI patch still sends) must never change which
+    baked model answers the request.
+    """
+    calls = []
+
+    class _RecordingHandler(tts.SpeechHandler):
+        def __init__(self, payload):
+            self._payload = payload
+            self.server = SimpleNamespace(
+                voices={
+                    AMY: SimpleNamespace(name=AMY),
+                    IRINA: SimpleNamespace(name=IRINA),
+                    CLAUDE: SimpleNamespace(name=CLAUDE),
+                },
+                default_voice_name=AMY,
+            )
+
+        def resolve(self):
+            voice_name = tts.resolve_voice_name(self._payload.get("language"))
+            calls.append(voice_name)
+            return voice_name
+
+    # A payload that supplies an attacker/legacy "voice" value but no
+    # language must resolve to the safe default, never the "voice" value.
+    handler = _RecordingHandler({"input": "hi", "voice": IRINA})
+    assert handler.resolve() == AMY
+
+    # A payload supplying both must still be governed by "language" alone.
+    handler = _RecordingHandler({"input": "hi", "voice": CLAUDE, "language": "ru"})
+    assert handler.resolve() == IRINA
+
+    assert calls == [AMY, IRINA]
+
+
+def test_no_client_string_reaches_a_filesystem_path(tts):
+    """resolve_voice_name must only ever return a fixed, baked literal.
+
+    This is the property that keeps a client from ever causing the server to
+    build a Path out of attacker-controlled text: the return value is always
+    a member of the fixed allow-list, regardless of input shape.
+    """
+    hostile_inputs = [
+        "../../../../etc/passwd",
+        "/etc/passwd",
+        "en_US-amy-medium/../../../etc/passwd",
+        "ru_RU-irina-medium\x00.onnx",
+        "es_MX-claude-high; cat /etc/shadow",
+        "\n\ren",
+        "en" + "/" * 200,
+        " ",
+    ]
+    for value in hostile_inputs:
+        result = tts.resolve_voice_name(value)
+        assert result in tts.BAKED_VOICE_NAMES
+        assert "/" not in result
+        assert ".." not in result
+        assert "\x00" not in result
+
+
+def test_normalize_language_rejects_non_string_input(tts):
+    assert tts.normalize_language(None) is None
+    assert tts.normalize_language(123) is None
+    assert tts.normalize_language([]) is None
+    assert tts.normalize_language("") is None
+    assert tts.normalize_language("   ") is None
+    assert tts.normalize_language("En_US") == "en-us"
+
+
+def test_default_voice_name_is_one_of_the_baked_voices(tts):
+    assert tts.DEFAULT_VOICE_NAME in tts.BAKED_VOICE_NAMES
+
+
+def test_load_voices_fails_closed_when_a_baked_model_is_missing(tts, tmp_path):
+    with pytest.raises(RuntimeError, match="baked Piper voice is missing"):
+        tts.load_voices(tmp_path, threads=1)