atlas-iac/testing/tests/test_hermes_handsfree_stt.py

"""Browser-to-ffmpeg contracts for Hermes hands-free transcription uploads."""

from __future__ import annotations

import importlib.util
import json
import os
from pathlib import Path
import shutil
import subprocess
import sys


ROOT = Path(__file__).resolve().parents[2]
VOICE_JS = ROOT / "dockerfiles/hermes-webui-atlas-voice.js"
STT_PATCHER = ROOT / "dockerfiles/hermes-webui-stt-patch.py"
AGENT_FIXTURE = ROOT / "testing/fixtures/hermes-agent-9de9c25f"
MEDIA_FIXTURE = ROOT / "testing/fixtures/mediarecorder/chromium-webm-opus.json"
SAFE_CONVERSION_ERROR = (
    "Audio conversion failed: upload is invalid, incomplete, or uses an "
    "unsupported codec"
)


def _media_chunks() -> tuple[dict[str, object], list[bytes]]:
    fixture = json.loads(MEDIA_FIXTURE.read_text(encoding="utf-8"))
    payload = MEDIA_FIXTURE.with_name(str(fixture["payload_file"])).read_bytes()
    chunks = []
    offset = 0
    for size in fixture["chunk_sizes"]:
        end = offset + int(size)
        chunks.append(payload[offset:end])
        offset = end
    assert offset == len(payload)
    return fixture, chunks


def _late_speech_upload(*, preserve_header: bool) -> bytes:
    fixture, chunks = _media_chunks()
    count = int(fixture["pre_speech_chunk_count"])
    retained = chunks[count - 3 :]
    if preserve_header:
        retained.insert(0, chunks[0])
    return b"".join(retained)


def _patched_transcription_module(tmp_path: Path):
    target = tmp_path / "hermes-agent"
    shutil.copytree(AGENT_FIXTURE, target)
    env = os.environ.copy()
    env["HERMES_AGENT_PATCH_ROOT"] = str(target)
    subprocess.run(
        [sys.executable, str(STT_PATCHER)],
        cwd=ROOT,
        env=env,
        check=True,
        capture_output=True,
        text=True,
    )
    module_path = target / "tools/transcription_tools.py"
    spec = importlib.util.spec_from_file_location(
        "pinned_transcription_tools", module_path
    )
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    return module


def _convert(module, tmp_path: Path, payload: bytes, suffix: str):
    source = tmp_path / f"voice-input{suffix}"
    output_dir = tmp_path / f"converted-{suffix.removeprefix('.')}"
    output_dir.mkdir()
    source.write_bytes(payload)
    return module._prepare_local_audio(str(source), str(output_dir))


def test_fixture_reproduces_the_current_ebml_header_failure(tmp_path: Path):
    """The red control is browser data after the old sliding pre-roll dropped chunk 0."""
    broken = tmp_path / "current-late-speech.webm"
    broken.write_bytes(_late_speech_upload(preserve_header=False))

    result = subprocess.run(
        ["ffmpeg", "-v", "error", "-y", "-i", str(broken), str(tmp_path / "bad.wav")],
        check=False,
        capture_output=True,
        text=True,
        timeout=15,
    )

    assert result.returncode != 0
    assert "EBML header" in result.stderr


def test_header_preserved_browser_webm_reaches_real_conversion_boundary(
    tmp_path: Path,
):
    module = _patched_transcription_module(tmp_path)
    prepared, error = _convert(
        module,
        tmp_path,
        _late_speech_upload(preserve_header=True),
        ".webm",
    )

    assert error is None
    assert prepared is not None
    output = Path(prepared)
    assert output.is_file()
    assert output.read_bytes().startswith(b"RIFF")


def test_invalid_browser_upload_returns_bounded_error_without_ffmpeg_spam(
    tmp_path: Path,
):
    module = _patched_transcription_module(tmp_path)
    prepared, error = _convert(
        module,
        tmp_path,
        _late_speech_upload(preserve_header=False),
        ".webm",
    )

    assert prepared is None
    assert error == SAFE_CONVERSION_ERROR
    assert len(error) < 128
    lowered = error.lower()
    assert "ffmpeg version" not in lowered
    assert "configuration:" not in lowered
    assert "/tmp/" not in lowered


def test_local_conversion_accepts_browser_fallback_containers(tmp_path: Path):
    module = _patched_transcription_module(tmp_path)
    formats = ((".ogg", "libopus"), (".mp4", "aac"))

    for suffix, codec in formats:
        source = tmp_path / f"source{suffix}"
        subprocess.run(
            [
                "ffmpeg",
                "-v",
                "error",
                "-y",
                "-f",
                "lavfi",
                "-i",
                "sine=frequency=440:duration=0.4",
                "-c:a",
                codec,
                str(source),
            ],
            check=True,
            capture_output=True,
            text=True,
            timeout=15,
        )
        prepared, error = _convert(module, tmp_path, source.read_bytes(), suffix)
        assert error is None
        assert prepared is not None
        assert Path(prepared).read_bytes().startswith(b"RIFF")


def test_browser_uses_actual_recorder_mime_and_supported_extensions():
    source = VOICE_JS.read_text(encoding="utf-8")

    assert "let initialChunk=null" in source
    assert "if(initialChunk){chunks.push(initialChunk);initialChunk=null;}" in source
    assert "let recordedMime=recorder.mimeType||mime||''" in source
    assert "if(event.data.type) recordedMime=event.data.type" in source
    assert "audio/mp4;codecs=mp4a.40.2" in source
    assert "if(normalized.indexOf('ogg')>=0) return 'ogg'" in source
    assert "if(normalized.indexOf('mp4')>=0) return 'mp4'" in source


def test_stt_patch_is_built_fail_closed_into_the_webui_image(tmp_path: Path):
    dockerfile = (ROOT / "dockerfiles/Dockerfile.hermes-webui").read_text(
        encoding="utf-8"
    )
    assert "COPY dockerfiles/hermes-webui-stt-patch.py" in dockerfile
    assert "python /tmp/hermes-webui-stt-patch.py" in dockerfile
    assert "/opt/hermes/tools/transcription_tools.py" in dockerfile

    drifted = tmp_path / "drifted-agent"
    shutil.copytree(AGENT_FIXTURE, drifted)
    transcription = drifted / "tools/transcription_tools.py"
    source = transcription.read_text(encoding="utf-8")
    transcription.write_text(
        source.replace("Failed to convert audio for local STT", "upstream drift", 1),
        encoding="utf-8",
    )
    env = os.environ.copy()
    env["HERMES_AGENT_PATCH_ROOT"] = str(drifted)
    result = subprocess.run(
        [sys.executable, str(STT_PATCHER)],
        cwd=ROOT,
        env=env,
        check=False,
        capture_output=True,
        text=True,
    )
    assert result.returncode != 0
    assert "patch context changed" in result.stderr