"""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 when the user talked over the # response: the interrupted thought and the follow-up still form one # stitched message. assert scenario["sends"] == ["alpha", "alpha bravo charlie"] 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_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. assert "const VAD_EARLY_SILENCE_MS=1800" in source assert "const VAD_COMMITTED_SPEECH_MS=1200" in source assert ( "const endpointSilenceMs=speechMs=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()}" in region # Normal completion still clears the stitch context. assert ( "if(isFinal){\n // The completion callback marks a normally " "completed response" in source )