atlas-iac/testing/tests/test_hermes_agent_runtime_patches.py

"""Pinned operator tooling and runtime patch contracts."""

from __future__ import annotations

from testing.tests.test_hermes_cli_support import (
    HERMES,
    Path,
    SCRIPTS,
    _agent_deployment,
    auth_patch,
    client_config,
    codex_runtime_patch,
    json,
    pytest,
    ttyd_patch,
    tui_gateway_patch,
    yaml,
)


def test_owner_agent_installs_the_pinned_operator_toolchain():
    script = (SCRIPTS / "install_agent_tools.sh").read_text()
    for value in [
        "flux",
        "helm",
        "kustomize",
        "jq",
        "yq",
        "gh",
        "vault",
        "sops",
        "age",
        "age-keygen",
        "k9s",
        "terraform",
        "go",
        "gofmt",
    ]:
        assert value in script
    # The toolchain install resolves the running node's arch instead of a
    # hardcoded one, so the Go URL is templated on ${dl_arch} and both arches'
    # pinned Go checksums are present.
    assert "go1.26.5.linux-${dl_arch}.tar.gz" in script
    assert "dl_arch=arm64" in script
    assert "dl_arch=amd64" in script
    assert (
        "fe4789e92b1f33358680864bbe8704289e7bb5fc207d80623c308935bd696d49"
        in script
    )
    assert (
        "5c2c3b16caefa1d968a94c1daca04a7ca301a496d9b086e17ad77bb81393f053"
        in script
    )
    assert script.count("sha256sum -c -") == 1

    deployment = _agent_deployment()
    installer = next(
        item
        for item in deployment["spec"]["template"]["spec"]["initContainers"]
        if item["name"] == "install-agent-tools"
    )
    assert "/bin/sh /opt/coordinator/install_agent_tools.sh" in installer["command"][2]
    assert any(mount["name"] == "coordinator" for mount in installer["volumeMounts"])
    init_config = next(
        item
        for item in deployment["spec"]["template"]["spec"]["initContainers"]
        if item["name"] == "init-config"
    )
    init_command = init_config["command"][2]
    assert "# Hermes managed operator PATH." in init_command
    assert "/opt/coordinator:/opt/data/tools/bin:/opt/hermes/.venv/bin" in init_command
    assert 'chmod 0644 "${profile_file}"' in init_command


def test_owner_agent_uses_the_canonical_hostname():
    paths = [
        HERMES / "agent-configmap.yaml",
        HERMES / "agent-deployment.yaml",
        HERMES / "agent-ingress.yaml",
        Path(__file__).parents[2] / "scripts/ops/hermes_triage_monitor.py",
    ]
    for path in paths:
        content = path.read_text()
        assert "agent.bstein.dev" not in content
        assert "worker.bstein.dev" in content


def test_legacy_owner_host_stays_served_until_it_is_retired_separately():
    """The rename must not cut off the host the operator reaches Hermes on.

    ``agent.hermes.bstein.dev`` keeps resolving, routing and validating TLS
    until it is retired in its own change, so rolling the rename back never
    needs more than reverting the oauth2-proxy callback.
    """
    legacy = "agent.hermes.bstein.dev"
    canonical = "worker.bstein.dev"

    coredns = (
        Path(__file__).parents[2] / "infrastructure/core/coredns-custom.yaml"
    ).read_text()
    assert f"192.168.22.9 {legacy}" in coredns
    assert f"192.168.22.9 {canonical}" in coredns

    certificate = yaml.safe_load((HERMES / "agent-certificate.yaml").read_text())
    assert {legacy, canonical} <= set(certificate["spec"]["dnsNames"])

    ingresses = [
        doc
        for doc in yaml.safe_load_all((HERMES / "agent-ingress.yaml").read_text())
        if doc
        and doc.get("kind") == "Ingress"
        and doc["metadata"]["name"].startswith("hermes-agent-")
    ]
    assert len(ingresses) == 2
    for ingress in ingresses:
        assert {legacy, canonical} <= {rule["host"] for rule in ingress["spec"]["rules"]}
        for tls in ingress["spec"]["tls"]:
            assert {legacy, canonical} <= set(tls["hosts"])


def test_agent_reconnect_retains_complete_history_and_long_tool_budget():
    configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
    config = yaml.safe_load(configmap["data"]["config.yaml"])
    display = config["display"]
    assert display["resume_exchanges"] >= 10000
    assert display["resume_max_user_chars"] >= 10000000
    assert display["resume_max_assistant_chars"] >= 10000000
    assert config["agent"]["max_turns"] == 180
    assert config["delegation"]["max_iterations"] == 120


def test_auth_patch_honors_explicit_shared_store(tmp_path: Path):
    source = tmp_path / "auth.py"
    destination = tmp_path / "patched/auth.py"
    source.write_text(
        'from pathlib import Path\nimport os\n\ndef _auth_file_path() -> Path:\n    path = get_hermes_home() / "auth.json"\n    return path\n',
        encoding="utf-8",
    )
    auth_patch.patch(source, destination)
    content = destination.read_text()
    assert 'os.environ.get("HERMES_AUTH_FILE"' in content


def test_auth_patch_fails_closed_on_upstream_drift(tmp_path: Path):
    source = tmp_path / "auth.py"
    source.write_text("def changed():\n    pass\n", encoding="utf-8")
    with pytest.raises(RuntimeError, match="context changed"):
        auth_patch.patch(source, tmp_path / "patched.py")


def test_codex_runtime_patch_uses_cli_and_forwards_route(tmp_path: Path):
    provider = tmp_path / "runtime_provider.py"
    provider.write_text(codex_runtime_patch.PROVIDER_BEFORE, encoding="utf-8")
    provider_out = tmp_path / "patched/runtime_provider.py"
    codex_runtime_patch.patch_provider(provider, provider_out)
    assert '"api_mode": "codex_app_server"' in provider_out.read_text()

    session = tmp_path / "codex_app_server_session.py"
    session.write_text(
        codex_runtime_patch.SESSION_SIGNATURE_BEFORE
        + codex_runtime_patch.SESSION_REQUEST_BEFORE,
        encoding="utf-8",
    )
    session_out = tmp_path / "patched/codex_app_server_session.py"
    codex_runtime_patch.patch_session(session, session_out)
    session_content = session_out.read_text()
    assert 'turn_params["model"] = model' in session_content
    assert 'turn_params["effort"] = effort' in session_content
    assert '"approvalPolicy": "never"' in session_content
    assert '"sandboxPolicy": {"type": "dangerFullAccess"}' in session_content

    turn = tmp_path / "codex_runtime.py"
    turn.write_text(
        codex_runtime_patch.FALLBACK_CONTEXT_BEFORE
        + codex_runtime_patch.TURN_BEFORE,
        encoding="utf-8",
    )
    turn_out = tmp_path / "patched/codex_runtime.py"
    codex_runtime_patch.patch_turn(turn, turn_out)
    turn_content = turn_out.read_text()
    assert "model=str(getattr(agent" in turn_content
    assert "build_cross_provider_codex_prompt" in turn_content

    fallback = tmp_path / "chat_completion_helpers.py"
    fallback.write_text(
        codex_runtime_patch.FALLBACK_RESOLUTION_BEFORE,
        encoding="utf-8",
    )
    fallback_out = tmp_path / "patched/chat_completion_helpers.py"
    codex_runtime_patch.patch_fallback(fallback, fallback_out)
    fallback_content = fallback_out.read_text()
    assert 'agent.api_mode = "codex_app_server"' in fallback_content
    assert "agent._codex_cross_provider_fallback = True" in fallback_content

    loop = tmp_path / "conversation_loop.py"
    loop.write_text(
        codex_runtime_patch.FALLBACK_DISPATCH_BEFORE
        + codex_runtime_patch.RETRY_FALLBACK_DISPATCH_BEFORE
        + codex_runtime_patch.STREAM_RECOVERY_BEFORE,
        encoding="utf-8",
    )
    loop_out = tmp_path / "patched/conversation_loop.py"
    codex_runtime_patch.patch_loop(loop, loop_out)
    loop_content = loop_out.read_text()
    assert loop_content.count('if agent.api_mode == "codex_app_server"') == 2
    assert "build_cross_provider_codex_prompt" in loop_content
    retry_dispatch = loop_content.index(
        "Fallback activation happens inside this retry loop"
    )
    api_kwargs = loop_content.find("agent._build_api_kwargs", retry_dispatch)
    assert api_kwargs == -1 or retry_dispatch < api_kwargs
    assert "Provider stream ended before a complete response" in loop_content
    assert "_is_transport_stub" in loop_content
    assert "rerouting ({truncated_tool_call_retries}/4)" in loop_content

    auxiliary = tmp_path / "auxiliary_client.py"
    auxiliary.write_text(
        codex_runtime_patch.AUXILIARY_TOKEN_BEFORE,
        encoding="utf-8",
    )
    auxiliary_out = tmp_path / "patched/auxiliary_client.py"
    codex_runtime_patch.patch_auxiliary(auxiliary, auxiliary_out)
    auxiliary_content = auxiliary_out.read_text()
    assert 'os.environ.get("CODEX_HOME"' in auxiliary_content
    assert 'Path(codex_home).expanduser() / "auth.json"' in auxiliary_content
    assert "never creates a metered API-key lane" in auxiliary_content


def test_agent_mounts_codex_auxiliary_runtime_patch():
    deployment = _agent_deployment()
    pod = deployment["spec"]["template"]["spec"]
    patch_init = next(
        item for item in pod["initContainers"]
        if item["name"] == "patch-codex-runtime"
    )
    assert patch_init["command"][-2:] == [
        "/opt/hermes/agent/auxiliary_client.py",
        "/patched/auxiliary_client.py",
    ]
    expected_mount = {
        "name": "codex-runtime-patch",
        "mountPath": "/opt/hermes/agent/auxiliary_client.py",
        "subPath": "auxiliary_client.py",
    }
    containers = {item["name"]: item for item in pod["containers"]}
    for name in ("hermes", "terminal"):
        assert expected_mount in containers[name]["volumeMounts"]


def test_codex_auxiliary_patch_reads_cli_token_without_copying_it(
    tmp_path: Path,
    monkeypatch,
):
    codex_home = tmp_path / ".codex"
    codex_home.mkdir()
    (codex_home / "auth.json").write_text(
        json.dumps({"tokens": {"access_token": "cli-access-token"}}),
        encoding="utf-8",
    )
    monkeypatch.setenv("CODEX_HOME", str(codex_home))

    source = tmp_path / "auxiliary_client.py"
    source.write_text(
        "import json, logging, os, time\n"
        "from pathlib import Path\n"
        "logger = logging.getLogger(__name__)\n"
        "def read_token():\n"
        "    try:\n"
        "        raise RuntimeError('Hermes provider store intentionally empty')\n"
        + codex_runtime_patch.AUXILIARY_TOKEN_BEFORE,
        encoding="utf-8",
    )
    destination = tmp_path / "patched/auxiliary_client.py"
    codex_runtime_patch.patch_auxiliary(source, destination)
    namespace: dict = {}
    exec(compile(destination.read_text(), str(destination), "exec"), namespace)

    assert namespace["read_token"]() == "cli-access-token"


def test_codex_runtime_patch_fails_closed_on_upstream_drift(tmp_path: Path):
    source = tmp_path / "runtime_provider.py"
    source.write_text("def changed():\n    pass\n", encoding="utf-8")
    with pytest.raises(RuntimeError, match="context changed"):
        codex_runtime_patch.patch_provider(source, tmp_path / "patched.py")


def test_codex_runtime_migration_uses_owner_unsafe_mode(tmp_path: Path):
    config = tmp_path / "config.yaml"
    config.write_text("model: {}\n", encoding="utf-8")
    codex_home = tmp_path / ".codex"
    calls = []

    class Report:
        errors = []

        @staticmethod
        def summary():
            return "configured"

    def migrate(value, **kwargs):
        calls.append((value, kwargs))
        return Report()

    client_config.configure_codex_runtime(config, migrate, codex_home)

    assert calls[0][1]["default_permission_profile"] is None
    assert calls[0][1]["codex_home"] == codex_home
    content = (codex_home / "config.toml").read_text(encoding="utf-8")
    assert 'approval_policy = "never"' in content
    assert 'sandbox_mode = "danger-full-access"' in content
    assert "default_permissions" not in content


def test_codex_owner_permissions_replace_stale_profile(tmp_path: Path):
    config = tmp_path / "config.toml"
    config.write_text(
        'default_permissions = ":danger-no-sandbox"\n\n[features]\nhooks = true\n',
        encoding="utf-8",
    )

    client_config.configure_codex_owner_permissions(config)
    client_config.configure_codex_owner_permissions(config)

    content = config.read_text(encoding="utf-8")
    assert content.count(client_config.OWNER_PERMISSIONS_BEGIN) == 1
    assert content.count('approval_policy = "never"') == 1
    assert content.count('sandbox_mode = "danger-full-access"') == 1
    assert "default_permissions" not in content
    assert "[features]\nhooks = true" in content


def test_tui_gateway_patch_extends_and_bounds_agent_startup(tmp_path: Path):
    source = tmp_path / "server.py"
    destination = tmp_path / "patched/server.py"
    source.write_text(
        "import os\n\n" + tui_gateway_patch.BEFORE + "\ndef unchanged():\n    pass\n",
        encoding="utf-8",
    )

    tui_gateway_patch.patch(source, destination)

    content = destination.read_text(encoding="utf-8")
    assert "HERMES_TUI_AGENT_INIT_TIMEOUT_S" in content
    assert 'configured = 180.0' in content
    assert "return max(30.0, min(configured, 900.0))" in content
    assert "timeout: float | None = None" in content
    assert "ready.wait(timeout=wait_timeout)" in content
    assert "def unchanged():" in content


def test_tui_gateway_patch_fails_closed_on_upstream_drift(tmp_path: Path):
    source = tmp_path / "server.py"
    source.write_text("def changed():\n    pass\n", encoding="utf-8")
    with pytest.raises(RuntimeError, match="context changed"):
        tui_gateway_patch.patch(source, tmp_path / "patched.py")


def test_ttyd_clipboard_and_reconnect_patch_remain_enabled():
    source = '<html><body><script>document.execCommand("copy")</script></body></html>'
    content = ttyd_patch.patch_html(source)
    assert 'id="atlas-ttyd-clipboard"' in content
    assert "navigator.clipboard.writeText(text)" in content
    assert "class AtlasRecoveringWebSocket" in content
    assert "window.location.reload()" in content
    assert "event.stopImmediatePropagation()" in content


def test_ttyd_patch_fails_closed_on_upstream_drift():
    with pytest.raises(RuntimeError, match="context changed"):
        ttyd_patch.patch_html("<html><body>changed</body></html>")