atlas-iac/testing/tests/test_hermes_voice_extraction.py

"""Assistant-response extraction contract for conversation-mode captions + TTS.

Round 3's caption/TTS extraction was validated only against a single synthetic
``{dataset:{rawText}}`` node, so it never exercised the REAL rendered assistant
turn and shipped a ``turn.textContent`` fallback that scraped the avatar "H", the
"Hermes" author name and the "Processed 13s" worklog chip into the conversation
caption and the one-ahead TTS synthesizer ("HHermesProcessed 13s"), and truncated
multi-sentence replies to their first sentence.

``hermes_voice_response_probe.js`` builds a faithful rendered turn matching the
live build-24 DOM (ui.js ``_createAssistantTurn`` / ``renderMessages``, messages.js
``ensureAssistantRow``) with a real ``querySelectorAll`` / ``closest`` and drives the
actual exported extraction + one-ahead chunker. These tests assert the three
symptoms are fixed and locked.
"""

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_response_probe.js"


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


def test_error_segment_does_not_discard_a_real_answer(results):
    scenario = results["error_segment_with_answer"]
    assert scenario["speaksAnswer"] is True, "a stray error segment discarded the real answer"
    assert scenario["error"] is False, "a turn with a real answer must not surface the error state"


def test_error_only_turn_still_reports_error(results):
    scenario = results["error_only_reports_error"]
    assert scenario["isEmpty"] is True
    assert scenario["error"] is True


def test_caption_is_answer_body_only(results):
    """Symptom 1: caption/TTS text must be the answer body, never row chrome."""
    scenario = results["finalized_multi_segment"]
    assert scenario["text"] == (
        "Sentence one is here. Sentence two follows it. And sentence three concludes."
    )
    assert scenario["leaksAvatar"] is False, "avatar 'H' / 'Hermes' leaked into caption"
    assert scenario["leaksProcessed"] is False, "'Processed 13s' worklog chip leaked"
    assert scenario["leaksReasoning"] is False, "reasoning leaked into the spoken answer"
    assert scenario["leaksInterim"] is False, "folded interim segment leaked"
    assert scenario["error"] is False


def test_every_sentence_reaches_the_tts_queue(results):
    """Symptom 2: the one-ahead chunker enqueues the WHOLE reply, not sentence one."""
    scenario = results["finalized_multi_segment"]
    assert scenario["chunkConsumedAll"] is True, "chunker left part of the reply unspoken"
    assert scenario["sentenceOnePresent"] is True
    assert scenario["sentenceTwoPresent"] is True
    assert scenario["sentenceThreePresent"] is True
    assert scenario["chunkCount"] >= 3


def test_presettle_worklog_only_extracts_empty(results):
    """Symptom 3 root: a pre-settle worklog-only frame must extract to '' so the
    completion pump retries rather than finalizing 'HHermesProcessed 13s' into
    'Listening' after the first sentence."""
    scenario = results["presettle_worklog_only"]
    assert scenario["isEmpty"] is True, f"chrome leaked pre-settle: {scenario['text']!r}"
    assert scenario["error"] is False


def test_live_streaming_reads_message_body(results):
    """While streaming, the live segment has no data-raw-text yet — the answer
    still comes from its .msg-body, never the row textContent."""
    scenario = results["live_streaming_reads_body"]
    assert scenario["matches"] is True
    assert scenario["leaksAvatar"] is False


def test_reasoning_and_worklog_chrome_excluded(results):
    scenario = results["chrome_excluded"]
    assert scenario["matches"] is True
    assert scenario["leaksReasoning"] is False


def test_extraction_grows_to_full_reply(results):
    """Extraction grows monotonically from the streaming partial to the full
    settled reply, and every settled sentence reaches the queue."""
    scenario = results["streaming_growth"]
    assert scenario["grows"] is True
    assert scenario["allFourChunked"] is True
    assert scenario["consumedAll"] is True