"""Agent configuration and dashboard contracts for Hermes CLI lanes.""" from __future__ import annotations import json import sys from contextlib import nullcontext from pathlib import Path from types import SimpleNamespace import pytest import yaml from testing.tests.test_hermes_cli_lanes_support import ( HERMES, _agent_deployment, client_config, lanes, migration, policy, ) def test_workspace_preparation_failure_durably_blocks_the_claim( tmp_path: Path, monkeypatch ): task = SimpleNamespace(id="t_bad_worktree", current_run_id=7, assignee="cli-auto") calls = [] class Connection: def close(self): return None fake_db = SimpleNamespace( scoped_current_board=lambda _board: nullcontext(), connect=lambda board: Connection(), get_task=lambda _conn, _task_id: task, worker_log_path=lambda _task_id, board: tmp_path / "worker.log", _resolve_worktree_workspace=lambda _task, board: (_ for _ in ()).throw( ValueError("not a Git repository") ), block_task=lambda *_args, **kwargs: calls.append(kwargs), ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) monkeypatch.setattr( lanes, "state_path", lambda _board, _task_id: tmp_path / "state.json" ) lanes.execute_claim("cassandra", "t_bad_worktree") assert calls[0]["kind"] == "capability" assert calls[0]["expected_run_id"] == 7 assert "not a Git repository" in calls[0]["reason"] def test_artifacts_cannot_escape_the_task_worktree(tmp_path: Path): workspace = tmp_path / "workspace" workspace.mkdir() inside = workspace / "report.json" outside = tmp_path / "auth.json" inside.write_text("{}\n", encoding="utf-8") outside.write_text("secret\n", encoding="utf-8") assert lanes.workspace_artifacts( workspace, ["report.json", str(outside), "missing.json"], ) == [str(inside)] def test_restart_provider_change_includes_explicit_workspace_handoff( tmp_path: Path, monkeypatch ): task = SimpleNamespace( id="t_resume", current_run_id=9, assignee="cli-auto", max_runtime_seconds=60, ) state_file = tmp_path / "state.json" state_file.write_text( json.dumps({"current_route": {"provider": "claude"}}), encoding="utf-8", ) (tmp_path / "worker.log").write_text("prior provider evidence", encoding="utf-8") prompts = [] class Connection: def close(self): return None fake_db = SimpleNamespace( scoped_current_board=lambda _board: nullcontext(), connect=lambda board: Connection(), get_task=lambda _conn, _task_id: task, worker_log_path=lambda _task_id, board: tmp_path / "worker.log", _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_resume"), set_branch_name=lambda *_args: None, set_workspace_path=lambda *_args: None, build_worker_context=lambda *_args: "resume objective", add_comment=lambda *_args: None, complete_task=lambda *_args, **_kwargs: None, block_task=lambda *_args, **_kwargs: None, ) monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db)) monkeypatch.setattr(lanes, "state_path", lambda _board, _task_id: state_file) monkeypatch.setattr( lanes, "select_route", lambda *_args, **_kwargs: lanes.Route( "codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, () ), ) monkeypatch.setattr( lanes, "git_handoff", lambda _workspace, output: f"HANDOFF:{output}", ) monkeypatch.setattr( lanes, "run_provider", lambda _route, prompt, *_args, **_kwargs: ( prompts.append(prompt) or lanes.ProcessResult( 0, "", { "status": "completed", "summary": "done", "changed_files": [], "tests_run": [], "artifacts": [], "blockers": [], }, False, ) ), ) lanes.execute_claim("cassandra", "t_resume") assert "HANDOFF:prior provider evidence" in prompts[0] def test_provider_commands_are_structured_unattended_and_capped(tmp_path: Path): route = lanes.Route( "codex", "gpt-5.6-sol", "xhigh", "codex-xhigh", "jetson", "vote", 1, () ) command = lanes._codex_command( route, "Work.", tmp_path, {}, tmp_path / "result.json" ) assert "--dangerously-bypass-approvals-and-sandbox" in command assert "--json" in command assert "--output-schema" in command assert 'model_reasoning_effort="xhigh"' in command claude_state = {"claude_session_id": "13864642-2985-4f91-bef5-53f145f878e8"} claude = lanes._claude_command( lanes.Route( "claude", "claude-opus-5", "xhigh", "claude-xhigh", "jetson", "vote", 1, () ), "Review.", claude_state, False, ) assert "--dangerously-skip-permissions" in claude assert "--output-format" in claude and "stream-json" in claude assert "--json-schema" in claude assert "--disallowedTools" in claude assert "Bash(kubectl apply *)" not in claude assert "Bash(flux reconcile *)" not in claude assert "max" not in claude def test_worker_contract_separates_review_findings_from_task_blockers(tmp_path: Path): prompt = lanes.build_prompt("Review the change.", tmp_path) assert "put defects and risks in findings" in prompt assert "blockers array must be empty whenever status is completed" in prompt assert "findings" in lanes.RESULT_SCHEMA["properties"] assert set(lanes.RESULT_SCHEMA["required"]) == set( lanes.RESULT_SCHEMA["properties"] ) assert ( "assigned task itself" in lanes.RESULT_SCHEMA["properties"]["blockers"]["description"] ) @pytest.mark.parametrize( "command", [ "git push --force origin main", "git reset --hard HEAD~1", "git clean -fd", ], ) def test_claude_pretool_hook_blocks_hard_denies(command: str): assert policy.denial_reason(command) def test_claude_pretool_hook_allows_normal_engineering(): assert policy.denial_reason("pytest -q testing/tests") is None assert policy.denial_reason("git push origin feature/hermes") is None assert policy.denial_reason("kubectl delete pod -n cassandra stuck-worker") is None assert policy.denial_reason("flux reconcile kustomization hermes") is None assert policy.denial_reason("vault kv get kv/atlas/hermes") is None def test_claude_settings_preserve_state_and_install_three_guardrail_layers( tmp_path: Path, ): state = tmp_path / ".claude.json" settings = tmp_path / "settings.json" state.write_text('{"promptQueueUseCount": 4}\n', encoding="utf-8") settings.write_text( json.dumps( { "theme": "dark", "permissions": { "deny": [ "Bash(kubectl apply *)", "Bash(flux reconcile *)", "Bash(vault kv *)", "Bash(custom-owner-rule *)", ] }, } ) + "\n", encoding="utf-8", ) client_config.configure_claude_state(state) client_config.configure_claude_settings(settings) state_value = json.loads(state.read_text()) settings_value = json.loads(settings.read_text()) assert state_value["promptQueueUseCount"] == 4 assert state_value["bypassPermissionsModeAccepted"] is True assert settings_value["theme"] == "dark" assert "Bash(git reset --hard *)" in settings_value["permissions"]["deny"] assert "Bash(custom-owner-rule *)" in settings_value["permissions"]["deny"] assert "Bash(kubectl apply *)" not in settings_value["permissions"]["deny"] assert "Bash(flux reconcile *)" not in settings_value["permissions"]["deny"] assert "Bash(vault kv *)" not in settings_value["permissions"]["deny"] hook = settings_value["hooks"]["PreToolUse"][0]["hooks"][0] assert "claude_command_policy.py" in hook["command"] def test_legacy_state_is_archived_without_removing_provider_transcripts(tmp_path: Path): session = tmp_path / "home/.config/herdr/session.json" session.parent.mkdir(parents=True) session.write_text( '{"agents":[{"agent":"claude","session_id":"abc"}]}', encoding="utf-8" ) binary = tmp_path / "tools/bin/herdr" binary.parent.mkdir(parents=True) binary.write_text("legacy", encoding="utf-8") (tmp_path / "home/.claude").mkdir() (tmp_path / "home/.codex").mkdir() archive = migration.archive_legacy_state(tmp_path) value = json.loads(archive.read_text()) assert value["legacy_session"]["agents"][0]["session_id"] == "abc" assert (tmp_path / "home/.claude").is_dir() assert (tmp_path / "home/.codex").is_dir() assert not binary.exists() assert not (tmp_path / "home/.config/herdr").exists() def test_agent_uses_one_native_kanban_control_plane(): configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text()) config = yaml.safe_load(configmap["data"]["config.yaml"]) assert config["model"] == { "provider": "atlas-switchyard", "default": "atlas/auto/balanced", "model": "atlas/auto/balanced", } assert config["kanban"]["dispatch_in_gateway"] is True assert config["kanban"]["default_assignee"] == "cli-auto" assert config["plugins"]["enabled"] == ["auto-router"] deployment = _agent_deployment() pod = deployment["spec"]["template"]["spec"] assert pod["enableServiceLinks"] is False names = {item["name"] for item in pod["containers"]} assert "cli-lane-runner" in names assert "terminal" in names assert not any("herdr" in name for name in names) rendered = (HERMES / "agent-deployment.yaml").read_text() assert "herdr server" not in rendered assert "herdr-dispatch" not in rendered def test_cli_lane_reserves_cpu_headroom_for_ui_and_auth(): deployment = _agent_deployment() containers = { item["name"]: item for item in deployment["spec"]["template"]["spec"]["containers"] } lane = containers["cli-lane-runner"] environment = {item["name"]: item["value"] for item in lane["env"]} assert environment["HERMES_CLI_LANE_CONCURRENCY"] == "2" assert lane["resources"] == { "requests": {"cpu": "100m", "memory": "256Mi"}, "limits": {"cpu": "2", "memory": "6Gi"}, } def test_agent_avoids_unhealthy_nodes_and_fits_its_remaining_capacity(): """Placement correction: keep the agent off nodes that cannot hold it. titan-04 is cordoned after repeated kernel undervoltage and kubelet failure, and titan-19 was probe/Longhorn unstable under worker load, so both must join the existing hard exclusions. That leaves titan-05 as the healthy candidate, which is tight enough on requested CPU that the main container has to give back 50m to schedule there. """ pod = _agent_deployment()["spec"]["template"]["spec"] hostnames = next( item for item in pod["affinity"]["nodeAffinity"][ "requiredDuringSchedulingIgnoredDuringExecution" ]["nodeSelectorTerms"][0]["matchExpressions"] if item["key"] == "kubernetes.io/hostname" ) assert hostnames["operator"] == "NotIn" assert set(hostnames["values"]) >= {"titan-04", "titan-19"} hermes = next(item for item in pod["containers"] if item["name"] == "hermes") assert hermes["resources"]["requests"]["cpu"] == "300m" def test_agent_root_is_stock_dashboard_and_terminal_is_a_separate_path(): deployment = _agent_deployment() pod = deployment["spec"]["template"]["spec"] containers = {item["name"]: item for item in pod["containers"]} assert "webui" not in containers assert "dashboard" not in containers hermes = containers["hermes"] hermes_env = {item["name"]: item["value"] for item in hermes["env"]} assert hermes_env["HERMES_STREAM_STALE_TIMEOUT"] == "600" assert hermes_env["HERMES_API_CALL_STALE_TIMEOUT"] == "600" assert hermes["command"] == ["/bin/sh", "-ec"] startup = hermes["args"][0] assert ". /opt/data/.env" in startup assert "exec /init /opt/hermes/docker/main-wrapper.sh gateway run" in startup hermes_env = {item["name"]: item["value"] for item in hermes["env"]} assert hermes_env["HERMES_DASHBOARD"] == "1" assert hermes_env["HERMES_DASHBOARD_HOST"] == "127.0.0.1" assert hermes_env["HERMES_DASHBOARD_PORT"] == "9119" assert hermes_env["HERMES_TUI_AGENT_INIT_TIMEOUT_S"] == "180" assert hermes["securityContext"]["runAsUser"] == 0 assert hermes["securityContext"]["runAsGroup"] == 0 for probe_name in ("startupProbe", "readinessProbe", "livenessProbe"): probe = hermes[probe_name] assert probe["exec"]["command"] == [ "curl", "-fsS", "http://127.0.0.1:9119/api/status", ] terminal = containers["terminal"] command = terminal["args"][0] assert "--base-path /terminal" in command assert "--check-origin" not in command assert "/usr/bin/tmux new-session -A" in command assert "--continue" in command assert "--yolo" in command terminal_env = {item["name"]: item["value"] for item in terminal["env"]} assert terminal_env["HERMES_TUI_AGENT_INIT_TIMEOUT_S"] == "180" claude_broker = containers["claude-broker"] claude_env = {item["name"]: item["value"] for item in claude_broker["env"]} assert claude_env["HERMES_CLAUDE_BROKER_CONCURRENCY"] == "2" assert claude_broker["readinessProbe"]["tcpSocket"] == {"port": "claude-broker"} assert claude_broker["livenessProbe"]["tcpSocket"] == {"port": "claude-broker"} args = containers["oauth2-proxy"]["args"] terminal_upstream = "--upstream=http://127.0.0.1:7681/terminal/" dashboard_upstream = "--upstream=http://127.0.0.1:9119/" assert terminal_upstream in args assert dashboard_upstream in args assert args.index(terminal_upstream) < args.index(dashboard_upstream) assert "--pass-host-header=false" in args assert "--cookie-refresh=19m" in args assert "--session-store-type=redis" in args assert any( arg.startswith("--redis-connection-url=redis://hermes-oauth-sessions.") for arg in args ) patch_init = next( item for item in pod["initContainers"] if item["name"] == "patch-tui-gateway" ) assert patch_init["command"][-1] == "/patched/server.py" for name in ("hermes", "terminal"): mounts = containers[name]["volumeMounts"] assert { "name": "tui-gateway-patch", "mountPath": "/opt/hermes/tui_gateway/server.py", "subPath": "server.py", } in mounts ingress_documents = [ item for item in yaml.safe_load_all((HERMES / "agent-ingress.yaml").read_text()) if item ] middlewares = { item["metadata"]["name"]: item for item in ingress_documents if item["kind"] == "Middleware" } assert middlewares["hermes-agent-terminal-slash"]["spec"]["redirectRegex"][ "replacement" ].endswith("/terminal/") assert ( middlewares["hermes-agent-stock-dashboard-headers"]["spec"]["headers"][ "customRequestHeaders" ]["Origin"] == "http://127.0.0.1:9119" ) ingresses = { item["metadata"]["name"]: item for item in ingress_documents if item["kind"] == "Ingress" } assert ( ingresses["hermes-agent-dashboard"]["metadata"]["annotations"][ "traefik.ingress.kubernetes.io/router.middlewares" ] == "hermes-hermes-agent-stock-dashboard-headers@kubernetescrd" ) assert ( ingresses["hermes-agent-terminal"]["metadata"]["annotations"][ "traefik.ingress.kubernetes.io/router.middlewares" ] == "hermes-hermes-agent-terminal-slash@kubernetescrd" )