"""Native provider health, API lineage, and runtime-image contracts.""" from __future__ import annotations import importlib.util import json from pathlib import Path from testing.tests.test_hermes_chat_support import ( HERMES, ROOT, _documents, _load_broker_module, ) 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"] )