atlas-iac/dockerfiles/hermes-webui-atlas-patch.py

#!/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")


index = ROOT / "static/index.html"
replace_exact(
    index,
    '<option value="browser">Browser speech synthesis</option><option value="edge">Edge TTS (server)</option>',
    '<option value="atlas">Atlas Jetson (private)</option><option value="browser">Browser speech synthesis</option><option value="edge">Edge TTS (server)</option>',
)
replace_exact(
    index,
    '<script src="static/boot.js?v=__WEBUI_VERSION__" defer></script>',
    '<script src="static/boot.js?v=__WEBUI_VERSION__" defer></script>\n<script src="static/atlas-voice.js?v=__WEBUI_VERSION__" defer></script>',
)

ui = ROOT / "static/ui.js"
replace_exact(ui, "function _playEdgeTtsChunked(text, btn){", "function _playEdgeTtsChunked(text, btn, engineOverride){")
replace_exact(
    ui,
    "body:JSON.stringify({text:chunk, voice:voice, rate:rate, pitch:pitch})",
    "body:JSON.stringify({text:chunk, voice:voice, rate:rate, pitch:pitch, engine:engineOverride||'edge'})",
)
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);",
)

# 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,
    "def _tts_open(req, *, timeout=30, opener_factory=None):",
    '''ATLAS_TTS_LANGUAGES = ("en", "ru", "es")


def _atlas_tts_language(body):
    """Resolve the private TTS language from a request body, English by default.

    Only a language the private Piper deployment actually bakes a voice for is
    honoured. Everything else — a missing field, a hostile string, a wrong type,
    or a client-supplied "voice" — resolves to English, so nothing a browser
    sends can steer synthesis outside the fixed policy. The Jetson service
    applies the same allow-list again as the final authority.
    """
    if not isinstance(body, dict):
        return "en"
    value = body.get("language")
    if not isinstance(value, str):
        return "en"
    code = value.strip().lower().replace("_", "-").split("-", 1)[0]
    return code if code in ATLAS_TTS_LANGUAGES else "en"


def _tts_open(req, *, timeout=30, opener_factory=None):''',
)
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
        request_body = json.dumps({
            "model": "piper",
            "input": text,
            "voice": "en_US-lessac-high",
            "speed": speed,
            "language": _atlas_tts_language(data),
        }).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)