"""Operator toolchain and runtime-patch contracts for Hermes CLI lanes."""

from __future__ import annotations

import json
from pathlib import Path

import pytest
import yaml

from testing.tests.test_hermes_cli_lanes_support import (
    HERMES,
    SCRIPTS,
    _agent_deployment,
    auth_patch,
    client_config,
    codex_runtime_patch,
    tui_gateway_patch,
    ttyd_patch,
)


def test_worker_route_broker_accepts_pod_network_health_checks():
    """Kubelet probes the pod IP, so the broker cannot bind to loopback only."""
    script = (SCRIPTS / "worker_route_broker.py").read_text()
    assert 'ThreadingHTTPServer(("0.0.0.0", PORT), Handler)' in script


def test_switchyard_network_boundary_allows_vault_bootstrap():
    """The pre-populate init container must reach Vault before routing starts."""
    documents = [
        item
        for item in yaml.safe_load_all((HERMES / "networkpolicy.yaml").read_text())
        if item
    ]
    isolation = next(
        item
        for item in documents
        if item.get("metadata", {}).get("name") == "hermes-switchyard-isolation"
    )
    assert any(
        rule.get("to")
        == [
            {
                "namespaceSelector": {
                    "matchLabels": {"kubernetes.io/metadata.name": "vault"}
                },
                "podSelector": {"matchLabels": {"app": "vault"}},
            }
        ]
        and rule.get("ports") == [{"protocol": "TCP", "port": 8200}]
        for rule in isolation["spec"]["egress"]
    )


def test_switchyard_network_boundary_allows_metrics_scraping():
    """VictoriaMetrics may scrape Switchyard without widening its API boundary."""
    documents = [
        item
        for item in yaml.safe_load_all((HERMES / "networkpolicy.yaml").read_text())
        if item
    ]
    isolation = next(
        item
        for item in documents
        if item.get("metadata", {}).get("name") == "hermes-switchyard-isolation"
    )
    assert any(
        rule.get("from")
        == [
            {
                "namespaceSelector": {
                    "matchLabels": {"kubernetes.io/metadata.name": "monitoring"}
                },
                "podSelector": {"matchLabels": {"app": "server"}},
            }
        ]
        and {entry.get("port") for entry in rule.get("ports", [])} == {9005, 9009}
        for rule in isolation["spec"]["ingress"]
    )


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
    # Arch-aware: the Go URL is templated on ${dl_arch} and both arches' pinned
    # Go checksums are present so amd64 nodes install native binaries.
    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_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>")