atlas-iac/testing/tests/test_hermes_voice_instrument.py

"""Shipped patch and DOM contracts for the hands-free conversation instrument."""

from __future__ import annotations

import os
from pathlib import Path
import re
import shutil
import subprocess
import sys

from hux_node_gate import require_node


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"
DOM_PROBE = ROOT / "testing/probes/hermes_voice_instrument_probe.js"
MEDIARECORDER_FIXTURE = (
    ROOT / "testing/fixtures/mediarecorder/chromium-webm-opus.json"
)


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,
        env=env,
        check=True,
        capture_output=True,
        text=True,
    )
    return target


def test_real_upstream_fixture_receives_visual_instrument_contract(tmp_path: Path):
    """Apply the production patcher to exact fragments from pinned WebUI 0.52.181."""
    target = _patched_fixture(tmp_path)
    index = (target / "static/index.html").read_text(encoding="utf-8")

    assert index.count('id="voiceInstrumentStyles"') == 1
    assert (
        'href="static/atlas-voice.css?v=__WEBUI_VERSION__"' in index
    )
    assert 'role="status"' in index
    assert 'aria-live="polite"' in index
    assert 'aria-atomic="true"' in index
    for layer in (
        "voice-instrument-halo",
        "voice-instrument-ripple",
        "voice-instrument-orbit",
        "voice-instrument-core",
        "voice-instrument-symbol",
    ):
        assert layer in index
    assert '<button' not in index[index.index('id="voiceModeBar"') : index.index('<textarea')]

    service_worker = (target / "static/sw.js").read_text(encoding="utf-8")
    for asset in ("atlas-voice.css", "atlas-voice.js", "atlas-voice-worklet.js"):
        assert f"'./static/{asset}' + VQ" in service_worker


def test_patched_webui_has_no_user_voice_choice_or_client_voice_field(
    tmp_path: Path,
):
    """The pinned settings DOM and every TTS path leave speakers to policy."""
    target = _patched_fixture(tmp_path)
    index = (target / "static/index.html").read_text(encoding="utf-8")
    ui = (target / "static/ui.js").read_text(encoding="utf-8")
    panels = (target / "static/panels.js").read_text(encoding="utf-8")
    boot = (target / "static/boot.js").read_text(encoding="utf-8")
    i18n = (target / "static/i18n.js").read_text(encoding="utf-8")
    config = (target / "api/config.py").read_text(encoding="utf-8")
    routes = (target / "api/routes.py").read_text(encoding="utf-8")

    assert "settingsTtsVoice" not in index
    assert "settings_label_tts_voice" not in index
    assert "settings_desc_tts_voice" not in index
    assert "Default system voice" not in index
    assert 'id="settingsTtsEngine"' in index
    assert 'id="btnVoiceMode"' in index
    assert 'id="voiceModeBar"' in index

    assert "hermes-tts-voice" not in ui
    assert "voice:voice" not in ui
    assert (
        "body:JSON.stringify({text:chunk, rate:rate, pitch:pitch, "
        "engine:engineOverride||'edge'})"
    ) in ui
    assert "settingsTtsVoice" not in panels
    assert "tts_voice" not in panels
    assert "localStorage.removeItem('hermes-tts-voice')" in panels
    assert panels.count("hermes-tts-voice") == 1
    assert "hermes-tts-voice" not in boot
    assert "tts_voice" not in boot
    assert "text: clean, voice" not in boot
    assert '"tts_voice"' not in config
    assert "settings_label_tts_voice" not in i18n
    assert "settings_desc_tts_voice" not in i18n

    atlas_route = routes.split('if engine == "atlas":', 1)[1].split(
        "# ── ElevenLabs TTS", 1
    )[0]
    assert '"input": text' in atlas_route
    assert '"voice"' not in atlas_route
    assert 'request_payload["language"] = _atlas_language' in atlas_route


def test_visual_states_have_distinct_layers_finite_error_and_reduced_motion():
    css = VOICE_CSS.read_text(encoding="utf-8")

    for state in ("listening", "transcribing", "thinking", "speaking", "error"):
        assert f".voice-mode-indicator.{state}" in css
    for animation in (
        "voice-instrument-breathe",
        "voice-instrument-orbit",
        "voice-instrument-speaking-pulse",
    ):
        assert f"@keyframes {animation}" in css
    assert (
        ".voice-mode-indicator.speaking.is-playing .voice-instrument-halo"
        in css
    )

    error_rules = "\n".join(
        match.group(0)
        for match in re.finditer(r"[^{}]*\.error[^{}]*\{[^{}]*\}", css)
    )
    assert "animation:" not in error_rules
    assert "@media (prefers-reduced-motion: reduce)" in css
    reduced = css.split("@media (prefers-reduced-motion: reduce)", 1)[1]
    assert "animation: none !important" in reduced
    assert "transition: none !important" in reduced


def test_dom_probe_exercises_actual_injected_voice_script():
    require_node()
    result = subprocess.run(
        ["node", str(DOM_PROBE), str(VOICE_JS), str(MEDIARECORDER_FIXTURE)],
        cwd=ROOT,
        check=True,
        capture_output=True,
        text=True,
    )
    assert result.stdout.strip() == "voice instrument DOM contract passed"


def test_visual_slice_preserves_private_voice_request_and_capture_contract():
    script = VOICE_JS.read_text(encoding="utf-8")

    assert script.count("navigator.mediaDevices.getUserMedia(") == 1
    assert "form.append('file',new File([blob],'voice-input.'+ext" in script
    assert "fetch('/api/transcribe',{method:'POST',body:form})" in script
    assert (
        "const request={text:chunk,engine:'atlas',turn_id:turnId,speed:ttsSpeed()}"
        in script
    )
    assert "if(language) request.language=language" in script
    assert "speakResponse(generation)" in script
    assert "window._voiceModeImmediateSend" in script
    # Conversation-mode mute must silence the retained capture tracks rather
    # than opening a second microphone lease or a speaker-choice surface.
    assert "track.enabled=!conversation.muted" in script
    assert "hermes-tts-voice" not in script
    assert "const TTS_LANGUAGES=['en','ru','es']" in script


def test_release_candidate_streaming_keeps_one_ahead_and_safe_fallbacks():
    script = VOICE_JS.read_text(encoding="utf-8")
    worklet = (
        ROOT / "dockerfiles" / "hermes-webui-atlas-voice-worklet.js"
    ).read_text(encoding="utf-8")

    assert "At most one later synthesis request exists" in script
    assert "const nextPrepared=next.then" in script
    assert "await playPrepared(asset,turn.token)" in script
    assert "playBlob(await fetchSpeech(asset.chunk,asset.language,asset.turnId,token),token)" in script
    assert "await closePcmBeforeBlob(playbackSession,token)" in script
    assert "if(asset.started) throw error" in script
    assert "playbackSession.controllers.forEach(function(controller){controller.abort();})" in script
    assert "sendJson({type:'cancel',turn_id:turnId})" in script
    assert "payload.turn_id!==turnId" in script
    assert "window.setTimeout(wake,3000)" in script

    assert "registerProcessor('atlas-pcm-capture'" in worklet
    assert "registerProcessor('atlas-pcm-playback'" in worklet
    assert "output[index]=right===null?left:left+((right-left)*this.phase)" in worklet
    assert "this.port.postMessage({type:'buffer',frames:this.queuedFrames})" in worklet
    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text(
        encoding="utf-8"
    )
    assert "COPY dockerfiles/hermes-webui-atlas-voice-worklet.js" in dockerfile


def test_release_candidate_keeps_finalized_webm_as_quality_fallback():
    script = VOICE_JS.read_text(encoding="utf-8")

    assert "recorder.start();" in script
    assert "recorder.start(250)" not in script
    assert "new Blob(chunks,{type:recordedMime||'audio/webm'})" in script
    assert "transcribeStreamingOrFallback" in script
    assert "transcribe(blob,token,turnId)" in script


def test_session_continuity_profile_gate_survives_transient_lookup_failures(
    tmp_path,
):
    """The /api/session profile gate must not 409 ("This session is
    unavailable to this account.") on a cold or transiently failing
    root-alias lookup — only on a confirmed cross-profile mismatch."""
    target = _patched_fixture(tmp_path)
    profiles = (target / "api/profiles.py").read_text(encoding="utf-8")

    assert "_root_profile_names_confirmed" in profiles
    assert (
        'if "default" in (row, active) and not _root_profile_names_confirmed():'
        in profiles
    )
    # Listing failure now answers from the last known alias set (one extra
    # cache read next to the two pinned upstream reads).
    assert profiles.count("return name in _root_profile_name_cache") == 3