diff --git a/testing/fixtures/hermes-webui-0.52.181/static/panels.js b/testing/fixtures/hermes-webui-0.52.181/static/panels.js
new file mode 100644
index 00000000..856dd8b1
--- /dev/null
+++ b/testing/fixtures/hermes-webui-0.52.181/static/panels.js
@@ -0,0 +1,44 @@
+const _SETTINGS_SPEECH_STORAGE_KEYS={
+ tts_engine:'hermes-tts-engine',
+ tts_voice:'hermes-tts-voice',
+ tts_rate:'hermes-tts-rate',
+};
+let _settingsSpeechChangedKeys=new Set();
+
+function _speechPreferencesPayloadFromUi(){
+ const payload={};
+ const ttsVoiceSel=$('settingsTtsVoice');
+ if(ttsVoiceSel) _setOwnedSpeechPayload(payload,'tts_voice',ttsVoiceSel.value||'');
+ return payload;
+}
+
+function loadSettingsPanel(){
+ const ttsEngineSel=$('settingsTtsEngine');
+ if(ttsEngineSel){
+ ttsEngineSel.onchange=function(){
+ localStorage.setItem('hermes-tts-engine',this.value);
+ window._populateTtsVoices();
+ _schedulePreferencesAutosave();
+ };
+ }
+ // Populate voice selector based on engine
+ const ttsVoiceSel=$('settingsTtsVoice');
+ window._populateTtsVoices=function(){
+ if(!ttsVoiceSel) return;
+ const engine=localStorage.getItem('hermes-tts-engine')||'browser';
+ const current=String(_speechSetting('tts_voice','hermes-tts-voice','')||'');
+ _syncSpeechPreferenceCache('tts_voice',current);
+ if(engine==='edge'){
+ const edgeVoices=[
+ {value:'en-US-AriaNeural',label:'Aria (English, Female)'},
+ ];
+ ttsVoiceSel.innerHTML='
';
+ edgeVoices.forEach(v=>ttsVoiceSel.appendChild(v));
+ }
+ };
+ if(ttsVoiceSel&&'speechSynthesis' in window){
+ window._populateTtsVoices();
+ ttsVoiceSel.onchange=function(){_markSpeechPreferenceChanged('tts_voice');localStorage.setItem('hermes-tts-voice',this.value);_schedulePreferencesAutosave();};
+ }
+ // TTS rate/pitch sliders
+}
diff --git a/testing/fixtures/hermes-webui-0.52.181/static/ui.js b/testing/fixtures/hermes-webui-0.52.181/static/ui.js
index e6546193..11ff370e 100644
--- a/testing/fixtures/hermes-webui-0.52.181/static/ui.js
+++ b/testing/fixtures/hermes-webui-0.52.181/static/ui.js
@@ -1,4 +1,15 @@
+function _buildBrowserUtterance(text, btn){
+ const utter=new SpeechSynthesisUtterance(text);
+ const savedVoice=localStorage.getItem('hermes-tts-voice');
+ const voices=speechSynthesis.getVoices();
+ if(savedVoice&&voices.length){
+ const match=voices.find(v=>v.name===savedVoice);
+ if(match) utter.voice=match;
+ }
+ return utter;
+}
function _playEdgeTtsChunked(text, btn){
+ const voice=localStorage.getItem('hermes-tts-voice')||'zh-CN-XiaoxiaoNeural';
return fetch('/api/tts',{body:JSON.stringify({text:chunk, voice:voice, rate:rate, pitch:pitch})});
}
function speakSelected(clean, btn, engine){
@@ -11,3 +22,14 @@ function speakAutomatically(clean, engine){
_playEdgeTtsChunked(clean, null);
}
}
+function registeredTts(engine, clean){
+ const _opts={
+ voice: localStorage.getItem('hermes-tts-voice')||'',
+ rate: parseFloat(localStorage.getItem('hermes-tts-rate')),
+ };
+ const autoOpts={
+ voice: localStorage.getItem('hermes-tts-voice')||'',
+ pitch: parseFloat(localStorage.getItem('hermes-tts-pitch')),
+ };
+ return [engine, clean, _opts, autoOpts];
+}
diff --git a/testing/tests/data/atlas_voice_language_probe.js b/testing/tests/data/atlas_voice_language_probe.js
new file mode 100644
index 00000000..75dcf7f6
--- /dev/null
+++ b/testing/tests/data/atlas_voice_language_probe.js
@@ -0,0 +1,463 @@
+// Deterministic browser stub that drives dockerfiles/hermes-webui-atlas-voice.js
+// through complete hands-free turns with no microphone, audio device or GPU.
+//
+// The script under test is an IIFE with no exported seams, so the only honest
+// way to assert what reaches POST /api/tts is to run it against a fake DOM and
+// fake clock and record the requests it actually makes. Usage:
+//
+// node atlas_voice_language_probe.js
+//
+// It prints one JSON object describing every scenario to stdout.
+'use strict';
+
+const fs = require('fs');
+const vm = require('vm');
+
+const SCRIPT_PATH = process.argv[2];
+if (!SCRIPT_PATH) {
+ throw new Error('usage: atlas_voice_language_probe.js ');
+}
+const SOURCE = fs.readFileSync(SCRIPT_PATH, 'utf8');
+
+function flush() {
+ // Four macrotask hops drain the promise chains the script builds around
+ // fetch()/json()/blob()/play() without ever waiting on wall-clock time.
+ return new Promise((resolve) => {
+ let hops = 0;
+ (function hop() {
+ hops += 1;
+ if (hops > 12) {
+ resolve();
+ return;
+ }
+ setImmediate(hop);
+ })();
+ });
+}
+
+function makeElement(id) {
+ return {
+ id,
+ style: {
+ values: new Map(),
+ setProperty(name, value) { this.values.set(name, String(value)); },
+ removeProperty(name) { this.values.delete(name); },
+ getPropertyValue(name) { return this.values.get(name) || ''; },
+ },
+ dataset: {},
+ attributes: {},
+ value: '',
+ textContent: '',
+ className: '',
+ classList: {
+ entries: new Set(),
+ add(name) { this.entries.add(name); },
+ remove(name) { this.entries.delete(name); },
+ contains(name) { return this.entries.has(name); },
+ },
+ listeners: [],
+ setAttribute(name, value) { this.attributes[name] = String(value); },
+ getAttribute(name) {
+ return Object.prototype.hasOwnProperty.call(this.attributes, name)
+ ? this.attributes[name] : null;
+ },
+ addEventListener(type, handler) { this.listeners.push({ type, handler }); },
+ removeEventListener(type, handler) {
+ this.listeners = this.listeners.filter((entry) => entry.handler !== handler);
+ },
+ click() {
+ const event = { preventDefault() {}, stopImmediatePropagation() {} };
+ this.listeners
+ .filter((entry) => entry.type === 'click')
+ .forEach((entry) => entry.handler(event));
+ },
+ querySelector() { return null; },
+ insertBefore() {},
+ appendChild() {},
+ };
+}
+
+function makeHarness() {
+ const clock = { now: 1000000 };
+ const timeouts = [];
+ const intervals = new Map();
+ let timerId = 1;
+
+ const ttsRequests = [];
+ const transcribeCalls = [];
+ const toasts = [];
+ const sends = [];
+
+ let capability = { ok: true, available: true, provider: 'local_command' };
+ let transcribeResponse = { ok: true, transcript: 'hello', language: 'en' };
+ let transcribeStatus = 200;
+ let assistantRows = [];
+ let loud = false;
+ let recorder = null;
+
+ const storage = new Map();
+ const elements = {};
+ ['btnVoiceMode', 'voiceModeBar', 'voiceModeIndicator', 'voiceModeLabel', 'msg']
+ .forEach((id) => { elements[id] = makeElement(id); });
+
+ function MediaRecorder() {
+ this.state = 'recording';
+ this.ondataavailable = null;
+ this.onstop = null;
+ recorder = this;
+ }
+ MediaRecorder.prototype.start = function start() { this.state = 'recording'; };
+ MediaRecorder.prototype.stop = function stop() {
+ if (this.state === 'inactive') return;
+ this.state = 'inactive';
+ if (this.onstop) this.onstop();
+ };
+ MediaRecorder.isTypeSupported = function isTypeSupported() { return true; };
+
+ function AudioContext() {
+ this.createAnalyser = () => ({
+ fftSize: 2048,
+ getByteTimeDomainData(samples) {
+ for (let i = 0; i < samples.length; i += 1) {
+ samples[i] = loud ? (i % 2 ? 200 : 56) : 128;
+ }
+ },
+ });
+ this.createBiquadFilter = () => ({
+ type: '', frequency: { value: 0 }, Q: { value: 0 }, connect() {},
+ });
+ this.createMediaStreamSource = () => ({ connect() {} });
+ this.close = () => {};
+ }
+
+ function AudioElement() {
+ this.currentTime = 0;
+ this.onended = null;
+ this.onerror = null;
+ this.pause = () => {};
+ this.play = () => {
+ setImmediate(() => { if (this.onended) this.onended(); });
+ return Promise.resolve();
+ };
+ }
+
+ async function fetchStub(url, init) {
+ if (url === '/api/transcribe/capability') {
+ return { ok: true, status: 200, json: async () => capability };
+ }
+ if (url === '/api/transcribe') {
+ transcribeCalls.push({ body: init && init.body });
+ return {
+ ok: transcribeStatus < 400,
+ status: transcribeStatus,
+ json: async () => transcribeResponse,
+ };
+ }
+ if (url === '/api/tts') {
+ ttsRequests.push(JSON.parse(init.body));
+ return {
+ ok: true,
+ status: 200,
+ blob: async () => ({ synthetic: true }),
+ json: async () => ({}),
+ };
+ }
+ throw new Error(`unexpected fetch: ${url}`);
+ }
+
+ const context = {
+ console,
+ Uint8Array,
+ Promise,
+ Math,
+ JSON,
+ String,
+ Number,
+ Error,
+ parseInt,
+ isNaN,
+ Set,
+ Map,
+ Array,
+ Object,
+ Date: { now: () => clock.now },
+ Blob: function Blob(parts, options) { this.parts = parts; this.type = (options || {}).type || ''; },
+ File: function File(parts, name, options) {
+ this.parts = parts; this.name = name; this.type = (options || {}).type || '';
+ },
+ FormData: function FormData() { this.entries = []; this.append = (k, v) => this.entries.push([k, v]); },
+ URL: { createObjectURL: () => 'blob:atlas-test', revokeObjectURL() {} },
+ Audio: AudioElement,
+ MediaRecorder,
+ AudioContext,
+ fetch: fetchStub,
+ localStorage: {
+ getItem: (key) => (storage.has(key) ? storage.get(key) : null),
+ setItem: (key, value) => { storage.set(key, String(value)); },
+ removeItem: (key) => { storage.delete(key); },
+ },
+ navigator: {
+ mediaDevices: {
+ getUserMedia: async () => ({ getTracks: () => [{ stop() {} }] }),
+ getSupportedConstraints: () => ({}),
+ },
+ },
+ document: {
+ getElementById: (id) => elements[id] || null,
+ querySelectorAll: () => assistantRows,
+ createElement: () => ({ value: '', textContent: '' }),
+ },
+ S: { session: { session_id: 'session-1' }, busy: false },
+ setTimeout: (fn, delay) => {
+ const id = timerId; timerId += 1;
+ timeouts.push({ id, fn, at: clock.now + (delay || 0) });
+ return id;
+ },
+ clearTimeout: (id) => {
+ const index = timeouts.findIndex((entry) => entry.id === id);
+ if (index >= 0) timeouts.splice(index, 1);
+ },
+ setInterval: (fn, delay) => {
+ const id = timerId; timerId += 1;
+ intervals.set(id, { fn, delay: delay || 0 });
+ return id;
+ },
+ clearInterval: (id) => { intervals.delete(id); },
+ };
+ context.window = context;
+ context.showToast = (message) => { toasts.push(message); };
+ context.send = () => { sends.push(elements.msg.value); };
+ context.autoResize = () => {};
+
+ vm.createContext(context);
+ vm.runInContext(SOURCE, context, { filename: 'atlas-voice.js' });
+
+ function runDueTimeouts() {
+ const due = timeouts.filter((entry) => entry.at <= clock.now);
+ due.forEach((entry) => {
+ const index = timeouts.indexOf(entry);
+ if (index >= 0) timeouts.splice(index, 1);
+ entry.fn();
+ });
+ }
+
+ function tick(ms) {
+ clock.now += ms;
+ Array.from(intervals.values()).forEach((entry) => entry.fn());
+ runDueTimeouts();
+ }
+
+ return {
+ context,
+ elements,
+ ttsRequests,
+ transcribeCalls,
+ toasts,
+ sends,
+ clock,
+ recorder: () => recorder,
+ setLoud: (value) => { loud = value; },
+ setCapability: (value) => { capability = value; },
+ setTranscribeResponse: (value, status) => {
+ transcribeResponse = value;
+ transcribeStatus = status === undefined ? 200 : status;
+ },
+ setAssistantReply: (text) => { assistantRows = [{ dataset: { rawText: text } }]; },
+ setSession: (id) => { context.S.session = { session_id: id }; },
+ advance: (ms) => { clock.now += ms; },
+ tick,
+ runDueTimeouts,
+ flush,
+ };
+}
+
+// Walk one capture window: pre-roll audio, three loud frames so the VAD latches
+// speech, then silence past the hangover so MediaRecorder.stop() fires.
+async function captureSpeech(harness) {
+ const active = harness.recorder();
+ if (!active) throw new Error('voice mode never created a recorder');
+ active.ondataavailable({ data: { size: 512 } });
+ harness.setLoud(true);
+ for (let i = 0; i < 4; i += 1) harness.tick(100);
+ active.ondataavailable({ data: { size: 512 } });
+ harness.setLoud(false);
+ harness.advance(2500);
+ harness.tick(100);
+ await harness.flush();
+}
+
+async function startVoiceMode(harness) {
+ await harness.flush();
+ harness.elements.btnVoiceMode.click();
+ await harness.flush();
+}
+
+// One complete turn: speak, transcribe, let the app "answer", read it back.
+async function runTurn(harness, { transcript, language, reply }) {
+ const payload = { ok: true, transcript };
+ if (language !== undefined) payload.language = language;
+ harness.setTranscribeResponse(payload);
+ await captureSpeech(harness);
+ harness.setAssistantReply(reply || 'An answer.');
+ harness.context.autoReadLastAssistant();
+ await harness.flush();
+}
+
+async function restartListening(harness) {
+ harness.advance(1000);
+ harness.runDueTimeouts();
+ await harness.flush();
+}
+
+const scenarios = {};
+
+scenarios.english_turn_speaks_english = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ await runTurn(harness, { transcript: 'What is the weather?', language: 'en' });
+ return { tts: harness.ttsRequests };
+};
+
+scenarios.russian_turn_speaks_russian = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ await runTurn(harness, { transcript: 'Как дела?', language: 'ru', reply: 'Всё хорошо.' });
+ return { tts: harness.ttsRequests };
+};
+
+scenarios.spanish_turn_speaks_spanish = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ await runTurn(harness, { transcript: '¿Qué tal?', language: 'es', reply: 'Muy bien.' });
+ return { tts: harness.ttsRequests };
+};
+
+scenarios.missing_language_falls_back = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ await runTurn(harness, { transcript: 'Hello there.' });
+ return { tts: harness.ttsRequests };
+};
+
+scenarios.unsupported_language_falls_back = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ await runTurn(harness, { transcript: 'Bonjour tout le monde.', language: 'fr' });
+ return { tts: harness.ttsRequests };
+};
+
+scenarios.hostile_language_values_are_dropped = async () => {
+ const results = [];
+ const hostile = [
+ 'ru; rm -rf /',
+ '../../ru_RU-irina-medium',
+ 'ru ',
+ 'RUSSIAN',
+ { language: 'ru' },
+ ['ru'],
+ 42,
+ null,
+ 'r',
+ 'ru ru',
+ 'x'.repeat(4096),
+ ];
+ for (const language of hostile) {
+ const harness = makeHarness();
+ // eslint-disable-next-line no-await-in-loop
+ await startVoiceMode(harness);
+ // eslint-disable-next-line no-await-in-loop
+ await runTurn(harness, { transcript: 'Say something.', language });
+ results.push({ sent: String(language), tts: harness.ttsRequests });
+ }
+ return { results };
+};
+
+scenarios.voice_field_is_never_sent = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ await runTurn(harness, { transcript: 'Как дела?', language: 'ru' });
+ return { tts: harness.ttsRequests };
+};
+
+scenarios.language_does_not_leak_into_later_turn = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ await runTurn(harness, { transcript: 'Как дела?', language: 'ru', reply: 'Всё хорошо.' });
+ await restartListening(harness);
+ await runTurn(harness, { transcript: 'And in English?', language: undefined });
+ await restartListening(harness);
+ await runTurn(harness, { transcript: '¿Y ahora?', language: 'es' });
+ return { tts: harness.ttsRequests };
+};
+
+scenarios.empty_transcript_does_not_arm_a_language = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ harness.setTranscribeResponse({ ok: true, transcript: ' ', language: 'ru' });
+ await captureSpeech(harness);
+ const sendsAfterBlank = harness.sends.slice();
+ // A reply landing while the blank turn winds down must not inherit a
+ // language that transcript never earned.
+ harness.setAssistantReply('A stray answer.');
+ harness.context.autoReadLastAssistant();
+ await harness.flush();
+ await restartListening(harness);
+ await runTurn(harness, { transcript: 'Hello.', language: undefined });
+ return { sendsAfterBlank, tts: harness.ttsRequests, sends: harness.sends };
+};
+
+scenarios.session_change_discards_language = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ harness.setTranscribeResponse({ ok: true, transcript: 'Как дела?', language: 'ru' });
+ await captureSpeech(harness);
+ harness.setSession('session-2');
+ harness.setAssistantReply('Reply that belongs to another chat.');
+ harness.context.autoReadLastAssistant();
+ await harness.flush();
+ const afterSwitch = harness.ttsRequests.slice();
+ await restartListening(harness);
+ await runTurn(harness, { transcript: 'Hello again.', language: undefined });
+ return { afterSwitch, tts: harness.ttsRequests };
+};
+
+scenarios.deactivation_discards_language = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ harness.setTranscribeResponse({ ok: true, transcript: 'Как дела?', language: 'ru' });
+ await captureSpeech(harness);
+ harness.elements.btnVoiceMode.click();
+ await harness.flush();
+ harness.setAssistantReply('Late reply after the user left voice mode.');
+ harness.context.autoReadLastAssistant();
+ await harness.flush();
+ const afterDeactivate = harness.ttsRequests.slice();
+ harness.elements.btnVoiceMode.click();
+ await harness.flush();
+ await runTurn(harness, { transcript: 'Fresh start.', language: undefined });
+ return { afterDeactivate, tts: harness.ttsRequests };
+};
+
+scenarios.transcribe_error_speaks_nothing = async () => {
+ const harness = makeHarness();
+ await startVoiceMode(harness);
+ harness.setTranscribeResponse({ error: 'Whisper is down' }, 503);
+ await captureSpeech(harness);
+ harness.setAssistantReply('Some earlier answer.');
+ harness.context.autoReadLastAssistant();
+ await harness.flush();
+ return { tts: harness.ttsRequests, toasts: harness.toasts };
+};
+
+(async () => {
+ const output = {};
+ const names = Object.keys(scenarios);
+ for (const name of names) {
+ // eslint-disable-next-line no-await-in-loop
+ output[name] = await scenarios[name]();
+ }
+ process.stdout.write(JSON.stringify(output, null, 1));
+})().catch((error) => {
+ process.stderr.write(String((error && error.stack) || error));
+ process.exit(1);
+});
diff --git a/testing/tests/test_hermes_chat_quality.py b/testing/tests/test_hermes_chat_quality.py
index 4cf45ffb..425944ad 100644
--- a/testing/tests/test_hermes_chat_quality.py
+++ b/testing/tests/test_hermes_chat_quality.py
@@ -343,7 +343,7 @@ def test_chat_voice_uses_private_jetson_services_and_shared_auto_route():
assert "/api/tts" in voice_script
assert "speakResponse(generation)" in voice_script
assert "window._splitForTTS(text,280)" in voice_script
- assert "pending=fetchSpeech(chunks[index+1])" in voice_script
+ assert "pending=fetchSpeech(chunks[index+1],language)" in voice_script
assert "restartSoon(token,450)" in voice_script
assert "constraints.voiceIsolation=true" in voice_script
assert "highpass.frequency.value=140" in voice_script
@@ -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(
diff --git a/testing/tests/test_hermes_tts_language_routing.py b/testing/tests/test_hermes_tts_language_routing.py
new file mode 100644
index 00000000..5a53f808
--- /dev/null
+++ b/testing/tests/test_hermes_tts_language_routing.py
@@ -0,0 +1,220 @@
+"""Language allow-list contracts for the private Hermes chat TTS voice policy."""
+
+from __future__ import annotations
+
+import importlib.util
+import io
+import json
+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 frozenset({AMY, IRINA, CLAUDE}) == tts.BAKED_VOICE_NAMES
+ 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):
+ """The POST handler must select the voice from "language" only.
+
+ A malicious or stale "voice" field in a hostile/legacy request must never
+ change which baked model answers the request.
+ """
+ calls: list[str] = []
+
+ class _RecordingVoice:
+ def __init__(self, name: str) -> None:
+ self.name = name
+
+ def synthesize_wav(self, text, wav_file, syn_config) -> None:
+ calls.append(self.name)
+ wav_file.setnchannels(1)
+ wav_file.setsampwidth(2)
+ wav_file.setframerate(16_000)
+ wav_file.writeframes(b"\x00\x00")
+
+ class _RecordingHandler(tts.SpeechHandler):
+ def __init__(self, payload):
+ request = json.dumps(payload).encode("utf-8")
+ self.path = "/v1/audio/speech"
+ self.headers = {"Content-Length": str(len(request))}
+ self.rfile = io.BytesIO(request)
+ self.wfile = io.BytesIO()
+ self.status = None
+ self.response_headers = {}
+ self.server = SimpleNamespace(
+ voices={
+ AMY: _RecordingVoice(AMY),
+ IRINA: _RecordingVoice(IRINA),
+ CLAUDE: _RecordingVoice(CLAUDE),
+ },
+ default_voice_name=AMY,
+ )
+
+ def send_response(self, status, message=None):
+ self.status = status
+
+ def send_header(self, name, value):
+ self.response_headers[name] = value
+
+ def end_headers(self):
+ return None
+
+ # 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})
+ handler.do_POST()
+ assert handler.status == 200
+ assert handler.response_headers["X-TTS-Voice"] == AMY
+
+ # A payload supplying both must still be governed by "language" alone.
+ handler = _RecordingHandler({"input": "hi", "voice": CLAUDE, "language": "ru"})
+ handler.do_POST()
+ assert handler.status == 200
+ assert handler.response_headers["X-TTS-Voice"] == 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)
diff --git a/testing/tests/test_hermes_voice_instrument.py b/testing/tests/test_hermes_voice_instrument.py
index eada0199..62f9e489 100644
--- a/testing/tests/test_hermes_voice_instrument.py
+++ b/testing/tests/test_hermes_voice_instrument.py
@@ -12,6 +12,7 @@ import sys
ROOT = Path(__file__).resolve().parents[2]
FIXTURE = ROOT / "testing/fixtures/hermes-webui-0.52.181"
+AGENT_FIXTURE = ROOT / "testing/fixtures/hermes-agent"
PATCHER = ROOT / "dockerfiles/hermes-webui-atlas-patch.py"
VOICE_JS = ROOT / "dockerfiles/hermes-webui-atlas-voice.js"
VOICE_CSS = ROOT / "dockerfiles/hermes-webui-atlas-voice.css"
@@ -20,9 +21,12 @@ DOM_PROBE = ROOT / "testing/probes/hermes_voice_instrument_probe.js"
def _patched_fixture(tmp_path: Path) -> Path:
target = tmp_path / "hermes-webui"
+ agent_target = tmp_path / "hermes-agent"
shutil.copytree(FIXTURE, target)
+ shutil.copytree(AGENT_FIXTURE, agent_target)
env = os.environ.copy()
env["HERMES_WEBUI_PATCH_ROOT"] = str(target)
+ env["HERMES_AGENT_PATCH_ROOT"] = str(agent_target)
subprocess.run(
[sys.executable, str(PATCHER)],
cwd=ROOT,
@@ -57,6 +61,52 @@ def test_real_upstream_fixture_receives_visual_instrument_contract(tmp_path: Pat
assert ''
+ ''
+ '\n'
+ '\n'
+)
+
+UI_FIXTURE = """function _playEdgeTtsChunked(text, btn){
+ fetch('/api/tts',{method:'POST',body:JSON.stringify({text:chunk, voice:voice, rate:rate, pitch:pitch})});
+}
+function readAloud(clean, btn, engine){
+ if(engine==='edge'){
+ _playEdgeTtsChunked(clean, btn);
+ }
+}
+function autoRead(clean, engine){
+ if(engine==='edge'){
+ _playEdgeTtsChunked(clean, null);
+ }
+}
+"""
+
+HELPERS_FIXTURE = '''"""Stand-in for the WebUI helper module the patched code imports."""
+
+
+def bad(handler, message, status=400):
+ return {"status": status, "error": message}
+'''
+
+ROUTES_FIXTURE = '''"""Stand-in carrying the exact upstream anchors the Atlas TTS patch pins."""
+
+import json
+import os
+from urllib.request import ProxyHandler, Request, build_opener
+
+
+class _NoRedirectTtsHandler:
+ """Placeholder for the upstream no-redirect opener handler."""
+
+
+class _Logger:
+ def __init__(self):
+ self.failures = []
+
+ def exception(self, message):
+ self.failures.append(message)
+
+
+logger = _Logger()
+UPSTREAM_REQUESTS = []
+
+
+class _Upstream:
+ def __init__(self, payload):
+ self._payload = payload
+
+ def read(self):
+ return self._payload
+
+ def __enter__(self):
+ return self
+
+ def __exit__(self, *exc_info):
+ return False
+
+
+def _buffer_tts_audio_response(response):
+ return response.read()
+
+
+def _tts_open(req, *, timeout=30, opener_factory=None):
+ """Thin network seam for the TTS upstream fetch so tests can intercept it."""
+ UPSTREAM_REQUESTS.append(json.loads(req.data.decode("utf-8")))
+ return _Upstream(b"RIFFsynthetic")
+
+
+def _handle_tts(handler, data, text, rate_str, engine):
+ # ── ElevenLabs TTS ──────────────────────────────────────────────────
+ return None
+'''
+
+UPLOAD_FIXTURE = '''"""Stand-in carrying the exact upstream /api/transcribe response anchor."""
+
+
+def j(handler, payload, status=200):
+ return {"status": status, "payload": payload}
+
+
+def handle_transcribe(handler, result):
+ try:
+ transcript = str(result.get('transcript') or '').strip()
+ return j(handler, {'ok': True, 'transcript': transcript})
+ except ValueError as error:
+ return j(handler, {'error': str(error)}, status=400)
+'''
+
+TRANSCRIPTION_FIXTURE = '''"""Stand-in carrying the exact upstream local-command STT envelope anchor."""
+
+import contextlib
+from pathlib import Path
+
+
+class _Logger:
+ def info(self, *args):
+ return None
+
+
+logger = _Logger()
+
+
+def _transcribe_local_command(file_path, normalized_model, output_dir):
+ try:
+ with contextlib.nullcontext(output_dir):
+ txt_files = sorted(Path(output_dir).glob("*.txt"))
+ transcript_text = txt_files[0].read_text(encoding="utf-8").strip()
+ logger.info(
+ "Transcribed %s via local STT command (%s, %d chars)",
+ Path(file_path).name,
+ normalized_model,
+ len(transcript_text),
+ )
+ return {"success": True, "transcript": transcript_text, "provider": "local_command"}
+ except OSError as error:
+ return {"success": False, "transcript": "", "error": str(error)}
+'''
+
+
+def _write(path: Path, text: str) -> None:
+ path.parent.mkdir(parents=True, exist_ok=True)
+ path.write_text(text, encoding="utf-8")
+
+
+@pytest.fixture
+def patched_webui(tmp_path, monkeypatch):
+ """Apply the real Atlas image patch to fixture trees and import the result."""
+ webui = tmp_path / "hermes-webui"
+ agent = tmp_path / "hermes"
+ # Start with the pinned full-surface fixture introduced by PR #39 so this
+ # test proves the language pipeline composes with its voice-selector removal
+ # and conversation instrument, not merely with the older #27 anchors.
+ shutil.copytree(WEBUI_FIXTURE, webui)
+ _write(webui / "api" / "__init__.py", "")
+ _write(webui / "api" / "helpers.py", HELPERS_FIXTURE)
+ _write(webui / "api" / "routes.py", ROUTES_FIXTURE)
+ _write(webui / "api" / "upload.py", UPLOAD_FIXTURE)
+ _write(agent / "tools" / "__init__.py", "")
+ _write(agent / "tools" / "transcription_tools.py", TRANSCRIPTION_FIXTURE)
+
+ environment = dict(os.environ)
+ environment["HERMES_WEBUI_PATCH_ROOT"] = str(webui)
+ environment["HERMES_AGENT_PATCH_ROOT"] = str(agent)
+ completed = subprocess.run(
+ [sys.executable, str(ATLAS_PATCH)],
+ env=environment,
+ capture_output=True,
+ text=True,
+ )
+ assert completed.returncode == 0, completed.stderr or completed.stdout
+
+ for name in ("api", "api.helpers", "api.routes", "api.upload", "tools",
+ "tools.transcription_tools"):
+ sys.modules.pop(name, None)
+ monkeypatch.syspath_prepend(str(agent))
+ monkeypatch.syspath_prepend(str(webui))
+ import api.routes as routes # noqa: PLC0415
+ import api.upload as upload # noqa: PLC0415
+ import tools.transcription_tools as transcription # noqa: PLC0415
+
+ yield SimpleNamespace(
+ webui=webui,
+ agent=agent,
+ routes=routes,
+ upload=upload,
+ transcription=transcription,
+ )
+
+ for name in ("api", "api.helpers", "api.routes", "api.upload", "tools",
+ "tools.transcription_tools"):
+ sys.modules.pop(name, None)
+
+
+class _Handler:
+ """Just enough BaseHTTPRequestHandler surface for the Atlas TTS branch."""
+
+ def __init__(self):
+ self.status = None
+ self.headers_sent = {}
+ self.wfile = SimpleNamespace(write=self._write)
+ self.body = b""
+
+ def send_response(self, status):
+ self.status = status
+
+ def send_header(self, name, value):
+ self.headers_sent[name] = value
+
+ def end_headers(self):
+ return None
+
+ def _write(self, payload):
+ self.body += payload
+
+
+def _atlas_tts(patched, monkeypatch, data):
+ """Run the patched Atlas branch and return the JSON it sent to hermes-tts."""
+ monkeypatch.setenv("HERMES_WEBUI_ATLAS_TTS_URL", ATLAS_TTS_URL)
+ patched.routes.UPSTREAM_REQUESTS.clear()
+ handler = _Handler()
+ result = patched.routes._handle_tts(handler, data, "Some reply.", "", "atlas")
+ assert result is True, "the Atlas branch must own the response"
+ assert handler.status == 200
+ assert len(patched.routes.UPSTREAM_REQUESTS) == 1
+ return patched.routes.UPSTREAM_REQUESTS[0]
+
+
+# ---------------------------------------------------------------------------
+# 1. Whisper service reports the language it actually decoded with
+# ---------------------------------------------------------------------------
+
+
+def test_stt_response_carries_whisper_detected_language(monkeypatch):
+ module = _load_stt_server(monkeypatch)
+ payload = module._transcription_payload(
+ {
+ "language": "ru",
+ "segments": [
+ {"text": " Как дела?", "no_speech_prob": 0.1, "avg_logprob": -0.2}
+ ],
+ }
+ )
+ assert payload["text"] == "Как дела?"
+ assert payload["language"] == "ru"
+ assert payload["model"] == module.MODEL_NAME
+
+
+@pytest.mark.parametrize(
+ ("raw", "expected"),
+ [
+ ("en", "en"),
+ ("RU", "ru"),
+ (" es ", "es"),
+ ("yue", "yue"),
+ ("fr", "fr"),
+ ("en-US", ""),
+ ("en_US", ""),
+ ("e", ""),
+ ("english", ""),
+ ("", ""),
+ ("../en", ""),
+ ("en\x00", ""),
+ ("ru; rm -rf /", ""),
+ ("рус", ""),
+ (None, ""),
+ (7, ""),
+ (["ru"], ""),
+ ({"language": "ru"}, ""),
+ ],
+)
+def test_stt_language_field_is_shape_validated(monkeypatch, raw, expected):
+ module = _load_stt_server(monkeypatch)
+ assert module._detected_language({"language": raw}) == expected
+
+
+def test_stt_language_absent_when_whisper_omits_it(monkeypatch):
+ module = _load_stt_server(monkeypatch)
+ assert module._detected_language({}) == ""
+ assert module._detected_language("not a result") == ""
+ assert module._transcription_payload({"text": "hi"})["language"] == ""
+
+
+# ---------------------------------------------------------------------------
+# 2. The local-command client carries the language without breaking the
+# .txt contract Hermes reads the transcript from
+# ---------------------------------------------------------------------------
+
+
+def test_stt_client_writes_language_sidecar_beside_the_txt_contract(tmp_path):
+ module = _load_stt_client()
+ module._write_result(tmp_path, "voice-input", "Как дела?", "ru")
+ assert (tmp_path / "voice-input.txt").read_text(encoding="utf-8") == "Как дела?"
+ assert (tmp_path / "voice-input.language").read_text(encoding="utf-8") == "ru"
+ # Hermes globs *.txt and reads the first match: the sidecar must not join it.
+ assert sorted(p.name for p in tmp_path.glob("*.txt")) == ["voice-input.txt"]
+
+
+def test_stt_client_omits_the_sidecar_when_no_language_was_detected(tmp_path):
+ module = _load_stt_client()
+ module._write_result(tmp_path, "voice-input", "Hello.", "")
+ assert (tmp_path / "voice-input.txt").exists()
+ assert not (tmp_path / "voice-input.language").exists()
+
+
+@pytest.mark.parametrize(
+ ("raw", "expected"),
+ [
+ ("en", "en"),
+ ("ES", "es"),
+ (" ru ", "ru"),
+ ("en-US", ""),
+ ("", ""),
+ ("../../etc/passwd", ""),
+ ("en\n", "en"),
+ ("e", ""),
+ (None, ""),
+ (12, ""),
+ (["en"], ""),
+ ],
+)
+def test_stt_client_normalises_the_service_language_field(raw, expected):
+ module = _load_stt_client()
+ assert module._normalize_language(raw) == expected
+
+
+# ---------------------------------------------------------------------------
+# 3. The patched agent envelope and /api/transcribe response carry it through
+# ---------------------------------------------------------------------------
+
+
+def test_patched_local_command_envelope_carries_the_sidecar_language(patched_webui, tmp_path):
+ output = tmp_path / "stt-out"
+ output.mkdir()
+ (output / "voice-input.txt").write_text("Как дела?", encoding="utf-8")
+ (output / "voice-input.language").write_text("ru\n", encoding="utf-8")
+ result = patched_webui.transcription._transcribe_local_command(
+ "/tmp/voice-input.wav", "small", output
+ )
+ assert result == {
+ "success": True,
+ "transcript": "Как дела?",
+ "provider": "local_command",
+ "language": "ru",
+ }
+
+
+def test_patched_local_command_envelope_defaults_to_no_language(patched_webui, tmp_path):
+ output = tmp_path / "stt-out"
+ output.mkdir()
+ (output / "voice-input.txt").write_text("Hello.", encoding="utf-8")
+ result = patched_webui.transcription._transcribe_local_command(
+ "/tmp/voice-input.wav", "small", output
+ )
+ assert result["transcript"] == "Hello."
+ assert result["language"] == ""
+
+
+@pytest.mark.parametrize(
+ "hostile",
+ ["en-US", "../../en", "en; rm -rf /", "e", "english", "", "\x00en", "e n"],
+)
+def test_patched_local_command_envelope_rejects_malformed_sidecars(
+ patched_webui, tmp_path, hostile
+):
+ output = tmp_path / "stt-out"
+ output.mkdir()
+ (output / "voice-input.txt").write_text("Hello.", encoding="utf-8")
+ (output / "voice-input.language").write_text(hostile, encoding="utf-8")
+ result = patched_webui.transcription._transcribe_local_command(
+ "/tmp/voice-input.wav", "small", output
+ )
+ assert result["language"] == ""
+
+
+def test_patched_local_command_envelope_survives_an_undecodable_sidecar(
+ patched_webui, tmp_path
+):
+ """A corrupt sidecar must cost the language, never the transcript."""
+ output = tmp_path / "stt-out"
+ output.mkdir()
+ (output / "voice-input.txt").write_text("Hello.", encoding="utf-8")
+ (output / "voice-input.language").write_bytes(b"\xff\xfe\x00ru")
+ result = patched_webui.transcription._transcribe_local_command(
+ "/tmp/voice-input.wav", "small", output
+ )
+ assert result["success"] is True
+ assert result["transcript"] == "Hello."
+ assert result["language"] == ""
+
+
+def test_patched_transcribe_response_reports_the_language(patched_webui):
+ response = patched_webui.upload.handle_transcribe(
+ None, {"success": True, "transcript": " Как дела? ", "language": "ru"}
+ )
+ assert response["payload"] == {"ok": True, "transcript": "Как дела?", "language": "ru"}
+
+
+@pytest.mark.parametrize(
+ "hostile",
+ ["", None, "en-US", "englishhh", "../en", 5, ["ru"], {"a": "b"}, "e"],
+)
+def test_patched_transcribe_response_blanks_untrusted_languages(patched_webui, hostile):
+ response = patched_webui.upload.handle_transcribe(
+ None, {"success": True, "transcript": "Hello.", "language": hostile}
+ )
+ assert response["payload"]["language"] == ""
+ assert response["payload"]["transcript"] == "Hello."
+
+
+# ---------------------------------------------------------------------------
+# 4. The /api/tts trust boundary: allow-list only, and never `voice`
+# ---------------------------------------------------------------------------
+
+
+@pytest.mark.parametrize("language", SUPPORTED)
+def test_atlas_tts_forwards_allow_listed_languages(patched_webui, monkeypatch, language):
+ body = _atlas_tts(patched_webui, monkeypatch, {"engine": "atlas", "language": language})
+ assert body["language"] == language
+ assert body["model"] == "piper"
+
+
+@pytest.mark.parametrize(
+ "hostile",
+ [
+ "fr",
+ "de",
+ "",
+ None,
+ "EN-GB",
+ " RU ",
+ "ru-RU",
+ "es_MX",
+ "../../ru_RU-irina-medium",
+ "ru; rm -rf /",
+ "ru\x00",
+ "ру",
+ 5,
+ ["ru"],
+ {"language": "ru"},
+ True,
+ "x" * 8192,
+ ],
+)
+def test_atlas_tts_omits_untrusted_languages(
+ patched_webui, monkeypatch, hostile
+):
+ body = _atlas_tts(patched_webui, monkeypatch, {"engine": "atlas", "language": hostile})
+ assert "language" not in body
+
+
+def test_atlas_tts_language_is_absent_when_the_client_sends_none(patched_webui, monkeypatch):
+ body = _atlas_tts(patched_webui, monkeypatch, {"engine": "atlas"})
+ assert "language" not in body
+
+
+def test_atlas_tts_voice_field_cannot_steer_synthesis(patched_webui, monkeypatch):
+ body = _atlas_tts(
+ patched_webui,
+ monkeypatch,
+ {"engine": "atlas", "voice": "ru_RU-irina-medium", "language": "en"},
+ )
+ assert body["language"] == "en"
+ body = _atlas_tts(
+ patched_webui,
+ monkeypatch,
+ {"engine": "atlas", "voice": "es_MX-claude-high"},
+ )
+ assert "language" not in body
+
+
+def test_atlas_tts_language_helper_only_ever_returns_a_baked_voice_language(patched_webui):
+ resolve = patched_webui.routes._atlas_tts_language
+ hostile = [
+ None, 0, 1, -1, True, False, [], {}, set(), object(), b"ru",
+ "", " ", "\t\n", "en", "EN", "en-US", "en_us", "ru-RU", "es-MX",
+ "e", "eng", "english", "ru ru", "ru;es", "../ru", "ru\x00", "ру",
+ "x" * 65536, "en" * 4096,
+ ]
+ for value in hostile:
+ expected = value if value in SUPPORTED else ""
+ assert resolve({"language": value}) == expected
+ for value in hostile:
+ assert resolve(value) == ""
+ assert resolve({"voice": "ru_RU-irina-medium"}) == ""
+
+
+def test_manual_tts_button_body_still_carries_no_language(patched_webui):
+ """The read-aloud button has no trusted STT signal, so it must stay Amy."""
+ ui = (patched_webui.webui / "static" / "ui.js").read_text(encoding="utf-8")
+ assert "engine:engineOverride||'edge'" in ui
+ assert "language" not in ui
+
+
+# ---------------------------------------------------------------------------
+# 5. Browser contract, driven through the real atlas-voice.js
+# ---------------------------------------------------------------------------
+
+
+@pytest.fixture(scope="module")
+def voice_probe():
+ node = shutil.which("node")
+ if not node:
+ pytest.skip("node is required to drive the browser voice-mode contract")
+ completed = subprocess.run(
+ [node, str(VOICE_PROBE), str(VOICE_SCRIPT)],
+ capture_output=True,
+ text=True,
+ timeout=180,
+ )
+ assert completed.returncode == 0, completed.stderr
+ return json.loads(completed.stdout)
+
+
+@pytest.mark.parametrize(
+ ("scenario", "expected"),
+ [
+ ("english_turn_speaks_english", "en"),
+ ("russian_turn_speaks_russian", "ru"),
+ ("spanish_turn_speaks_spanish", "es"),
+ ],
+)
+def test_voice_mode_speaks_the_language_whisper_detected(voice_probe, scenario, expected):
+ requests = voice_probe[scenario]["tts"]
+ assert requests, "voice mode never reached /api/tts"
+ for request in requests:
+ assert request["engine"] == "atlas"
+ assert request["language"] == expected
+
+
+@pytest.mark.parametrize(
+ "scenario", ["missing_language_falls_back", "unsupported_language_falls_back"]
+)
+def test_voice_mode_omits_language_without_a_trusted_signal(voice_probe, scenario):
+ requests = voice_probe[scenario]["tts"]
+ assert requests, "voice mode never reached /api/tts"
+ for request in requests:
+ assert "language" not in request
+
+
+def test_voice_mode_drops_hostile_language_values(voice_probe):
+ for case in voice_probe["hostile_language_values_are_dropped"]["results"]:
+ for request in case["tts"]:
+ assert "language" not in request, case["sent"]
+
+
+def test_voice_mode_never_sends_a_voice_field(voice_probe):
+ for request in voice_probe["voice_field_is_never_sent"]["tts"]:
+ assert set(request) <= {"text", "engine", "language"}
+ assert "voice" not in request
+
+
+def test_voice_mode_does_not_reuse_a_previous_turn_language(voice_probe):
+ requests = voice_probe["language_does_not_leak_into_later_turn"]["tts"]
+ assert len(requests) == 3
+ assert requests[0]["language"] == "ru"
+ assert "language" not in requests[1]
+ assert requests[2]["language"] == "es"
+
+
+def test_voice_mode_ignores_language_from_an_empty_transcript(voice_probe):
+ result = voice_probe["empty_transcript_does_not_arm_a_language"]
+ assert result["sendsAfterBlank"] == []
+ assert result["sends"] == ["Hello."]
+ assert result["tts"], "the follow-up turn should still be spoken"
+ for request in result["tts"]:
+ assert "language" not in request
+
+
+def test_voice_mode_discards_language_when_the_session_changes(voice_probe):
+ result = voice_probe["session_change_discards_language"]
+ assert result["afterSwitch"] == []
+ for request in result["tts"]:
+ assert "language" not in request
+
+
+def test_voice_mode_discards_language_when_voice_mode_is_turned_off(voice_probe):
+ result = voice_probe["deactivation_discards_language"]
+ assert result["afterDeactivate"] == []
+ for request in result["tts"]:
+ assert "language" not in request
+
+
+def test_voice_mode_speaks_nothing_when_transcription_fails(voice_probe):
+ result = voice_probe["transcribe_error_speaks_nothing"]
+ assert result["tts"] == []
+ assert any("Whisper is down" in toast for toast in result["toasts"])
+
+
+# ---------------------------------------------------------------------------
+# 6. Build-time enforcement and documented semantics
+# ---------------------------------------------------------------------------
+
+
+def test_image_build_verifies_every_language_routing_patch():
+ dockerfile = (DOCKERFILES / "Dockerfile.hermes-webui").read_text(encoding="utf-8")
+ assert "'language': detected" in dockerfile
+ assert "def _atlas_tts_language(body):" in dockerfile
+ assert 'request_payload["language"] = _atlas_language' in dockerfile
+ assert '"language": detected_language' in dockerfile
+ assert "takeSttLanguage(token)" in dockerfile
+ assert "/opt/hermes-webui/api/upload.py" in dockerfile
+ assert "/opt/hermes/tools/transcription_tools.py" in dockerfile
+
+
+def test_atlas_patch_roots_are_overridable_for_offline_verification():
+ patch = ATLAS_PATCH.read_text(encoding="utf-8")
+ assert 'os.environ.get("HERMES_WEBUI_PATCH_ROOT", "/opt/hermes-webui")' in patch
+ assert 'os.environ.get("HERMES_AGENT_PATCH_ROOT", "/opt/hermes")' in patch
+
+
+def test_notes_document_the_stt_driven_voice_selection_and_its_limits():
+ notes = (HERMES / "NOTES.md").read_text(encoding="utf-8")
+ assert "STT-detected language" in notes
+ for marker in ("hands-free", "Typed messages", "en_US-amy-medium"):
+ assert marker in notes