atlas-iac/testing/tests/test_hermes_chat_provider_auth.py

"""Provider subscription and OAuth contracts for Hermes chat."""

from __future__ import annotations

import base64
import importlib.util
import json
import sys
import time
from pathlib import Path
from types import ModuleType

import pytest

from testing.tests.test_hermes_chat_support import (
    HERMES,
    ROOT,
    _documents,
    _load_broker_module,
)


def test_codex_broker_auth_and_request_contract(tmp_path: Path, monkeypatch):
    """The relay is bounded, stateless, and rejects unapproved models."""
    module = _load_broker_module("hermes_codex_broker", "codex_broker.py", monkeypatch)
    monkeypatch.setattr(module, "TOKEN", "relay-secret")

    assert module._authorized("Bearer relay-secret") is True
    assert module._authorized("Bearer wrong") is False
    assert module._real_model("route/codex/gpt-5.6-sol/xhigh") == "gpt-5.6-sol"
    payload = module._validate_payload(
        {
            "model": "gpt-5.6-terra",
            "input": "route this chat turn",
            "store": True,
            "stream": False,
            "max_output_tokens": 96,
            "max_completion_tokens": 96,
            "max_tokens": 96,
            "temperature": 0.7,
            "top_p": 0.9,
        }
    )
    assert payload["store"] is False
    assert payload["stream"] is True
    assert "max_output_tokens" not in payload
    assert "max_completion_tokens" not in payload
    assert "max_tokens" not in payload
    assert "temperature" not in payload
    assert "top_p" not in payload
    assert payload["input"] == [
        {
            "type": "message",
            "role": "user",
            "content": [{"type": "input_text", "text": "route this chat turn"}],
        }
    ]
    response_item = {
        "type": "message",
        "role": "user",
        "content": [{"type": "input_text", "text": "keep this item"}],
    }
    assert module._validate_payload({"model": "gpt-5.6-terra", "input": response_item})[
        "input"
    ] == [response_item]
    response_items = [response_item]
    assert (
        module._validate_payload({"model": "gpt-5.6-terra", "input": response_items})[
            "input"
        ]
        is response_items
    )
    image_items = [
        {
            "type": "message",
            "role": "user",
            "content": [
                {"type": "input_text", "text": "What color is this?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "data:image/png;base64,cHJpdmF0ZQ==",
                        "detail": "high",
                    },
                },
            ],
        }
    ]
    assert module._validate_payload({"model": "gpt-5.6-terra", "input": image_items})[
        "input"
    ][0]["content"][1] == {
        "type": "input_image",
        "image_url": "data:image/png;base64,cHJpdmF0ZQ==",
        "detail": "high",
    }
    switchyard_image_items = [
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "What color is this?"},
                {
                    "type": "image_url",
                    "image_url": {
                        "url": "data:image/png;base64,cHJpdmF0ZQ==",
                        "detail": "auto",
                    },
                },
            ],
        }
    ]
    assert module._validate_payload(
        {"model": "gpt-5.6-terra", "input": switchyard_image_items}
    )["input"][0]["content"][1] == {
        "type": "input_image",
        "image_url": "data:image/png;base64,cHJpdmF0ZQ==",
        "detail": "auto",
    }
    switchyard_base64_items = [
        {
            "role": "user",
            "content": [
                {"type": "input_text", "text": "What color is this?"},
                {
                    "type": "image",
                    "source": {
                        "type": "base64",
                        "media_type": "image/png",
                        "data": "cHJpdmF0ZQ==",
                    },
                },
            ],
        }
    ]
    normalized_base64 = module._validate_payload(
        {"model": "gpt-5.6-terra", "input": switchyard_base64_items}
    )["input"][0]["content"][1]
    assert normalized_base64 == {
        "type": "input_image",
        "image_url": "data:image/png;base64,cHJpdmF0ZQ==",
    }
    switchyard_enum_items = [
        {
            "role": "user",
            "content": [
                {
                    "type": "input_image",
                    "image_url": {
                        "type": "url",
                        "data": {
                            "url": "data:image/png;base64,cHJpdmF0ZQ==",
                            "detail": "high",
                        },
                    },
                }
            ],
        }
    ]
    nested_image = module._validate_payload(
        {"model": "gpt-5.6-terra", "input": switchyard_enum_items}
    )["input"][0]["content"][0]
    assert nested_image["image_url"] == "data:image/png;base64,cHJpdmF0ZQ=="
    assert nested_image["detail"] == "high"
    with pytest.raises(ValueError, match=r"non-empty Responses image URL.*str\[4\]"):
        module._validate_payload(
            {
                "model": "gpt-5.6-terra",
                "input": [
                    {
                        "type": "message",
                        "role": "user",
                        "content": [
                            {
                                "type": "input_image",
                                "image_url": {"detail": "high"},
                            }
                        ],
                    }
                ],
            }
        )
    routed = module._validate_payload(
        {
            "model": "route/codex/gpt-5.6-luna/low",
            "input": "use the low route",
            "stream": False,
        }
    )
    assert routed["model"] == "gpt-5.6-luna"
    with pytest.raises(ValueError, match="unsupported Codex model"):
        module._validate_payload({"model": "unapproved-model", "input": "hello"})
    with pytest.raises(ValueError, match="non-empty Responses input"):
        module._validate_payload({"model": "gpt-5.6-terra", "input": ""})
    with pytest.raises(ValueError, match="non-empty Responses input list"):
        module._validate_payload({"model": "gpt-5.6-terra", "input": []})

    completed = {
        "id": "resp_test",
        "object": "response",
        "status": "completed",
        "output": [],
    }
    completed_item = {
        "type": "message",
        "role": "assistant",
        "status": "completed",
        "content": [{"type": "output_text", "text": "done"}],
    }
    assert module._completed_response(
        [
            "event: response.created",
            'data: {"type":"response.created","response":{}}',
            "event: response.output_item.done",
            "data: "
            + json.dumps(
                {
                    "type": "response.output_item.done",
                    "output_index": 0,
                    "item": completed_item,
                }
            ),
            "event: response.completed",
            "data: "
            + json.dumps({"type": "response.completed", "response": completed}),
            "data: [DONE]",
        ]
    )["output"] == [completed_item]
    raw_stream = (
        "event: response.output_item.done\n"
        "data: "
        + json.dumps(
            {
                "type": "response.output_item.done",
                "output_index": 0,
                "item": completed_item,
            }
        )
        + "\n\nevent: response.completed\ndata: "
        + json.dumps({"type": "response.completed", "response": completed})
        + "\n\n"
    ).encode()
    normalized = module._normalized_stream(
        raw_stream, {**completed, "output": [completed_item]}
    ).decode()
    terminal_data = next(
        line for line in normalized.splitlines() if '"response.completed"' in line
    )
    assert json.loads(terminal_data.removeprefix("data: "))["response"]["output"] == [
        completed_item
    ]
    assert normalized.endswith("\n\n")
    streamed_function_item = {
        "type": "function_call",
        "name": "read_file",
        "status": "completed",
        "arguments": '{"path":"/tmp"}',
    }
    streamed_function_body = (
        "event: response.function_call_arguments.delta\n"
        'data: {"type":"response.function_call_arguments.delta",'
        '"item_id":"call_1","delta":"{\\"path\\":\\"/tmp\\"}"}\n\n'
        "event: response.function_call_arguments.done\n"
        'data: {"type":"response.function_call_arguments.done",'
        '"item_id":"call_1","arguments":"{\\"path\\":\\"/tmp\\"}"}\n\n'
        "event: response.output_item.done\n"
        'data: {"type":"response.output_item.done","output_index":0,'
        '"item":{"type":"function_call","name":"read_file",'
        '"arguments":"{\\"path\\":\\"/tmp\\"}"}}\n\n'
        "event: response.completed\n"
        "data: "
        + json.dumps(
            {
                "type": "response.completed",
                "response": {**completed, "output": [streamed_function_item]},
            }
        )
        + "\n\n"
    ).encode()
    normalized_function_stream = module._normalized_stream(
        streamed_function_body, {**completed, "output": [streamed_function_item]}
    ).decode()
    assert "response.function_call_arguments.delta" in normalized_function_stream
    assert "response.function_call_arguments.done" not in normalized_function_stream
    assert "response.output_item.done" not in normalized_function_stream
    normalized_terminal = next(
        line
        for line in normalized_function_stream.splitlines()
        if '"response.completed"' in line
    )
    assert (
        json.loads(normalized_terminal.removeprefix("data: "))["response"]["output"]
        == []
    )
    with pytest.raises(RuntimeError, match="retryable incomplete response"):
        module._completed_response(
            [
                "event: response.incomplete",
                'data: {"type":"response.incomplete","response":'
                '{"status":"incomplete","incomplete_details":'
                '{"reason":"max_output_tokens"}}}',
            ]
        )
    with pytest.raises(RuntimeError, match="provider unavailable"):
        module._completed_response(
            [
                "event: error",
                'data: {"type":"error","error":{"message":"provider unavailable"}}',
            ]
        )
    malformed_tool_item = {
        "type": "function_call",
        "name": "search_files",
        "status": "completed",
        "arguments": '{"path":"","offset":',
    }
    with pytest.raises(RuntimeError, match="malformed function arguments"):
        module._completed_response(
            [
                "event: response.output_item.done",
                "data: "
                + json.dumps(
                    {
                        "type": "response.output_item.done",
                        "output_index": 0,
                        "item": malformed_tool_item,
                    }
                ),
                "event: response.completed",
                "data: "
                + json.dumps({"type": "response.completed", "response": completed}),
            ]
        )
    valid_tool_item = {
        **malformed_tool_item,
        "arguments": '{"path":"","offset":0}',
    }
    assert module._completed_response(
        [
            "event: response.output_item.done",
            "data: "
            + json.dumps(
                {
                    "type": "response.output_item.done",
                    "output_index": 0,
                    "item": valid_tool_item,
                }
            ),
            "event: response.completed",
            "data: "
            + json.dumps({"type": "response.completed", "response": completed}),
        ]
    )["output"] == [valid_tool_item]
    with pytest.raises(RuntimeError, match="malformed function arguments"):
        module._completed_response(
            [
                "event: response.function_call_arguments.delta",
                'data: {"type":"response.function_call_arguments.delta",'
                '"item_id":"call_1","output_index":0,'
                '"delta":"{\\"path\\":\\"/tmp\\",\\"offset\\":"}',
                "event: response.completed",
                "data: "
                + json.dumps({"type": "response.completed", "response": completed}),
            ]
        )
    streamed_tool = module._completed_response(
        [
            "event: response.function_call_arguments.delta",
            'data: {"type":"response.function_call_arguments.delta",'
            '"item_id":"call_2","output_index":0,'
            '"delta":"{\\"path\\":\\"/tmp\\",\\"offset\\":"}',
            "event: response.function_call_arguments.done",
            'data: {"type":"response.function_call_arguments.done",'
            '"item_id":"call_2","output_index":0,'
            '"arguments":"{\\"path\\":\\"/tmp\\",\\"offset\\":0}"}',
            "event: response.completed",
            "data: "
            + json.dumps({"type": "response.completed", "response": completed}),
        ]
    )
    assert streamed_tool["status"] == "completed"

    auth_dir = tmp_path / ".codex"
    auth_dir.mkdir()
    # The token payload need only prove the broker reads CODEX_HOME directly.
    encoded = (
        base64.urlsafe_b64encode(json.dumps({"exp": time.time() + 3600}).encode())
        .decode()
        .rstrip("=")
    )
    (auth_dir / "auth.json").write_text(
        json.dumps({"tokens": {"access_token": f"header.{encoded}.signature"}})
    )
    monkeypatch.setenv("CODEX_HOME", str(auth_dir))
    assert module._access_token().startswith("header.")


def test_codex_broker_refreshes_and_persists_first_party_oauth(
    tmp_path: Path, monkeypatch
):
    """Expired ChatGPT OAuth refreshes in the canonical Codex CLI store."""
    module = _load_broker_module(
        "hermes_codex_refresh_broker", "codex_broker.py", monkeypatch
    )
    auth_dir = tmp_path / ".codex"
    auth_dir.mkdir()

    def jwt(expires_at: float) -> str:
        payload = (
            base64.urlsafe_b64encode(json.dumps({"exp": expires_at}).encode())
            .decode()
            .rstrip("=")
        )
        return f"header.{payload}.signature"

    expired = jwt(time.time() - 60)
    live = jwt(time.time() + 3600)
    auth_path = auth_dir / "auth.json"
    auth_path.write_text(
        json.dumps(
            {
                "auth_mode": "chatgpt",
                "tokens": {
                    "access_token": expired,
                    "refresh_token": "refresh-old",
                },
            }
        )
    )
    calls = []
    auth_module = ModuleType("hermes_cli.auth")

    def refresh(access_token, refresh_token, *, timeout_seconds):
        calls.append((access_token, refresh_token, timeout_seconds))
        return {
            "access_token": live,
            "refresh_token": "refresh-new",
            "last_refresh": "2026-08-12T20:00:00Z",
        }

    auth_module.refresh_codex_oauth_pure = refresh
    package = ModuleType("hermes_cli")
    package.auth = auth_module
    monkeypatch.setitem(sys.modules, "hermes_cli", package)
    monkeypatch.setitem(sys.modules, "hermes_cli.auth", auth_module)
    monkeypatch.setenv("CODEX_HOME", str(auth_dir))

    assert module._access_token() == live
    persisted = json.loads(auth_path.read_text())
    assert persisted["tokens"]["access_token"] == live
    assert persisted["tokens"]["refresh_token"] == "refresh-new"
    assert persisted["last_refresh"] == "2026-08-12T20:00:00Z"
    assert calls == [(expired, "refresh-old", 30.0)]
    assert auth_path.stat().st_mode & 0o777 == 0o600

    # A healthy token is reused, so repeated routed turns do not spend a
    # refresh token or create a second billing/authentication path.
    assert module._access_token() == live
    assert len(calls) == 1


def test_claude_broker_uses_native_subscription_without_api_billing(monkeypatch):
    """Claude traffic must use the native first-party CLI subscription lane."""
    module = _load_broker_module(
        "hermes_claude_broker", "claude_oauth_broker.py", monkeypatch
    )
    monkeypatch.setenv("ANTHROPIC_API_KEY", "must-not-leak")
    monkeypatch.setenv("CLAUDE_API_KEY", "must-not-leak")
    monkeypatch.setattr(
        module,
        "resolve_route",
        lambda route: "claude-fable-5" if "/fable/" in route else route,
    )

    model, effort = module._route(
        "route/claude/fable/xhigh", {"output_config": {"effort": "xhigh"}}
    )

    assert (model, effort) == ("claude-fable-5", "xhigh")
    assert "ANTHROPIC_API_KEY" not in module._claude_environment()
    assert "CLAUDE_API_KEY" not in module._claude_environment()
    assert module.CAPACITY_PATTERN.search("weekly usage limit exhausted")
def test_codex_native_health_overrides_historical_router_errors(
    tmp_path: Path, monkeypatch
):
    """Fresh first-party health is authoritative over old Switchyard probes."""
    plugin_path = HERMES / "plugins" / "auto-router" / "provider_status.py"
    spec = importlib.util.spec_from_file_location("hermes_provider_status", plugin_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    health_path = tmp_path / "codex.json"
    health_path.write_text(
        json.dumps(
            {
                "state": "available",
                "authenticated": True,
                "transport": "codex-chatgpt-subscription",
            }
        )
    )
    monkeypatch.setattr(module, "CODEX_HEALTH_PATH", health_path)
    monkeypatch.setattr(module, "CLAUDE_HEALTH_PATH", tmp_path / "missing.json")
    monkeypatch.setattr(
        module,
        "_get_json",
        lambda url: {"status": "ok"}
        if url.endswith("/health")
        else {
            "models": {
                "route/codex/terra/medium": {
                    "calls": 1,
                    "errors": 99,
                    "total_tokens": 12,
                }
            }
        },
    )
    monkeypatch.setattr(module, "_codex_account", lambda: {})
    monkeypatch.setattr(module, "_claude_account", lambda: {})

    codex = module.provider_status_payload()["providers"]["codex"]

    assert codex["errors"] == 99
    assert codex["state"] == "available"
    assert codex["native_health"]["transport"] == "codex-chatgpt-subscription"

def test_api_session_patch_accepts_parent_lineage(tmp_path: Path):
    """API-created workers must persist the originating Hermes session."""
    module_path = HERMES / "scripts" / "patch_api_server_sessions.py"
    spec = importlib.util.spec_from_file_location("patch_api_sessions", module_path)
    assert spec and spec.loader
    module = importlib.util.module_from_spec(spec)
    spec.loader.exec_module(module)
    source = tmp_path / "api_server.py"
    destination = tmp_path / "patched.py"
    source.write_text(
        "prefix\n"
        + module.BEFORE
        + "middle\n"
        + module.RUNS_BEFORE
        + "run body\n"
        + module.RUN_CLOSE_BEFORE
        + module.RESPONSES_SESSION_BEFORE
        + module.EVENT_CALLBACK_SIGNATURE_BEFORE
        + "callback docstring and push helper\n"
        + module.EVENT_CALLBACK_BODY_BEFORE
        + "tool start body\n"
        + module.EVENT_CALLBACK_END_BEFORE
        + module.EVENT_CALLBACK_CALL_BEFORE
        + module.RUN_SWEEP_BEFORE
        + "suffix\n",
        encoding="utf-8",
    )

    module.patch(source, destination)
    patched = destination.read_text(encoding="utf-8")

    assert "X-Hermes-Parent-Session-Id" in patched
    assert "parent_session_id=parent_session_id" in patched
    assert "Parent session not found" in patched
    assert "HERMES_API_DEFAULT_PARENT_MATCH_PREFIXES" in patched
    assert "user_message.startswith(default_prefixes)" in patched
    assert "session_parent_conflict" in patched
    assert "X-Hermes-Conversation-Platform" in patched
    assert "X-Hermes-Conversation-Title" in patched
    assert 'conversation_platform != "telegram"' in patched
    assert "db.record_gateway_session_peer(" in patched
    assert 'display_name="Telegram"' in patched
    assert "db.reopen_session(session_id)" in patched
    assert 'db.end_session(session_id, f"api_run_{terminal_status}")' in patched
    assert "def _record_run_activity(" in patched
    assert '"_thinking": "Hermes is reasoning"' in patched
    assert '"run.started": "Worker started"' in patched
    assert '"run.completed": "Worker completed"' in patched
    assert '"reasoning.available": "Hermes finished a reasoning step"' in patched
    assert '"subagent.progress": "Nested worker progress"' in patched
    assert "redact_sensitive_text" in patched
    assert 'getattr(os, "O_NOFOLLOW", 0)' in patched
    assert "os.fchmod(fd, 0o600)" in patched
    assert "session_id=session_id" in patched
    assert 'self._record_run_activity(session_id, "run.started")' in patched
    assert 'detail = tool_name if event_type in {' in patched
    assert 'if event_type == "subagent.tool"' in patched
    assert "_RUN_ACTIVITY_HEARTBEAT_SECONDS = 15.0" in patched
    assert 'heartbeats.get(session_id, 0.0)' in patched
    assert '"subagent.thinking",' in patched
    assert "Stream retention and run lifetime are separate" in patched
    assert 'terminal_status in {"completed", "failed", "cancelled"}' in patched
    assert patched.index("terminal_status = self._run_statuses") < patched.index(
        "self._active_run_tasks.pop(run_id, None)"
    )

def test_switchyard_brokers_and_native_claude_lane_use_the_right_images():
    """Thin brokers stay small while native Claude runs beside owner auth."""
    dockerfile = (ROOT / "dockerfiles" / "Dockerfile.hermes-switchyard-brokers").read_text()
    assert "httpx==0.28.1" in dockerfile
    assert "worker_route_broker.py" in dockerfile
    assert "routing_catalog.py" in dockerfile

    deployment = _documents(HERMES / "switchyard-deployment.yaml")[0]
    containers = {
        container["name"]: container
        for container in deployment["spec"]["template"]["spec"]["containers"]
    }
    expected = (
        "registry.bstein.dev/bstein/hermes-switchyard-brokers@"
        "sha256:ee7e95e060ef8083da505162d7e9030daba15fdd828cc047bbcbe6aa409d2083"
    )
    assert containers["worker-route-broker"]["image"] == expected
    assert containers["classifier-broker"]["image"] == expected
    assert "claude-oauth-broker" not in containers

    agent = _documents(HERMES / "agent-deployment.yaml")[0]
    agent_containers = {
        container["name"]: container
        for container in agent["spec"]["template"]["spec"]["containers"]
    }
    for container_name in ("hermes", "terminal"):
        container = agent_containers[container_name]
        environment = {item["name"]: item["value"] for item in container["env"]}
        mounts = {item["name"]: item for item in container["volumeMounts"]}
        assert environment["HERMES_ROUTING_CATALOG_PATH"] == "/routing-catalog/catalog.json"
        assert environment["HERMES_CODEX_HEALTH_PATH"] == "/opt/data/provider-health/codex.json"
        assert environment["HERMES_CLAUDE_HEALTH_PATH"] == "/opt/data/provider-health/claude.json"
        assert mounts["routing-catalog"]["mountPath"] == "/routing-catalog"
        assert mounts["routing-catalog"]["readOnly"] is True
    codex = agent_containers["codex-broker"]
    codex_environment = {item["name"]: item["value"] for item in codex["env"]}
    assert codex_environment["HERMES_CODEX_HEALTH_PATH"] == "/opt/data/provider-health/codex.json"
    claude = agent_containers["claude-broker"]
    assert claude["image"].startswith("registry.bstein.dev/bstein/hermes-agent@")
    assert "unset ANTHROPIC_API_KEY CLAUDE_API_KEY" in claude["args"][0]
    assert any(
        mount["name"] == "home" and mount["mountPath"] == "/opt/data"
        for mount in claude["volumeMounts"]
    )