atlas-iac/testing/tests/test_hermes_voice_capture_continuity.py

"""Continuous-capture contracts: no first-word clipping, stitching, overlay.

The scenarios drive the real ``atlas-voice.js`` against a stub streaming STT
WebSocket whose VAD, speculative EOS-freeze, resume and commit semantics
mirror ``hermes-jetson-stt-server.py``. Words are encoded as distinct PCM
amplitudes, so the transcript of a commit is exactly the words whose audio
survived endpointing — the live "only the first word was sent" regression is
directly observable in the composer text the probe records.
"""

from __future__ import annotations

import json
import shutil
import subprocess
from pathlib import Path

import pytest


ROOT = Path(__file__).resolve().parents[2]
VOICE_SCRIPT = ROOT / "dockerfiles" / "hermes-webui-atlas-voice.js"
PROBE = ROOT / "testing" / "probes" / "hermes_voice_capture_probe.js"


@pytest.fixture(scope="module")
def probe_results() -> dict:
    node = shutil.which("node")
    if not node:
        pytest.skip("node is required to drive the capture continuity contract")
    completed = subprocess.run(
        [node, str(PROBE), str(VOICE_SCRIPT)],
        check=False,
        capture_output=True,
        text=True,
        timeout=180,
    )
    assert completed.returncode == 0, completed.stderr
    return json.loads(completed.stdout)


def test_multiword_utterance_with_interword_pause_is_complete(probe_results):
    scenario = probe_results["multiword_utterance_with_interword_pause"]
    assert scenario["sends"] == ["alpha bravo"]
    assert scenario["state"] == "thinking"


def test_thinking_pause_after_first_word_does_not_split(probe_results):
    """A >1100ms pause right after the first word stays inside one utterance."""
    scenario = probe_results["thinking_pause_after_first_word_does_not_split"]
    assert scenario["sends"] == ["alpha bravo"]
    # The pause froze a first-word EOS snapshot on the server; resumed speech
    # must have cleared it before the commit.
    events = scenario["serverEvents"][0]
    assert "resume-clear" in events or "recover-by-rms" in events


def test_barge_in_stitches_regardless_of_partial_assistant_output(probe_results):
    scenario = probe_results["second_utterance_during_response_is_complete"]
    assert scenario["firstSends"] == ["alpha"]
    # Assistant text was already visible (and being spoken) when the user
    # talked over the response: the interrupted thought and the follow-up form
    # one stitched message, and the cut marker names the sentence that was
    # playing so the model knows where its reply stopped being heard.
    assert scenario["sends"] == [
        "alpha",
        "alpha bravo charlie\n[voice interruption: you were cut off after "
        '"Partial answer already visible."]',
    ]


def test_normal_completion_clears_stitch_and_mic_stays_hot(probe_results):
    scenario = probe_results["back_to_back_utterances_stay_hot"]
    assert scenario["sendsAfterFirst"] == ["alpha"]
    assert scenario["sends"] == ["alpha", "bravo"]
    # The whole session runs on one getUserMedia lease.
    assert scenario["micAcquisitions"] == 1


def test_interim_ack_fold_flushes_retained_tail_before_final(probe_results):
    """The reported TTS-truncation bug: a quick spoken acknowledgement streams as
    an interim segment, the pump chunks its first sentences and RETAINS the still-
    streaming tail; then a tool call folds the interim into the hidden worklog
    source and the final answer renders as its own segment.

    Before the fix the retained tail was dropped (the empty-read branch returned
    without flushing it) and the queue drained with the state fallen back to
    Thinking while a tail was outstanding, and the final answer resurfaced from
    the interim's stale offset. After the fix the WHOLE acknowledgement reaches
    the speech queue, in order, before the distinct final answer, and the state
    never falls back to Thinking with an unspoken interim tail outstanding.
    """
    scenario = probe_results["interim_ack_fold_flushes_tail_before_final"]
    # (i) every interim sentence reaches the queue — the retained tail is flushed
    # the moment the segment folds, not dropped.
    assert scenario["interimS1Reached"] is True
    assert scenario["interimS2Reached"] is True
    assert scenario["interimTailReached"] is True, "the folded interim tail was dropped"
    assert scenario["tailQueuedAfterFold"] is True
    assert scenario["noFinalBeforeFold"] is True
    # (ii) the interim finishes before the final message's chunks are enqueued.
    assert scenario["interimBeforeFinal"] is True
    # (iii) the distinct final answer is also fully queued, from its own start —
    # never resurfaced from the interim's stale offset.
    assert scenario["finalS1Reached"] is True
    assert scenario["finalS2Reached"] is True
    assert scenario["finalNotGarbled"] is True
    assert scenario["spokenOrder"] == [
        "Sure thing.",
        "Let me look that up.",
        "One moment while I che",
        "The weather today is sunny and warm.",
        "Enjoy your afternoon out there.",
    ]
    # (iv) the state never flipped to Thinking with an unspoken interim tail out.
    assert scenario["thinkingWithUnspokenTail"] == 0


def test_interim_fold_tail_flush_source_contract():
    source = VOICE_SCRIPT.read_text(encoding="utf-8")

    # The retained tail is flushed as final-quality chunks from the text the
    # client already holds — never by re-reading the folded/hidden DOM.
    assert "function flushRetainedTail(turn)" in source
    assert "const tail=turn.sourceText.slice(turn.consumed).trim();" in source
    assert "const flushed=adaptiveChunks(tail,true,false);" in source
    # The empty-read branch flushes the tail before anything else.
    assert "if(speechTurn&&speechTurn.consumed>0) flushRetainedTail(speechTurn);" in source
    # A distinct new message after the fold resets the offset against its OWN
    # text (never advancing the interim's consumed offset into it).
    assert "const spokenPrefix=turn.sourceText.slice(0,turn.consumed);" in source
    assert "if(turn.consumed>0&&spokenPrefix&&!text.startsWith(spokenPrefix)){" in source
    # One audio timeline, one voice: the new message keeps the resolved voice.
    assert "if(!turn.voiceResolved){" in source
    assert "turn.voiceResolved=true;" in source


def test_output_device_selector_routes_all_spoken_output(probe_results):
    """Unified sink + selector: when the browser can enumerate outputs and
    setSinkId, the corner control lists the routable outputs and choosing one
    routes the spoken output (the blob element here) to that device."""
    scenario = probe_results["output_device_selector_routes_spoken_output"]
    assert scenario["overlayPresent"] is True
    assert scenario["shownAfterRefresh"] is True
    assert scenario["itemLabels"] == ["System default", "Speaker One", "Headphones Two"]
    # Selection is reflected accessibly and marked as a non-default override.
    assert scenario["forcedChecked"] == "true"
    assert scenario["btnForced"] is True
    # The spoken reply is routed to the chosen sink.
    assert scenario["audioRoutedTo"] == ["spk-2"]


def test_output_selector_hidden_when_unsupported(probe_results):
    """No enumerateDevices/setSinkId: the control is in the DOM but hidden — it
    never appears as a dead control."""
    scenario = probe_results["output_selector_hidden_when_unsupported"]
    assert scenario["overlayPresent"] is True
    assert scenario["controlInDom"] is True
    assert scenario["hidden"] is True


def test_conversation_overlay_opens_despite_hostile_output_apis(probe_results):
    """REGRESSION LOCK: the full-screen overlay (orb + captions + mute/exit) must
    ALWAYS attach when conversation mode opens, even when the audio-output APIs
    are hostile — navigator.mediaDevices undefined, or enumerateDevices rejecting.
    A failure building the output/language selector can never drop the user back
    to the inline voice bar."""
    scenario = probe_results["overlay_opens_despite_hostile_output_apis"]
    for case in ("mediaDevicesUndefined", "enumerateRejects"):
        shape = scenario[case]
        assert shape["overlayPresent"] is True, case
        assert shape["isDialog"] is True, case
        assert shape["hasOrb"] is True, case
        assert shape["hasOrbMark"] is True, case
        assert shape["hasCaptions"] is True, case
        assert shape["captionCount"] == 2, case
        assert shape["hasMute"] is True, case
        assert shape["hasExit"] is True, case
    # mediaDevices undefined: the output control degrades to hidden, not broken.
    assert scenario["mediaDevicesUndefined"]["outputHidden"] is True
    # A rejecting enumerateDevices does not even briefly withhold the overlay.
    assert scenario["enumerateRejects"]["attachedBeforeReject"] is True


def test_output_defaults_to_loudspeaker(probe_results):
    """FIX 2(a): while the mic is open the OS routes the system default to the
    earpiece; the shared sink auto-defaults to the LOUDSPEAKER (never
    "communications") with no user interaction, and that default reaches real
    playback."""
    scenario = probe_results["output_defaults_to_loudspeaker"]
    assert scenario["overlayPresent"] is True
    assert scenario["defaultChecked"] == ["spk-1"]
    assert scenario["btnForced"] is True
    assert scenario["btnLabel"] == "Audio output: Speakerphone"
    # The auto-selected speaker routes the spoken reply with no user click.
    assert scenario["audioRoutedTo"] == ["spk-1"]


def test_loudspeaker_selection_logic(probe_results):
    """FIX 2(a) pure logic: prefer a labelled speaker, never the "communications"
    endpoint, fall back to the first concrete non-earpiece output, and yield ''
    (system default) when only an earpiece/comms endpoint exists."""
    scenario = probe_results["loudspeaker_selection_logic"]
    assert scenario["labelledSpeaker"] == "spk"
    assert scenario["skipsCommunications"] == "spk"
    assert scenario["concreteFallback"] == "dev-9"
    assert scenario["onlyEarpiece"] == ""
    assert scenario["ignoresInputs"] == "spk"


def test_natural_fillers_unified_sink_and_output_selector_source_contract():
    source = VOICE_SCRIPT.read_text(encoding="utf-8")
    css = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.css").read_text(encoding="utf-8")

    # (2) Natural spoken fillers per language, with the longer reassurances kept.
    assert "{id:'umm',text:'Umm.'}" in source
    assert "{id:'hmm',text:'Hmm.'}" in source
    assert "{id:'one_sec',text:'One sec.'}" in source
    assert "{id:'mmm',text:'Mmm.'}" in source
    assert "{id:'a_ver',text:'A ver.'}" in source
    assert "{id:'hmm',text:'Хм.'}" in source
    assert "{id:'sec',text:'Секунду.'}" in source
    # Answer preempts a filler; a genuine >~1.9s gap gates the first one.
    assert "const THINKING_CUE_FIRST_MS=1900" in source
    assert "cancelThinkingCues();\n    const turn=ensureSpeechTurn(token)" in source
    # Mute-aware: a muted conversation never chatters.
    assert "state==='thinking'&&!(conversation&&conversation.muted)" in source

    # (3) ONE shared sink for every spoken sound. The only AudioContext built for
    # OUTPUT is inside acquirePlaybackContext; cues and PCM replies both use it.
    assert "async function acquirePlaybackContext()" in source
    assert "function closeSharedPlayback()" in source
    assert "cue.context=await acquirePlaybackContext();" in source
    assert "const context=await acquirePlaybackContext();" in source
    # The shared context is never closed per playback session/cue.
    assert "try{cue.context=new Context(" not in source
    assert "context=new Context({sampleRate:asset.sampleRate" not in source
    # The blob fallback element is routed to the same chosen sink.
    assert "audio.setSinkId(selectedOutputSinkId)" in source

    # The output-device selector: feature-detected, session-only, styled.
    assert "function outputRoutingSupported()" in source
    assert "navigator.mediaDevices.enumerateDevices" in source
    assert "function buildOutputControl()" in source
    assert "function selectOutputDevice(deviceId)" in source
    assert "role:'menuitemradio','data-device':entry.deviceId" in source

    # FIX 2(a): auto-default the shared sink to the loudspeaker, never the OS
    # "communications"/earpiece route, and never override an explicit user choice.
    assert "function pickLoudspeakerSink(devices)" in source
    assert "device.deviceId!=='communications'" in source
    assert "if(!outputSinkUserChosen&&!selectedOutputSinkId){" in source
    assert "outputSinkUserChosen=true;" in source  # an explicit choice is sticky

    # REGRESSION: the overlay attaches in a first phase; the language/output
    # selectors are added in a guarded second phase so neither can tear the
    # full-screen overlay down to the inline voice bar. The builders never throw.
    assert "function inertControl(className)" in source
    assert "conversationWarn('language selector omitted'" in source
    assert "conversationWarn('output selector omitted'" in source
    assert "return inertControl('voice-conversation-out')" in source
    assert "return inertControl('voice-conversation-lang')" in source
    # Device enumeration happens async, after attach, tolerant of rejection.
    assert "refreshOutputDevices().catch(function(error)" in source
    # Session-only: the chosen output never touches storage.
    out_region = source.split("function buildOutputControl()", 1)[1].split(
        "function openConversationOverlay", 1
    )[0]
    assert "localStorage" not in out_region
    for token in (
        ".voice-conversation-out",
        ".voice-conversation-out-btn",
        ".voice-conversation-out-menu",
        ".voice-conversation-out-item",
    ):
        assert token in css


def test_conversation_overlay_lifecycle(probe_results):
    scenario = probe_results["conversation_overlay_lifecycle"]
    assert scenario["overlayPresent"] is True
    assert scenario["role"] == "dialog"
    assert scenario["ariaModal"] == "true"
    assert scenario["captionsLive"] == "polite"
    assert scenario["listeningState"] == "listening"
    assert scenario["thinkingState"] in {"thinking", "speaking"}
    assert scenario["userCaption"] == "alpha"
    assert scenario["assistantCaption"] == "A visible answer."
    # Mute drives the real capture track and is fully reversible.
    assert scenario["mutedPressed"] == "true"
    assert scenario["mutedTracks"] == [False]
    assert scenario["unmutedTracks"] == [True]
    # Both exit paths remove the overlay completely and end hands-free mode.
    assert scenario["removedOnExit"] is True
    assert scenario["inactiveAfterExit"] is True
    assert scenario["removedOnEscape"] is True


def test_overlay_source_contract():
    source = VOICE_SCRIPT.read_text(encoding="utf-8")
    css = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.css").read_text(
        encoding="utf-8"
    )

    # Lazily created, fully removed, storage-free, single-capture overlay.
    assert "function openConversationOverlay()" in source
    assert "function removeConversationOverlay()" in source
    assert source.count("navigator.mediaDevices.getUserMedia(") == 1
    overlay_region = source.split("Conversation mode overlay", 1)[1].split(
        "function clearErrorTimer", 1
    )[0]
    assert "localStorage" not in overlay_region
    assert "track.enabled=!conversation.muted" in overlay_region
    assert "'aria-label':'Voice conversation'" in overlay_region
    assert "'aria-live':'polite'" in overlay_region
    assert "event.key==='Escape'" in overlay_region
    assert "event.key==='Tab'" in overlay_region

    # Orb styling: breathing, level-driven, playback-driven, reduced-motion.
    assert ".voice-conversation" in css
    assert "--conversation-level" in css
    assert "voice-conversation-breathe" in css
    assert ".voice-conversation.is-playing .voice-conversation-orb-halo" in css
    reduced = css.split("@media (prefers-reduced-motion: reduce)", 1)[1]
    assert ".voice-conversation *" in reduced


def test_endpointing_and_streaming_hardening_source_contract():
    source = VOICE_SCRIPT.read_text(encoding="utf-8")

    # Young utterances hold their endpoint past a thinking pause — unless the
    # streaming partial already reads as a plausibly complete utterance
    # (>=3 words or terminal punctuation), which endpoints at the base window.
    assert "const VAD_EARLY_SILENCE_MS=1800" in source
    assert "const VAD_COMMITTED_SPEECH_MS=1200" in source
    assert "latestPartial:function(){return lastPartialText;}" in source
    assert "const partialComplete=partialWords>=3||(partialWords>0" in source
    assert (
        "const endpointSilenceMs=(speechMs<VAD_COMMITTED_SPEECH_MS&&!partialComplete)"
        "?Math.max(silenceMs,VAD_EARLY_SILENCE_MS):silenceMs" in source
    )
    # Resume always reaches the wire after a server-length silence gap.
    assert "const SERVER_EOS_SILENCE_MS=650" in source
    assert "speechGapMs>=SERVER_EOS_SILENCE_MS" in source
    assert "if(settled||committed||!speculative) return;" not in source.split(
        "resume:function()", 1
    )[1].split("speculate:function()", 1)[0]
    # No speculative Whisper pass on a one-word fragment.
    assert "const SPECULATE_MIN_SPEECH_MS=700" in source
    assert "speechMs>=SPECULATE_MIN_SPEECH_MS" in source
    # Echo residue can no longer ratchet the onset threshold above speech.
    assert "Math.min(rms,speechThreshold)*0.06" in source


def test_stitch_gate_ignores_partial_assistant_output():
    source = VOICE_SCRIPT.read_text(encoding="utf-8")
    region = source.split("function recordPendingStitch()", 1)[1].split(
        "async function sendTranscript", 1
    )[0]
    assert "currentAssistantText()" not in region
    assert "pendingStitch={text:sent.text,at:Date.now(),cut:cut}" in region
    # Normal completion still clears the stitch context.
    assert (
        "if(isFinal){\n      // The completion callback marks a normally "
        "completed response" in source
    )


def test_three_word_partial_endpoints_at_base_silence(probe_results):
    scenario = probe_results["three_word_partial_endpoints_at_base_silence"]
    assert scenario["sends"] == ["alpha bravo charlie"]
    assert scenario["state"] == "thinking"


def test_two_word_young_utterance_keeps_the_hold(probe_results):
    scenario = probe_results["two_word_young_utterance_keeps_the_hold"]
    assert scenario["sendsEarly"] == []
    assert scenario["sends"] == ["alpha bravo"]


def test_errored_turn_is_never_spoken_and_capture_resyncs(probe_results):
    scenario = probe_results["errored_turn_is_not_spoken_and_capture_resyncs"]
    assert scenario["stateAfterError"] == "listening"
    assert scenario["labelAfterError"] == "Let’s try that again — listening"
    assert scenario["ttsDuringError"] == 0
    assert scenario["freshSessions"] >= 1
    # The follow-up utterance is sent alone: no stitch with the errored turn.
    assert scenario["sends"] == ["alpha", "bravo"]


def test_transient_provider_error_auto_retries_then_speaks(probe_results):
    """A broker 5xx/502 blip is retried in place, not dropped: the error is
    never spoken, the overlay stays in Thinking, the app's regenerate action is
    clicked once, and the recovered answer is spoken normally."""
    scenario = probe_results["transient_error_auto_retries_then_speaks"]
    assert scenario["stateAfterError"] == "thinking"
    assert scenario["labelAfterError"] == "Thinking…"
    assert scenario["clicksAfterError"] == 1
    # The raw provider error text is never sent to TTS.
    assert scenario["ttsDuringError"] == []
    assert all("502" not in text and "Error" not in text for text in scenario["ttsTexts"])
    # The regenerated answer is spoken and the turn ends in Speaking.
    assert "Controlled Unclassified Information." in scenario["ttsTexts"]
    assert scenario["state"] == "speaking"


def test_transient_provider_error_retry_is_bounded(probe_results):
    """A provider stuck returning a transient error is retried at most twice,
    then the turn is dropped and capture resyncs so the failure surfaces."""
    scenario = probe_results["transient_error_retry_is_bounded"]
    assert scenario["clicks"] == 2
    assert scenario["state"] == "listening"
    assert scenario["label"] == "Let’s try that again — listening"


def test_barge_cut_marker_is_one_bounded_line(probe_results):
    scenario = probe_results["barge_cut_marker_records_spoken_tail"]
    assert scenario["ttsTexts"][0] == "The first point is ready."
    assert scenario["sends"][1] == (
        "alpha bravo\n[voice interruption: you were cut off after "
        '"The first point is ready."]'
    )
    marker = scenario["sends"][1].split("\n", 1)[1]
    assert "\n" not in marker


def test_sticky_language_biases_next_stt_session(probe_results):
    scenario = probe_results["sticky_language_biases_next_stt_session"]
    assert scenario["startLanguages"][0] == "auto"
    assert "en" in scenario["startLanguages"]


def test_cut_marker_error_resync_and_speaking_cycles_source_contract():
    source = VOICE_SCRIPT.read_text(encoding="utf-8")

    # Cut marker: exactly one appended line, tail bounded to 120 characters.
    assert "function voiceCutMarker(cut)" in source
    assert "tail.length>120?tail.slice(tail.length-120):tail" in source
    assert '\\n[voice interruption: you were cut off after "' in source
    # Error envelopes are detected structurally and trigger a capture resync.
    assert "function handleAssistantResponseError(token)" in source
    assert "function resyncCapture(token,statusLabel)" in source
    assert "Let’s try that again — listening" in source
    assert "segment.dataset.error==='1'" in source
    assert ".provider-error-details" in source
    # A TRANSIENT provider error is auto-retried in place (bounded) via the app's
    # own regenerate action, never spoken, before the drop path runs.
    assert "function retryTransientResponse(token,turn)" in source
    assert "function errorTurnIsTransient(turn)" in source
    assert "const MAX_TRANSIENT_RETRIES=2" in source
    assert "if(transientRetryCount>=MAX_TRANSIENT_RETRIES) return false" in source
    assert "regenerateResponse" in source  # clicks the app's regenerate button
    # A cancellation with no in-flight stream never gates the send.
    assert "if(!cancellation.streamId){" in source
    # Speaking ⇄ thinking cycles inside one interim-message turn.
    assert "function scheduleSpeakingIdleFallback(turn,session)" in source
    assert "if(state==='speaking') setState('thinking')" in source
    assert "function playbackAudible()" in source
    # Body-only scraping: never the avatar letter, author name or status chip.
    assert "querySelectorAll('.msg-body')" in source
    assert "function readSegmentBody(segment)" in source
    assert "function readAssistantTurn(turn)" in source


def test_caption_regions_are_bounded_scrollable_and_follow_tail():
    source = VOICE_SCRIPT.read_text(encoding="utf-8")
    css = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.css").read_text(
        encoding="utf-8"
    )

    assert "function updateCaptionRegion(element,text,limit)" in source
    assert "function attachCaptionScroll(element)" in source
    assert "element.dataset.follow=gap<=24?'1':'0'" in source
    for token in (
        "overflow-y: auto",
        "overscroll-behavior: contain",
        "-webkit-overflow-scrolling: touch",
        "touch-action: pan-y",
        "max-height: 22vh",
        "max-height: 38vh",
        "mask-image",
    ):
        assert token in css


def test_reply_language_stickiness_source_contract():
    source = VOICE_SCRIPT.read_text(encoding="utf-8")

    assert "let sessionLanguage=''" in source
    # A user-forced conversation-mode language (FIX 3) overrides the sticky
    # auto-detected hint for the streaming STT session.
    assert "language:forcedLanguage||sessionLanguage||'auto'" in source
    assert "function strongReplyLanguage(text)" in source
    assert "function detectReplyLanguage(text)" in source
    assert "scheduleThinkingCues(token,language||sessionLanguage,thinkingTurnId)" in source
    # The sticky language resets with each hands-free session.
    assert source.count("sessionLanguage='';") >= 2


def test_conversation_language_override_forces_stt_and_voice(probe_results):
    """FIX 3: picking Russian in the overlay forces BOTH the streaming STT hint
    and the reply TTS voice for the session (overriding auto-detection), and Auto
    releases the override."""
    scenario = probe_results["language_override_forces_stt_and_voice"]
    assert scenario["overlayPresent"] is True
    assert scenario["ruItemPresent"] is True
    # Selection state is reflected accessibly (menuitemradio aria-checked).
    assert scenario["forcedChecked"] == "true"
    assert scenario["autoChecked"] == "false"
    assert scenario["btnForced"] is True
    # Every STT session opened after the override carries the forced hint, and an
    # ENGLISH reply is still spoken by the Russian voice — auto-detection is
    # fully overridden.
    assert scenario["startLanguages"][1:] == ["ru", "ru", "ru"]
    assert scenario["ttsLanguages"] == ["ru", "ru"]
    # Auto releases the override.
    assert scenario["releasedChecked"] == "true"
    assert scenario["btnForcedAfterAuto"] is False


def test_conversation_language_selector_source_and_style_contract():
    """FIX 3: the corner language control is present, accessible, session-only
    (no localStorage), and lists Auto + the server voice-map languages."""
    source = VOICE_SCRIPT.read_text(encoding="utf-8")
    css = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.css").read_text(encoding="utf-8")

    assert "let forcedLanguage=''" in source
    assert "const CONVERSATION_LANGUAGES=[" in source
    # Auto + en/es/ru mirror hermes-jetson-tts-server.py LANGUAGE_VOICE_MAP.
    for code in ("{code:'',label:'Auto'", "code:'en'", "code:'es'", "code:'ru'"):
        assert code in source
    assert "function selectConversationLanguage(code)" in source
    assert "aria-haspopup" in source and "'aria-expanded':'false'" in source
    assert "role:'menuitemradio'" in source
    # Forced language overrides both STT (start hint) and the reply voice: it is
    # the first argument-gated branch of the reply-language resolver.
    assert "language:forcedLanguage||sessionLanguage||'auto'" in source
    assert "const resolved=resolveReplyLanguage(text,turn.sttLanguage,forcedLanguage)" in source
    assert "function resolveReplyLanguage(text,sttLanguage,forced)" in source
    assert "if(forced) return forced;" in source
    # Session-only: the forced language never touches localStorage.
    assert "localStorage" not in source.split("CONVERSATION_LANGUAGES", 1)[1].split("function openConversationOverlay", 1)[0]
    # Escape peels the menu before exiting; the language button joins the trap.
    assert "if(conversation.langMenu&&!conversation.langMenu.hidden){closeLanguageMenu(true);return;}" in source
    assert "[conversation.langBtn,outVisible?conversation.outBtn:null,conversation.muteBtn,conversation.exitBtn]" in source
    for token in (".voice-conversation-lang-btn", ".voice-conversation-lang-menu", "menuitemradio"):
        pass  # menu roles live in JS; assert CSS hooks below
    for token in (".voice-conversation-lang", ".voice-conversation-lang-btn", ".voice-conversation-lang-menu", ".voice-conversation-lang-item"):
        assert token in css


def test_conversation_orb_hermes_mark_source_and_style_contract():
    """FIX 1: the orb watermark is the Hermes CHARACTER art, a pre-processed
    data URI feathered to a circle so the orb crops it and the box disappears,
    keeping the character's facial features. It is a static, low-opacity layer
    that scales with the orb, uses a normal blend (the art itself, not a washed-
    out glow), respects reduced motion and never animates — NOT the muddy
    hermes-agent-192.png box it used to reference, and NOT the caduceus/staff SVG
    before that."""
    source = VOICE_SCRIPT.read_text(encoding="utf-8")
    css = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.css").read_text(encoding="utf-8")

    # The span is still created, but no inline SVG staff markup remains and the
    # character mark now comes entirely from CSS.
    assert "'voice-conversation-orb-mark'" in source
    assert "HERMES_MARK_SVG" not in source
    assert 'fill-rule="evenodd"' not in source  # the old caduceus path is removed
    assert "orbMark.innerHTML" not in source

    assert ".voice-conversation-orb-mark" in css
    mark_rule = css.split(".voice-conversation-orb-mark {", 1)[1].split("}", 1)[0]
    # The character art is inlined as a data URI (no static asset reference) and
    # painted as a scaling, centred, feather-cropped watermark with no animation.
    assert 'url("data:image/png;base64,' in mark_rule
    assert "mix-blend-mode: normal" in mark_rule
    assert "background-size: contain" in mark_rule
    assert "background-position: center" in mark_rule
    assert "background-repeat: no-repeat" in mark_rule
    assert "opacity: 0.5" in mark_rule
    assert "animation" not in mark_rule
    # The muddy raster box (hermes-agent-192.png) is gone from the whole sheet, as
    # is the staff-era monochrome currentColor tint.
    assert "url(hermes-agent-192.png)" not in css
    assert "color: rgba(233, 244, 255, 0.9)" not in css


def test_conversation_thinking_working_affordance_contract():
    """FIX 4: a delayed 'working…' affordance during silent Thinking, CSS-driven
    and reduced-motion aware, with no fabricated spoken acknowledgement."""
    source = VOICE_SCRIPT.read_text(encoding="utf-8")
    css = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.css").read_text(encoding="utf-8")

    assert "const AWAITING_AFFORDANCE_MS=2500" in source
    assert "function updateAwaitingAffordance(next)" in source
    assert "conversation.root.classList.add('is-awaiting')" in source
    # The affordance disarms the instant any reply text arrives.
    assert "if(text&&String(text).trim()) clearAwaitingAffordance();" in source
    assert ".voice-conversation.is-awaiting" in css
    # Reduced motion suppresses every overlay animation, including this one.
    rm_block = css.split("@media (prefers-reduced-motion: reduce)", 1)[1]
    assert "animation: none !important" in rm_block