atlas-iac/testing/tests/test_hermes_voice_preflight_delivery.py

"""Browser, proxy, and Flux delivery contracts for voice route preflight."""

from __future__ import annotations

import json
from types import SimpleNamespace

import yaml

from testing.tests.test_hermes_chat_support import ROOT, _documents
from testing.tests.test_hermes_voice_language_routing import patched_webui  # noqa: F401


HERMES = ROOT / "services" / "hermes"
VOICE = ROOT / "dockerfiles" / "hermes-webui-atlas-voice.js"


def test_webui_proxy_enforces_origin_exact_echo_and_no_store(
    patched_webui,  # noqa: F811
    monkeypatch,
):
    routes = patched_webui.routes
    request_payload = {
        "turn_id": "ef" * 16 + "-4-1",
        "revision": 2,
        "transcript": "Explain the safest non disruptive option",
    }
    monkeypatch.setattr(routes, "read_body", lambda _handler: request_payload, raising=False)
    monkeypatch.setattr(
        routes,
        "j",
        lambda _handler, payload, status=200, extra_headers=None: {
            "status": status,
            "payload": payload,
            "headers": extra_headers,
        },
        raising=False,
    )
    trusted = {"value": True}
    monkeypatch.setattr(
        routes,
        "_check_same_origin_browser_request",
        lambda _handler: trusted["value"],
        raising=False,
    )
    monkeypatch.setattr(
        routes,
        "bad",
        lambda _handler, message, status=400: {"status": status, "error": message},
        raising=False,
    )
    observed = {}

    class Upstream:
        def read(self, size):
            observed["size"] = size
            return json.dumps(
                {
                    "turn_id": request_payload["turn_id"],
                    "revision": 2,
                    "tier": "deep",
                    "target": "atlas/auto/deep",
                    "advisory": True,
                    "private": "must be removed",
                }
            ).encode()

        def __enter__(self):
            return self

        def __exit__(self, *args):
            return False

    def open_preflight(request, timeout):
        observed["url"] = request.full_url
        observed["body"] = json.loads(request.data)
        observed["timeout"] = timeout
        return Upstream()

    monkeypatch.setattr(routes, "_atlas_voice_preflight_open", open_preflight)
    result = routes._handle_atlas_voice_preflight(SimpleNamespace())
    assert observed["url"] == routes.ATLAS_VOICE_PREFLIGHT_URL
    assert observed["body"] == request_payload
    assert observed["timeout"] == 1.0
    assert result["headers"] == {"Cache-Control": "no-store"}
    assert result["payload"] == {
        "turn_id": request_payload["turn_id"],
        "revision": 2,
        "tier": "deep",
        "target": "atlas/auto/deep",
        "advisory": True,
    }
    trusted["value"] = False
    assert routes._handle_atlas_voice_preflight(SimpleNamespace())["status"] == 403


def test_browser_debounces_stable_partials_and_never_forces_final_route():
    source = VOICE.read_text(encoding="utf-8")
    partial = source.split("payload.type==='partial'", 1)[1].split(
        "payload.type==='final'", 1
    )[0]
    scheduler = source.split("function scheduleVoicePreflight", 1)[1].split(
        "function cancelThinkingCues", 1
    )[0]
    send = source.split("async function sendTranscript", 1)[1].split(
        "function audioExtension", 1
    )[0]

    assert "VOICE_PREFLIGHT_DEBOUNCE_MS=200" in source
    assert "window.crypto.getRandomValues(bytes)" in source
    assert "scheduleVoicePreflight(turnId,revision,stable)" in partial
    assert "cancelVoicePreflight(turnId)" in source
    assert "controller.abort()" in scheduler
    assert "advisory.turn_id!==turnId" in scheduler
    assert "advisory.revision!==revision" in scheduler
    assert "label.textContent='Listening · '+preview+' · '+tier" in scheduler
    assert "composer.value" not in scheduler
    assert "window.send" not in scheduler
    assert "advisory" not in send
    assert "atlas/auto/" not in send


def test_flux_wires_the_sibling_runtime_rollout_service_and_narrow_policy():
    kustomization = yaml.safe_load((HERMES / "kustomization.yaml").read_text())
    generator = next(
        item
        for item in kustomization["configMapGenerator"]
        if item["name"] == "hermes-coordinator"
    )
    assert "voice_route_preflight.py=scripts/voice_route_preflight.py" in generator["files"]

    deployment = _documents(HERMES / "switchyard-deployment.yaml")[0]
    template = deployment["spec"]["template"]
    assert template["metadata"]["annotations"]["ai.bstein.dev/config-rev"].endswith(
        "voice-route-preflight-v1"
    )
    classifier = next(
        item for item in template["spec"]["containers"] if item["name"] == "classifier-broker"
    )
    env = {item["name"]: item["value"] for item in classifier["env"]}
    assert env["HERMES_VOICE_PREFLIGHT_MODEL"] == "qwen2.5:14b-instruct-q4_0"

    service = _documents(HERMES / "switchyard-service.yaml")[0]
    ports = {item["name"]: item["port"] for item in service["spec"]["ports"]}
    assert ports["voice-preflight"] == 9009
    voice_port = next(
        item for item in service["spec"]["ports"] if item["name"] == "voice-preflight"
    )
    assert voice_port["targetPort"] == "voice-preflight"
    assert service["metadata"]["annotations"]["prometheus.io/port"] == "9009"

    policies = {
        item["metadata"]["name"]: item
        for item in _documents(HERMES / "networkpolicy.yaml")
    }
    ingress = policies["hermes-switchyard-isolation"]["spec"]["ingress"]
    preflight_rules = [
        rule
        for rule in ingress
        if any(port["port"] == 9009 for port in rule.get("ports", []))
    ]
    assert len(preflight_rules) == 2
    assert preflight_rules[0]["from"][0]["podSelector"]["matchLabels"] == {
        "app": "hermes-chat-tenant"
    }
    assert preflight_rules[1]["from"][0]["namespaceSelector"]["matchLabels"] == {
        "kubernetes.io/metadata.name": "monitoring"
    }

    chat = (HERMES / "chat-statefulset.yaml").read_text()
    assert "HERMES_WEBUI_VOICE_PREFLIGHT_URL" not in chat