#!/usr/bin/env python3
"""Apply fail-closed Atlas voice integration patches to pinned Hermes WebUI."""
import os
from pathlib import Path
ROOT = Path(os.environ.get("HERMES_WEBUI_PATCH_ROOT", "/opt/hermes-webui"))
# The WebUI imports the pinned agent's STT tooling from the same image, so the
# local-command transcription envelope is patched alongside the WebUI itself.
AGENT_ROOT = Path(os.environ.get("HERMES_AGENT_PATCH_ROOT", "/opt/hermes"))
def replace_exact(path: Path, before: str, after: str, count: int = 1) -> None:
"""Replace an exact upstream fragment and fail when the pin has drifted."""
source = path.read_text(encoding="utf-8")
if source.count(before) != count:
raise SystemExit(f"Atlas voice patch context changed in {path}: {before[:80]!r}")
path.write_text(source.replace(before, after, count), encoding="utf-8")
def replace_between_exact(
path: Path, start: str, end: str, after: str = "", count: int = 1
) -> None:
"""Replace one exact, bounded upstream region and fail when the pin drifts."""
source = path.read_text(encoding="utf-8")
if source.count(start) != count or source.count(end) != count:
raise SystemExit(
f"Atlas voice patch context changed in {path}: {start[:80]!r}"
)
start_index = source.index(start)
end_index = source.index(end, start_index) + len(end)
path.write_text(
source[:start_index] + after + source[end_index:], encoding="utf-8"
)
def assert_absent(path: Path, *needles: str) -> None:
"""Fail the image build if a removed voice-choice surface remains."""
source = path.read_text(encoding="utf-8")
remaining = [needle for needle in needles if needle in source]
if remaining:
raise SystemExit(f"Atlas voice choice remains in {path}: {remaining!r}")
def remove_lines_containing(path: Path, *needles: str) -> None:
"""Remove all pinned translation entries for a retired settings control."""
source = path.read_text(encoding="utf-8")
for needle in needles:
if needle not in source:
raise SystemExit(f"Atlas voice patch context changed in {path}: {needle!r}")
lines = source.splitlines(keepends=True)
path.write_text(
"".join(line for line in lines if not any(n in line for n in needles)),
encoding="utf-8",
)
index = ROOT / "static/index.html"
replace_exact(
index,
'',
'\n'
'',
)
replace_exact(
index,
'',
'',
)
replace_exact(
index,
'''
Preferred voice. Populated from your browser's available voices.
''',
)
service_worker = ROOT / "static/sw.js"
replace_exact(
service_worker,
" './static/style.css' + VQ,\n",
" './static/style.css' + VQ,\n"
" './static/atlas-voice.css' + VQ,\n"
" './static/atlas-voice.js' + VQ,\n"
" './static/atlas-voice-worklet.js' + VQ,\n",
)
ui = ROOT / "static/ui.js"
replace_exact(
ui,
''' 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;
}
''',
"",
)
replace_exact(ui, "function _playEdgeTtsChunked(text, btn){", "function _playEdgeTtsChunked(text, btn, engineOverride){")
replace_exact(
ui,
" const voice=localStorage.getItem('hermes-tts-voice')||'zh-CN-XiaoxiaoNeural';\n",
"",
)
replace_exact(
ui,
"body:JSON.stringify({text:chunk, voice:voice, rate:rate, pitch:pitch})",
"body:JSON.stringify({text:chunk, rate:rate, pitch:pitch, engine:engineOverride||'edge'})",
)
replace_exact(
ui,
" voice: localStorage.getItem('hermes-tts-voice')||'',\n",
"",
count=2,
)
replace_exact(
ui,
"if(engine==='edge'){\n _playEdgeTtsChunked(clean, btn);",
"if(engine==='edge'||engine==='atlas'){\n _playEdgeTtsChunked(clean, btn, engine);",
)
replace_exact(
ui,
"if(engine==='edge'){\n _playEdgeTtsChunked(clean, null);",
"if(engine==='edge'||engine==='atlas'){\n _playEdgeTtsChunked(clean, null, engine);",
)
panels = ROOT / "static/panels.js"
replace_exact(panels, " tts_voice:'hermes-tts-voice',\n", "")
replace_exact(
panels,
''' const ttsVoiceSel=$('settingsTtsVoice');
if(ttsVoiceSel) _setOwnedSpeechPayload(payload,'tts_voice',ttsVoiceSel.value||'');
''',
"",
)
replace_exact(
panels,
''' localStorage.setItem('hermes-tts-engine',this.value);
window._populateTtsVoices();
_schedulePreferencesAutosave();''',
''' localStorage.setItem('hermes-tts-engine',this.value);
_schedulePreferencesAutosave();''',
)
replace_between_exact(
panels,
" // Populate voice selector based on engine\n",
" // TTS rate/pitch sliders\n",
" // TTS speaker selection is intentionally server policy only.\n",
)
replace_exact(
panels,
"let _settingsSpeechChangedKeys=new Set();\n",
"let _settingsSpeechChangedKeys=new Set();\n"
"try{localStorage.removeItem('hermes-tts-voice');}catch(_){}\n",
)
boot = ROOT / "static/boot.js"
replace_exact(
boot,
''' voice: localStorage.getItem("hermes-tts-voice")||'',
''',
"",
)
replace_exact(
boot,
''' const voice=localStorage.getItem("hermes-tts-voice")||"zh-CN-XiaoxiaoNeural";
''',
"",
)
replace_exact(
boot,
" body: JSON.stringify({text: clean, voice, rate, pitch})",
" body: JSON.stringify({text: clean, rate, pitch})",
)
replace_exact(
boot,
''' 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;
}
''',
"",
)
replace_exact(boot, " tts_voice:'',\n", "")
replace_exact(boot, " ['tts_voice','hermes-tts-voice'],\n", "")
config = ROOT / "api/config.py"
replace_exact(config, ' "tts_voice": "",\n', "")
replace_exact(config, ' "tts_voice",\n', "")
replace_exact(
config,
''' if k == "tts_voice":
if not isinstance(v, str) or len(v) > 200 or "\\x00" in v:
continue
''',
"",
)
assert_absent(index, "settingsTtsVoice", "settings_label_tts_voice")
assert_absent(ui, "hermes-tts-voice", "voice:voice")
assert_absent(panels, "settingsTtsVoice", "tts_voice")
assert_absent(boot, "hermes-tts-voice", "tts_voice", "text: clean, voice")
assert_absent(config, '"tts_voice"')
i18n = ROOT / "static/i18n.js"
remove_lines_containing(
i18n,
"settings_label_tts_voice:",
"settings_desc_tts_voice:",
)
assert_absent(i18n, "settings_label_tts_voice", "settings_desc_tts_voice")
# The private Whisper service reports the language it decoded with. Carry that
# through the agent's local-command STT envelope so the WebUI can hand a voice
# hint to Piper instead of guessing the reply's language from its text.
transcription = AGENT_ROOT / "tools/transcription_tools.py"
replace_exact(
transcription,
''' 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"}
''',
''' 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),
)
detected_language = ""
language_files = sorted(Path(output_dir).glob("*.language"))
if language_files:
try:
candidate = language_files[0].read_text(encoding="utf-8").strip().lower()
except (OSError, ValueError):
candidate = ""
if 2 <= len(candidate) <= 3 and candidate.isascii() and candidate.isalpha():
detected_language = candidate
return {
"success": True,
"transcript": transcript_text,
"provider": "local_command",
"language": detected_language,
}
''',
)
upload = ROOT / "api/upload.py"
replace_exact(
upload,
""" transcript = str(result.get('transcript') or '').strip()
return j(handler, {'ok': True, 'transcript': transcript})
""",
""" transcript = str(result.get('transcript') or '').strip()
detected = str(result.get('language') or '').strip().lower()
if not (2 <= len(detected) <= 3 and detected.isascii() and detected.isalpha()):
detected = ''
return j(handler, {'ok': True, 'transcript': transcript, 'language': detected})
""",
)
routes = ROOT / "api/routes.py"
replace_exact(
routes,
"import html as _html\n",
"import base64\nimport html as _html\nimport secrets\n",
)
replace_exact(
routes,
"def _tts_open(req, *, timeout=30, opener_factory=None):",
'''ATLAS_TTS_LANGUAGES = ("en", "ru", "es")
ATLAS_TTS_CUE_IDS = ("thinking", "let_me_think", "still_working", "one_more_moment")
def _atlas_tts_language(body):
"""Return a plain, allow-listed en/ru/es code, or "" to send no language.
This is a trust boundary, not a parser. Only the exact normalized codes the
private Piper deployment bakes a voice for are forwarded; a missing field,
a wrong type, a region tag, padding, control characters, a traversal or
injection string, an oversized value, an object, an array, a number or a
client-supplied "voice" all resolve to "" and the language field is then
omitted entirely, so the Jetson service applies its own English default.
Coercing a malformed value into a supported code would let a browser
describe hostile input as a language we support; omission cannot.
"""
if not isinstance(body, dict):
return ""
value = body.get("language")
if not isinstance(value, str):
return ""
return value if value in ATLAS_TTS_LANGUAGES else ""
def _tts_open(req, *, timeout=30, opener_factory=None):''',
)
replace_exact(
routes,
"def _tts_open(req, *, timeout=30, opener_factory=None):",
'''ATLAS_TTS_STREAM_URL = "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech/stream"
ATLAS_STT_STREAM_URL = "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions/stream"
ATLAS_VOICE_PREFLIGHT_URL = "http://hermes-switchyard.hermes.svc.cluster.local:9009/voice/route-preflight"
ATLAS_VOICE_WS_PROTOCOL = "hermes-voice-v1"
ATLAS_VOICE_MAX_BYTES = 8 * 1024 * 1024
ATLAS_VOICE_DEADLINE_SECONDS = 100
ATLAS_VOICE_PREFLIGHT_TIERS = ("fast", "balanced", "deep", "maximum")
ATLAS_VOICE_PREFLIGHT_MAX_RESPONSE_BYTES = 1024
def _atlas_exact_stream_url(env_name, expected):
"""Return an exact private service URL, never a browser-selected target."""
configured = os.getenv(env_name, "").strip()
return configured if configured == expected else ""
def _handle_atlas_streaming_capability(handler):
"""Advertise only transports whose immutable in-cluster URLs are configured."""
tts = bool(_atlas_exact_stream_url("HERMES_WEBUI_ATLAS_TTS_STREAM_URL", ATLAS_TTS_STREAM_URL))
stt = bool(_atlas_exact_stream_url("HERMES_WEBUI_ATLAS_STT_STREAM_URL", ATLAS_STT_STREAM_URL))
preflight = True
j(handler, {
"tts": {
"available": tts,
"transport": "http",
"format": "pcm_s16le",
"sample_rate": 22050,
},
"stt": {
"available": stt,
"transport": "websocket",
"path": "/api/transcribe/stream",
"format": "pcm_s16le",
"sample_rate": 16000,
},
"preflight": {
"available": preflight,
"path": "/api/voice/route-preflight",
"advisory": True,
},
}, extra_headers={"Cache-Control": "no-store"})
return True
def _atlas_voice_preflight_payload(data):
"""Validate a provisional transcript without accepting routing authority."""
if not isinstance(data, dict):
raise ValueError("invalid request body")
turn_id = data.get("turn_id")
revision = data.get("revision")
transcript = data.get("transcript")
if not isinstance(turn_id, str) or not re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", turn_id):
raise ValueError("invalid turn_id")
if isinstance(revision, bool) or not isinstance(revision, int) or not 1 <= revision <= 1000000:
raise ValueError("invalid revision")
if not isinstance(transcript, str):
raise ValueError("invalid transcript")
transcript = " ".join(transcript.split())
if not 12 <= len(transcript) <= 512:
raise ValueError("invalid transcript")
return {"turn_id": turn_id, "revision": revision, "transcript": transcript}
def _atlas_voice_preflight_open(request, timeout=1.0):
"""Open only the immutable in-cluster advisory endpoint without proxies."""
return build_opener(ProxyHandler({}), _NoRedirectTtsHandler()).open(request, timeout=timeout)
def _handle_atlas_voice_preflight(handler):
"""Relay one same-origin local advisory and expose no classifier output."""
if not _check_same_origin_browser_request(handler):
return bad(handler, "Voice route preflight origin validation failed", 403)
target = ATLAS_VOICE_PREFLIGHT_URL
try:
payload = _atlas_voice_preflight_payload(read_body(handler))
except (TypeError, ValueError) as exc:
return bad(handler, str(exc), 400)
request = Request(
target,
data=json.dumps(payload, separators=(",", ":")).encode("utf-8"),
headers={"Content-Type": "application/json", "Accept": "application/json"},
)
try:
with _atlas_voice_preflight_open(request, timeout=1.0) as upstream:
raw = upstream.read(ATLAS_VOICE_PREFLIGHT_MAX_RESPONSE_BYTES + 1)
if len(raw) > ATLAS_VOICE_PREFLIGHT_MAX_RESPONSE_BYTES:
raise ValueError("oversized advisory response")
result = json.loads(raw)
tier = result.get("tier") if isinstance(result, dict) else None
target_hint = result.get("target") if isinstance(result, dict) else None
if (
result.get("turn_id") != payload["turn_id"]
or result.get("revision") != payload["revision"]
or tier not in ATLAS_VOICE_PREFLIGHT_TIERS
or target_hint != "atlas/auto/" + tier
or result.get("advisory") is not True
):
raise ValueError("invalid advisory response")
except Exception:
# Advisory failure never changes or delays the final Switchyard request.
return bad(handler, "Voice route preflight unavailable", 503)
return j(
handler,
{
"turn_id": payload["turn_id"],
"revision": payload["revision"],
"tier": tier,
"target": target_hint,
"advisory": True,
},
extra_headers={"Cache-Control": "no-store"},
)
def _atlas_tts_speed(body):
"""Clamp the optional client speech-rate to Piper's supported 0.5-2.0.
The value is a UX preference, not a trust decision: a missing, boolean,
non-numeric or NaN value falls back to the neutral 1.0 this proxy always
sent before the hands-free speed became client-tunable. Infinities clamp
to the range bounds like any other out-of-range number.
"""
if not isinstance(body, dict):
return 1.0
value = body.get("speed")
if isinstance(value, bool) or not isinstance(value, (int, float)):
return 1.0
speed = float(value)
if speed != speed:
return 1.0
return max(0.5, min(2.0, speed))
def _atlas_tts_stream_payload(data):
"""Build the narrow Piper payload used by the raw-PCM stream endpoint."""
if not isinstance(data, dict):
raise ValueError("invalid request body")
text = data.get("text")
if not isinstance(text, str) or not text.strip():
raise ValueError("text is required")
text = text.strip()
if len(text) > 500:
raise ValueError("text too long (max 500 characters)")
payload = {"model": "piper", "input": text, "speed": _atlas_tts_speed(data)}
language = _atlas_tts_language(data)
if language:
payload["language"] = language
cue_id = data.get("cue_id")
if "cue_id" in data:
if not language or cue_id not in ATLAS_TTS_CUE_IDS:
raise ValueError("invalid thinking cue")
payload["cue_id"] = cue_id
turn_id = data.get("turn_id")
if isinstance(turn_id, str) and re.fullmatch(r"[A-Za-z0-9._:-]{1,128}", turn_id):
payload["turn_id"] = turn_id
return payload
def _handle_atlas_tts_stream(handler):
"""Relay bounded private Piper PCM using HTTP/1.1 chunk framing."""
target = _atlas_exact_stream_url("HERMES_WEBUI_ATLAS_TTS_STREAM_URL", ATLAS_TTS_STREAM_URL)
if not target:
return bad(handler, "Atlas streaming TTS is not configured", 503)
headers_sent = False
try:
payload = _atlas_tts_stream_payload(read_body(handler))
except (TypeError, ValueError) as exc:
return bad(handler, str(exc), 400)
request = Request(
target,
data=json.dumps(payload).encode("utf-8"),
headers={"Content-Type": "application/json", "Accept": "audio/pcm"},
)
try:
upstream = _tts_open(
request,
timeout=45,
opener_factory=lambda: build_opener(ProxyHandler({}), _NoRedirectTtsHandler()),
)
with upstream:
content_type = str(upstream.headers.get("Content-Type", "")).split(";", 1)[0].lower()
if content_type not in ("audio/l16", "audio/pcm", "application/octet-stream"):
raise ValueError("Atlas streaming TTS returned an unsupported format")
try:
sample_rate = int(upstream.headers.get("X-Audio-Sample-Rate", "22050"))
channels = int(upstream.headers.get("X-Audio-Channels", "1"))
except (TypeError, ValueError):
raise ValueError("Atlas streaming TTS returned invalid audio metadata")
if not 8000 <= sample_rate <= 96000 or channels != 1:
raise ValueError("Atlas streaming TTS returned invalid audio metadata")
handler.send_response(200)
handler.send_header(
"Content-Type",
f"audio/pcm;rate={sample_rate};channels=1;encoding=signed-integer;bits=16;endian=little",
)
handler.send_header("X-Audio-Sample-Rate", str(sample_rate))
handler.send_header("X-Audio-Channels", "1")
handler.send_header("Cache-Control", "no-store")
handler.send_header("Transfer-Encoding", "chunked")
handler.end_headers()
headers_sent = True
sent = 0
while True:
chunk = upstream.read(16384)
if not chunk:
break
sent += len(chunk)
if sent > ATLAS_VOICE_MAX_BYTES:
raise ValueError("Atlas streaming TTS exceeded its audio limit")
handler.wfile.write(("%x\\r\\n" % len(chunk)).encode("ascii"))
handler.wfile.write(chunk)
handler.wfile.write(b"\\r\\n")
handler.wfile.flush()
handler.wfile.write(b"0\\r\\n\\r\\n")
handler.wfile.flush()
return True
except (BrokenPipeError, ConnectionResetError):
return True
except Exception:
logger.exception("Atlas streaming TTS generation failed")
if not headers_sent:
return bad(handler, "Atlas streaming TTS generation failed", 502)
handler.close_connection = True
return True
def _atlas_ws_protocols(handler):
return [value.strip() for value in handler.headers.get("Sec-WebSocket-Protocol", "").split(",") if value.strip()]
def _atlas_ws_authorized(handler):
"""Require same-origin plus the rendered session CSRF token when enabled."""
origin = handler.headers.get("Origin", "").strip()
if not origin or not _check_same_origin_browser_request(handler):
return False
protocols = _atlas_ws_protocols(handler)
if ATLAS_VOICE_WS_PROTOCOL not in protocols:
return False
from api.auth import csrf_token_for_session, is_auth_enabled, parse_cookie, verify_session
if not is_auth_enabled():
return True
cookie = parse_cookie(handler)
if not cookie:
cookie = getattr(handler, "_trusted_auth_session_cookie_value", None)
if not cookie or not verify_session(cookie):
return False
expected = csrf_token_for_session(cookie) or ""
supplied = next((value.removeprefix("hermes-csrf.") for value in protocols if value.startswith("hermes-csrf.")), "")
return bool(expected and supplied and secrets.compare_digest(expected, supplied))
def _atlas_ws_upstream_handshake(target):
parsed = urlsplit(target)
if parsed.scheme != "http" or not parsed.hostname or parsed.port is None:
raise ValueError("invalid private streaming STT URL")
upstream = _socket.create_connection((parsed.hostname, parsed.port), timeout=5)
upstream.settimeout(5)
key = base64.b64encode(os.urandom(16)).decode("ascii")
path = parsed.path or "/"
if parsed.query:
path += "?" + parsed.query
request = (
f"GET {path} HTTP/1.1\\r\\nHost: {parsed.hostname}:{parsed.port}\\r\\n"
"Upgrade: websocket\\r\\nConnection: Upgrade\\r\\n"
f"Sec-WebSocket-Key: {key}\\r\\nSec-WebSocket-Version: 13\\r\\n\\r\\n"
).encode("ascii")
upstream.sendall(request)
response = bytearray()
while b"\\r\\n\\r\\n" not in response and len(response) < 16384:
chunk = upstream.recv(4096)
if not chunk:
break
response.extend(chunk)
header, separator, remainder = bytes(response).partition(b"\\r\\n\\r\\n")
if not separator or not header.startswith(b"HTTP/1.1 101"):
upstream.close()
raise ConnectionError("private streaming STT rejected WebSocket upgrade")
expected = base64.b64encode(hashlib.sha1((key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii")).digest()).decode("ascii")
headers = {}
for line in header.split(b"\\r\\n")[1:]:
name, marker, value = line.partition(b":")
if marker:
headers[name.decode("ascii", "ignore").strip().lower()] = value.decode("ascii", "ignore").strip()
if not secrets.compare_digest(headers.get("sec-websocket-accept", ""), expected):
upstream.close()
raise ConnectionError("private streaming STT returned an invalid handshake")
upstream.settimeout(1)
return upstream, remainder
def _atlas_ws_relay(source, destination, stop, deadline):
transferred = 0
while not stop.is_set() and time.monotonic() < deadline:
try:
chunk = source.recv(16384)
except (_socket.timeout, TimeoutError):
continue
except OSError:
break
if not chunk:
break
transferred += len(chunk)
if transferred > ATLAS_VOICE_MAX_BYTES:
break
try:
destination.sendall(chunk)
except OSError:
break
stop.set()
def _handle_atlas_stt_stream(handler):
"""Bridge one authenticated same-origin browser WebSocket to private STT."""
target = _atlas_exact_stream_url("HERMES_WEBUI_ATLAS_STT_STREAM_URL", ATLAS_STT_STREAM_URL)
if not target:
return bad(handler, "Atlas streaming STT is not configured", 503)
if handler.headers.get("Upgrade", "").strip().lower() != "websocket":
return bad(handler, "WebSocket upgrade required", 426)
if not _atlas_ws_authorized(handler):
return bad(handler, "WebSocket origin or CSRF validation failed", 403)
browser_key = handler.headers.get("Sec-WebSocket-Key", "").strip()
if handler.headers.get("Sec-WebSocket-Version", "").strip() != "13":
return bad(handler, "WebSocket version 13 required", 426)
try:
decoded = base64.b64decode(browser_key, validate=True)
except Exception:
decoded = b""
if len(decoded) != 16:
return bad(handler, "Invalid WebSocket key", 400)
try:
upstream, remainder = _atlas_ws_upstream_handshake(target)
except Exception:
logger.exception("Atlas streaming STT connection failed")
return bad(handler, "Atlas streaming STT is unavailable", 502)
browser_accept = base64.b64encode(hashlib.sha1((browser_key + "258EAFA5-E914-47DA-95CA-C5AB0DC85B11").encode("ascii")).digest()).decode("ascii")
handler.send_response(101, "Switching Protocols")
handler.send_header("Upgrade", "websocket")
handler.send_header("Connection", "Upgrade")
handler.send_header("Sec-WebSocket-Accept", browser_accept)
handler.send_header("Sec-WebSocket-Protocol", ATLAS_VOICE_WS_PROTOCOL)
handler.end_headers()
handler.wfile.flush()
handler.close_connection = True
browser = handler.connection
browser.settimeout(1)
stop = threading.Event()
deadline = time.monotonic() + ATLAS_VOICE_DEADLINE_SECONDS
if remainder:
browser.sendall(remainder)
reverse = threading.Thread(
target=_atlas_ws_relay,
args=(upstream, browser, stop, deadline),
name="atlas-stt-ws-upstream",
daemon=True,
)
reverse.start()
try:
_atlas_ws_relay(browser, upstream, stop, deadline)
finally:
stop.set()
for sock in (upstream, browser):
try:
sock.shutdown(_socket.SHUT_RDWR)
except OSError:
pass
upstream.close()
reverse.join(timeout=2)
return True
def _tts_open(req, *, timeout=30, opener_factory=None):''',
)
replace_exact(
routes,
'''def handle_get(handler, parsed) -> bool:
"""Handle all GET routes. Returns True if handled, False for 404."""
''',
'''def handle_get(handler, parsed) -> bool:
"""Handle all GET routes. Returns True if handled, False for 404."""
if parsed.path == "/api/voice/streaming/capability":
return _handle_atlas_streaming_capability(handler)
if parsed.path == "/api/transcribe/stream":
return _handle_atlas_stt_stream(handler)
''',
)
replace_exact(
routes,
''' if parsed.path == "/api/transcribe":
return handle_transcribe(handler)
if parsed.path == "/api/tts":
return _handle_tts(handler, parsed)
''',
''' if parsed.path == "/api/transcribe":
return handle_transcribe(handler)
if parsed.path == "/api/voice/route-preflight":
return _handle_atlas_voice_preflight(handler)
if parsed.path == "/api/tts/stream":
return _handle_atlas_tts_stream(handler)
if parsed.path == "/api/tts":
return _handle_tts(handler, parsed)
''',
)
# The chat router's session-continuity poller renders whatever /api/session
# answers. Its 409 branch ("This session is unavailable to this account.")
# is only ever correct when the root-profile alias set is actually known —
# but list_profiles_api() is a hermes_cli subprocess call that fails
# transiently (cold pod after a roll, load spikes), and the pinned
# _is_root_profile() treats that failure as "not a root alias", flipping
# _profiles_match() to a false mismatch and painting the bogus banner over
# the input bar until the next successful listing. Two minimal grafts:
# answer alias checks from the last known alias set on listing failure, and
# never claim a default-vs-named mismatch while the alias set is unconfirmed.
profiles = ROOT / "api/profiles.py"
replace_exact(
profiles,
''' except Exception:
logger.debug("Failed to list profiles for root-profile lookup", exc_info=True)
return False
''',
''' except Exception:
logger.debug("Failed to list profiles for root-profile lookup", exc_info=True)
# Atlas voice patch: a transient listing failure must not deny a name
# that was already confirmed as a root alias — answer from the last
# known alias set instead of failing the alias outright.
with _root_profile_name_cache_lock:
return name in _root_profile_name_cache
''',
)
replace_exact(
profiles,
"def _is_root_profile(name: str) -> bool:\n",
'''def _root_profile_names_confirmed() -> bool:
"""True once list_profiles_api() has successfully populated the alias set.
Atlas voice patch: lets _profiles_match() distinguish "these profiles are
definitely different" from "the root-alias equivalence could not be
checked yet" (cold cache right after a pod roll, or a failing hermes_cli
listing), which previously produced transient /api/session 409s.
"""
with _root_profile_name_cache_lock:
return _root_profile_name_cache_loaded
def _is_root_profile(name: str) -> bool:
''',
)
replace_exact(
profiles,
''' # Cross-alias the renamed root.
if _is_root_profile(row) and _is_root_profile(active):
return True
return False
''',
''' # Cross-alias the renamed root.
if _is_root_profile(row) and _is_root_profile(active):
return True
# Atlas voice patch: while the root alias set is unconfirmed (cold cache
# after a pod roll, hermes_cli listing failure) a pair involving the
# 'default' alias cannot be *proven* mismatched — the named side may be
# the renamed root. Fail open for that pair only: a mismatch between two
# named profiles is still denied, and exact scoping resumes with the
# first successful listing. This removes the transient /api/session 409
# that rendered a false "unavailable to this account" banner.
if "default" in (row, active) and not _root_profile_names_confirmed():
return True
return False
''',
)
marker = " # ── ElevenLabs TTS ──────────────────────────────────────────────────\n"
atlas = ''' # ── Atlas private Jetson TTS ─────────────────────────────────────────
if engine == "atlas":
atlas_url = os.getenv("HERMES_WEBUI_ATLAS_TTS_URL", "").strip()
expected_url = "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech"
if atlas_url != expected_url:
from api.helpers import bad as _bad
return _bad(handler, "Atlas private TTS is not configured", 503)
speed = 1.0
if rate_str:
try:
speed = max(0.5, min(2.0, 1.0 + (float(rate_str.rstrip("%")) / 100.0)))
except ValueError:
speed = 1.0
if isinstance(data, dict) and "speed" in data:
# Hands-free clients send an explicit validated speed; it wins
# over the legacy percentage rate string.
speed = _atlas_tts_speed(data)
request_payload = {
"model": "piper",
"input": text,
"speed": speed,
}
# Attach a language ONLY when the browser sent a plain allow-listed
# code. Omitting it is the fail-safe: the Jetson service then speaks
# its own English default, which is also what every partially rolled
# out combination of these components degrades to.
_atlas_language = _atlas_tts_language(data)
if _atlas_language:
request_payload["language"] = _atlas_language
request_body = json.dumps(request_payload).encode("utf-8")
request = Request(atlas_url, data=request_body, headers={
"Content-Type": "application/json",
"Accept": "audio/wav",
})
try:
with _tts_open(
request,
timeout=45,
opener_factory=lambda: build_opener(ProxyHandler({}), _NoRedirectTtsHandler()),
) as response:
audio_data = _buffer_tts_audio_response(response)
except Exception:
logger.exception("Atlas private TTS generation failed")
from api.helpers import bad as _bad
return _bad(handler, "Atlas private TTS generation failed", 502)
handler.send_response(200)
handler.send_header("Content-Type", "audio/wav")
handler.send_header("Cache-Control", "no-store")
handler.send_header("Content-Length", str(len(audio_data)))
handler.end_headers()
try:
handler.wfile.write(audio_data)
except (BrokenPipeError, ConnectionResetError):
pass
return True
'''
replace_exact(routes, marker, atlas + marker)