atlas-iac/testing/tests/test_hermes_chat_quality.py

"""Contracts for isolated high-quality Hermes chat capabilities."""

from __future__ import annotations

import importlib.util
from pathlib import Path

import yaml


ROOT = Path(__file__).parents[2]
HERMES = ROOT / "services" / "hermes"


def _documents(path: Path) -> list[dict]:
    return [doc for doc in yaml.safe_load_all(path.read_text()) if doc]


def test_chat_config_enables_real_research_compute_and_delegation():
    configmap = _documents(HERMES / "chat-configmap.yaml")[0]
    config = yaml.safe_load(configmap["data"]["config.yaml"])

    assert config["agent"]["reasoning_effort"] == "high"
    assert config["web"] == {
        "backend": "ddgs",
        "search_backend": "ddgs",
        "extract_backend": "public-extract",
    }
    assert config["delegation"]["max_concurrent_children"] == 2
    assert config["delegation"]["max_iterations"] == 80
    assert config["agent"]["max_turns"] == 120
    assert config["tool_loop_guardrails"]["hard_stop_enabled"] is True
    for platform in ("cli", "api_server"):
        toolsets = config["platform_toolsets"][platform]
        assert "delegation" in toolsets
        assert "browser" in toolsets
        assert "python_sandbox" in toolsets
        assert "vision" in toolsets
        assert "web" in toolsets
        assert "terminal" not in toolsets
        assert "code_execution" not in toolsets


def test_sandbox_shares_only_the_tenant_workspace_without_credentials():
    sandbox_docs = _documents(HERMES / "chat-sandbox.yaml")
    deployments = [doc for doc in sandbox_docs if doc["kind"] == "Deployment"]
    assert len(deployments) == 4
    for ordinal, deployment in enumerate(deployments):
        pod_spec = deployment["spec"]["template"]["spec"]
        container = pod_spec["containers"][0]
        assert pod_spec["automountServiceAccountToken"] is False
        assert container["securityContext"]["readOnlyRootFilesystem"] is True
        assert container["securityContext"]["runAsNonRoot"] is True
        assert container["securityContext"]["runAsGroup"] == 10000
        assert not container.get("env")
        assert {mount["mountPath"] for mount in container["volumeMounts"]} == {
            "/tmp",
            "/workspace",
            "/opt/data/workspace",
        }
        workspace_volume = next(
            item for item in pod_spec["volumes"] if item["name"] == "workspace"
        )
        assert workspace_volume["persistentVolumeClaim"]["claimName"] == (
            f"workspace-hermes-chat-tenant-{ordinal}"
        )

    statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
    templates = statefulset["spec"]["volumeClaimTemplates"]
    workspace = next(item for item in templates if item["metadata"]["name"] == "workspace")
    assert workspace["spec"]["resources"]["requests"]["storage"] == "10Gi"
    assert workspace["spec"]["accessModes"] == ["ReadWriteMany"]

    pod_spec = statefulset["spec"]["template"]["spec"]
    hermes = next(item for item in pod_spec["containers"] if item["name"] == "hermes")
    startup = hermes["args"][0]
    assert "hermes-chat-sandbox-${ordinal}.hermes-chat-sandbox" in startup
    assert any(
        mount["name"] == "workspace" and mount["mountPath"] == "/opt/data/workspace"
        for mount in hermes["volumeMounts"]
    )

    policies = _documents(HERMES / "networkpolicy.yaml")
    deny = next(
        item
        for item in policies
        if item["metadata"]["name"] == "hermes-chat-sandbox-deny"
    )
    assert deny["spec"]["ingress"] == []
    assert deny["spec"]["egress"] == []
    for ordinal in range(4):
        policy = next(
            item
            for item in policies
            if item["metadata"]["name"] == f"hermes-chat-sandbox-tenant-{ordinal}"
        )
        assert policy["spec"]["podSelector"]["matchLabels"][
            "ai.bstein.dev/tenant-ordinal"
        ] == str(ordinal)
        source = policy["spec"]["ingress"][0]["from"][0]["podSelector"][
            "matchLabels"
        ]
        assert source["statefulset.kubernetes.io/pod-name"] == (
            f"hermes-chat-tenant-{ordinal}"
        )


def test_gateway_image_honors_ui_model_and_caps_reasoning():
    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-agent").read_text()
    assert "_resolve_request_route" in dockerfile
    assert 'allowed_providers = {"openai-codex", "anthropic"}' in dockerfile
    assert 'reasoning_effort=body.get("reasoning_effort")' in dockerfile
    assert 'reasoning_config = {"enabled": True, "effort": "xhigh"}' in dockerfile
    assert "ddgs==9.14.4" in dockerfile
    assert "specific not in _LEGACY_WEB_BACKENDS" in dockerfile


def test_chat_oauth_allows_stale_service_worker_retirement():
    documents = _documents(HERMES / "oauth2-proxy.yaml")
    deployment = next(
        document
        for document in documents
        if document["kind"] == "Deployment"
        and document["metadata"]["name"] == "oauth2-proxy-hermes-chat"
    )
    args = deployment["spec"]["template"]["spec"]["containers"][0]["args"]

    assert "--skip-auth-route=GET=^/sw[.]js([?].*)?$" in args
    assert "--custom-templates-dir=/etc/oauth2-proxy/templates" in args

    template = (HERMES / "oauth2-proxy-templates" / "error.html").read_text()
    assert 'http-equiv="refresh"' in template
    assert "Unable to find a valid CSRF token" in template
    assert "expired or was already used" in template
    assert "/sign_in?rd=/" in template
    container = deployment["spec"]["template"]["spec"]["containers"][0]
    assert "v7.15.3@sha256:10a1165743a192e" in container["image"]
    assert "--cookie-csrf-per-request=true" in args
    assert "--cookie-csrf-per-request-limit=8" in args
    assert "--trusted-proxy-ip=10.42.0.0/16" in args
    assert "--api-route=^/api/" in args
    assert "--api-route=^/health$" in args
    assert "--cookie-expire=168h" in args


def test_webui_recovers_auth_and_labels_session_scoped_controls():
    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text()

    assert "res.status===401||res.status===403" in dockerfile
    assert "window.location.assign('/oauth2/start?rd='" in dockerfile
    assert "childrenExpanded?'▾ ':'▸ '" in dockerfile
    assert "profile default: ' + p.model" in dockerfile


def test_chat_voice_uses_private_jetson_services_and_shared_auto_route():
    statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
    containers = statefulset["spec"]["template"]["spec"]["containers"]
    webui = next(item for item in containers if item["name"] == "webui")
    env = {item["name"]: item["value"] for item in webui["env"]}

    assert env["HERMES_STT_URL"] == (
        "http://hermes-stt.hermes.svc.cluster.local:9000/v1/audio/transcriptions"
    )
    assert env["HERMES_WEBUI_ATLAS_TTS_URL"] == (
        "http://hermes-tts.hermes.svc.cluster.local:9001/v1/audio/speech"
    )
    assert "hermes_stt_client.py" in env["HERMES_LOCAL_STT_COMMAND"]

    configmap = _documents(HERMES / "chat-configmap.yaml")[0]
    config = yaml.safe_load(configmap["data"]["config.yaml"])
    assert config["stt"] == {
        "enabled": True,
        "provider": "local_command",
        "local": {"model": "large-v3-turbo", "language": "auto"},
    }

    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-webui").read_text()
    assert "hermes-webui-atlas-patch.py" in dockerfile
    assert "hermes-webui-atlas-voice.js" in dockerfile
    voice_script = (ROOT / "dockerfiles" / "hermes-webui-atlas-voice.js").read_text()
    assert "/api/transcribe/capability" in voice_script
    assert "/api/transcribe" in voice_script
    assert "/api/tts" in voice_script
    assert "speakResponse(generation)" in voice_script
    assert "restartSoon(token,450)" in voice_script


def test_voice_models_are_baked_and_runtime_has_no_public_egress():
    stt_dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-stt").read_text()
    tts_dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-jetson-tts").read_text()
    assert "ADD --checksum=sha256:aff26ae4" in stt_dockerfile
    assert "--chmod=0444" in stt_dockerfile
    assert "chmod 0555 /opt/models /opt/models/whisper" in stt_dockerfile
    assert "HERMES_STT_CACHE=/opt/models/whisper" in stt_dockerfile
    assert "ADD --checksum=sha256:4cabf7c3" in tts_dockerfile
    assert "ADD --checksum=sha256:db42b97d" in tts_dockerfile
    assert tts_dockerfile.count("--chmod=0444") == 2
    assert "chmod 0555 /opt/models /opt/models/piper" in tts_dockerfile
    assert "HERMES_TTS_CACHE=/opt/models/piper" in tts_dockerfile
    tts_server = (ROOT / "dockerfiles" / "hermes-jetson-tts-server.py").read_text()
    assert "download_voice" not in tts_server
    assert "baked Piper voice is missing" in tts_server

    policies = _documents(HERMES / "networkpolicy.yaml")
    voice_policy = next(
        item for item in policies if item["metadata"]["name"] == "hermes-private-voice"
    )
    assert voice_policy["spec"]["policyTypes"] == ["Ingress", "Egress"]
    assert not any(
        "ipBlock" in destination
        for rule in voice_policy["spec"]["egress"]
        for destination in rule.get("to", [])
    )


def test_voice_workloads_have_deliberate_xavier_placement():
    documents = _documents(HERMES / "voice-deployment.yaml")
    deployments = {
        item["metadata"]["name"]: item
        for item in documents
        if item["kind"] == "Deployment"
    }
    stt = deployments["hermes-stt"]["spec"]["template"]["spec"]
    tts = deployments["hermes-tts"]["spec"]["template"]["spec"]

    assert "@sha256:" in stt["containers"][0]["image"]
    assert "@sha256:" in tts["containers"][0]["image"]
    assert stt["nodeSelector"] == {"kubernetes.io/hostname": "titan-21"}
    assert tts["nodeSelector"] == {"kubernetes.io/hostname": "titan-20"}
    assert stt["automountServiceAccountToken"] is False
    assert tts["automountServiceAccountToken"] is False
    assert stt["enableServiceLinks"] is False
    assert tts["enableServiceLinks"] is False
    assert stt["runtimeClassName"] == "nvidia"
    stt_resources = stt["containers"][0]["resources"]
    assert stt_resources["requests"]["nvidia.com/gpu.shared"] == 1
    assert stt_resources["limits"]["nvidia.com/gpu.shared"] == 1
    assert "nvidia.com/gpu.shared" not in tts["containers"][0]["resources"]["requests"]
    assert all("hostPath" not in volume for volume in stt["volumes"])
    assert all("hostPath" not in volume for volume in tts["volumes"])


def test_chat_auth_file_mount_survives_atomic_provider_refresh():
    statefulset = _documents(HERMES / "chat-statefulset.yaml")[0]
    containers = statefulset["spec"]["template"]["spec"]["containers"]

    for name in ("hermes", "webui"):
        container = next(item for item in containers if item["name"] == name)
        env = {item["name"]: item["value"] for item in container["env"]}
        assert env["HERMES_AUTH_FILE"] == "/shared-auth/auth.json"
        mount = next(
            item for item in container["volumeMounts"] if item["name"] == "provider-auth"
        )
        assert mount["mountPath"] == "/shared-auth"
        assert "subPath" not in mount

    hermes_env = {
        item["name"]: item["value"]
        for item in next(item for item in containers if item["name"] == "hermes")["env"]
    }
    assert hermes_env["AGENT_BROWSER_EXECUTABLE_PATH"].endswith("/chrome-linux/headless_shell")
    assert "--no-sandbox" in hermes_env["AGENT_BROWSER_ARGS"]


def test_sandbox_executes_python_with_bounded_output(tmp_path: Path, monkeypatch):
    source = ROOT / "dockerfiles" / "hermes-chat-sandbox-server.py"
    spec = importlib.util.spec_from_file_location("hermes_chat_sandbox_server", source)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    monkeypatch.setattr(module, "WORKSPACE", tmp_path)

    result = module._execute("import math\nprint(math.comb(10, 3))")

    assert result["success"] is True
    assert result["stdout"] == "120\n"
    assert result["stderr"] == ""