526 lines
19 KiB
Python
526 lines
19 KiB
Python
"""Focused tests for Hermes-to-Herdr routing and the shared auth patch."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
|
|
SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
|
|
HERMES = Path(__file__).parents[2] / "services/hermes"
|
|
|
|
|
|
def _load(name: str):
|
|
spec = importlib.util.spec_from_file_location(name, SCRIPTS / f"{name}.py")
|
|
assert spec and spec.loader
|
|
module = importlib.util.module_from_spec(spec)
|
|
sys.modules[spec.name] = module
|
|
spec.loader.exec_module(module)
|
|
return module
|
|
|
|
|
|
dispatch = _load("herdr_dispatch")
|
|
tab_router = _load("herdr_tab_router")
|
|
auth_patch = _load("patch_hermes_auth")
|
|
ttyd_patch = _load("patch_ttyd_index")
|
|
|
|
|
|
def test_herdr_plan_chooses_task_shape_and_caps_effort():
|
|
status = {
|
|
"routes": {
|
|
"codex-high": [
|
|
"openai-codex/gpt-5.6-sol",
|
|
"anthropic/claude-opus-5",
|
|
"custom/qwen2.5:14b-instruct-q4_0",
|
|
],
|
|
"claude-medium": [
|
|
"anthropic/claude-sonnet-5",
|
|
"openai-codex/gpt-5.6-terra",
|
|
],
|
|
}
|
|
}
|
|
implementation = dispatch.select_plan(status, "implementation", "high")
|
|
architecture = dispatch.select_plan(status, "architecture", "medium")
|
|
assert implementation["worker"] == "codex"
|
|
assert implementation["model"] == "gpt-5.6-sol"
|
|
assert architecture["worker"] == "claude"
|
|
assert architecture["model"] == "claude-sonnet-5"
|
|
with pytest.raises(ValueError, match="effort"):
|
|
dispatch.select_plan(status, "review", "max")
|
|
|
|
|
|
def test_claude_worker_waits_for_prompt_readiness(tmp_path: Path, monkeypatch):
|
|
herdr = tmp_path / "herdr"
|
|
herdr.touch()
|
|
project = tmp_path / "project"
|
|
project.mkdir()
|
|
calls = []
|
|
|
|
def fake_run(command, env):
|
|
calls.append(command)
|
|
if command[1:3] == ["workspace", "create"]:
|
|
return {"result": {"root_pane": {"pane_id": "w2:p1"}}}
|
|
return {"result": {"ok": True}}
|
|
|
|
monkeypatch.setattr(dispatch, "HERDR_BIN", herdr)
|
|
monkeypatch.setattr(dispatch, "_run", fake_run)
|
|
plan = {
|
|
"worker": "claude",
|
|
"model": "claude-haiku-4-5-20251001",
|
|
"effort": "low",
|
|
}
|
|
|
|
dispatch.launch_worker(plan, project, "review", "Check the implementation.")
|
|
|
|
ready = calls[-2]
|
|
assert ready[1:] == [
|
|
"pane",
|
|
"wait-output",
|
|
"w2:p1",
|
|
"--match",
|
|
"Claude Code",
|
|
"--source",
|
|
"recent",
|
|
"--lines",
|
|
"120",
|
|
"--timeout",
|
|
"120000",
|
|
]
|
|
prompt = calls[-1]
|
|
assert prompt[1:5] == [
|
|
"agent",
|
|
"prompt",
|
|
"project-claude-review",
|
|
"Check the implementation.",
|
|
]
|
|
assert prompt[-9:] == [
|
|
"--wait",
|
|
"--until",
|
|
"working",
|
|
"--until",
|
|
"done",
|
|
"--until",
|
|
"blocked",
|
|
"--timeout",
|
|
"15000",
|
|
]
|
|
|
|
workspace_create = calls[0]
|
|
assert workspace_create[-3:] == ["--label", "project-claude-review", "--no-focus"]
|
|
start = calls[1]
|
|
assert start[1:4] == ["agent", "start", "project-claude-review"]
|
|
assert start[-2:] == ["--permission-mode", "auto"]
|
|
|
|
|
|
def test_codex_worker_uses_automatic_review_without_prompts(tmp_path: Path, monkeypatch):
|
|
herdr = tmp_path / "herdr"
|
|
herdr.touch()
|
|
project = tmp_path / "cassandra"
|
|
project.mkdir()
|
|
auth = tmp_path / "auth.json"
|
|
auth.write_text("{}", encoding="utf-8")
|
|
calls = []
|
|
|
|
def fake_run(command, env):
|
|
calls.append(command)
|
|
if command[1:3] == ["workspace", "create"]:
|
|
return {"result": {"root_pane": {"pane_id": "w3:p1"}}}
|
|
return {"result": {"ok": True}}
|
|
|
|
monkeypatch.setattr(dispatch, "HERDR_BIN", herdr)
|
|
monkeypatch.setattr(dispatch, "CODEX_AUTH", auth)
|
|
monkeypatch.setattr(dispatch, "_run", fake_run)
|
|
plan = {"worker": "codex", "model": "gpt-5.6-sol", "effort": "high"}
|
|
|
|
dispatch.launch_worker(plan, project, "api-fix", None)
|
|
|
|
assert calls[0][-3:] == ["--label", "cassandra-codex-api-fix", "--no-focus"]
|
|
start = calls[1]
|
|
assert start[1:4] == ["agent", "start", "cassandra-codex-api-fix"]
|
|
assert "--approve-for-me" in start
|
|
assert start[-2:] == ["--sandbox", "workspace-write"]
|
|
assert not any("on-request" in value for value in start)
|
|
|
|
|
|
def test_agent_config_uses_bounded_noninteractive_approvals():
|
|
configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
|
|
config = yaml.safe_load(configmap["data"]["config.yaml"])
|
|
|
|
assert config["approvals"]["mode"] == "off"
|
|
denied = config["approvals"]["deny"]
|
|
assert "*kubectl apply*" in denied
|
|
assert "*git push --force*" in denied
|
|
assert "*git reset --hard*" in denied
|
|
|
|
|
|
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\n"
|
|
"def _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(encoding="utf-8")
|
|
assert 'os.environ.get("HERMES_AUTH_FILE"' in content
|
|
assert "Path(configured)" 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_ttyd_clipboard_patch_uses_system_clipboard_and_preserves_interrupt():
|
|
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 "term.hasSelection()" in content
|
|
assert "event.stopImmediatePropagation()" in content
|
|
assert 'document.execCommand("copy")' not in content
|
|
assert "document.execCommand('copy')" in content
|
|
|
|
|
|
def test_ttyd_client_patch_recovers_with_a_fresh_authenticated_page():
|
|
source = (
|
|
'<html><body><script>document.execCommand("copy")</script></body></html>'
|
|
)
|
|
content = ttyd_patch.patch_html(source)
|
|
|
|
assert "class AtlasRecoveringWebSocket" in content
|
|
assert "new URL('token', window.location.href)" in content
|
|
assert "credentials: 'same-origin'" in content
|
|
assert "response.type === 'opaqueredirect'" in content
|
|
assert "typeof payload.token === 'string'" in content
|
|
assert "window.location.reload()" in content
|
|
assert "Reconnecting terminal automatically..." in content
|
|
assert "if (!recoveryStarted || event.key !== 'Enter') return" in content
|
|
assert content.index('id="atlas-ttyd-clipboard"') < content.index(
|
|
'<script>void 0</script>'
|
|
)
|
|
|
|
|
|
def test_ttyd_clipboard_patch_fails_closed_on_upstream_drift():
|
|
with pytest.raises(RuntimeError, match="context changed"):
|
|
ttyd_patch.patch_html("<html><body>changed</body></html>")
|
|
|
|
|
|
def test_agent_tab_router_starts_hermes_in_matching_project(tmp_path: Path):
|
|
projects = tmp_path / "projects"
|
|
cassandra = projects / "cassandra"
|
|
cassandra.mkdir(parents=True)
|
|
calls = []
|
|
|
|
def fake_run(command):
|
|
calls.append(command)
|
|
if command[1:3] == ["tab", "list"]:
|
|
return {
|
|
"result": {
|
|
"tabs": [
|
|
{
|
|
"tab_id": "w2:t2",
|
|
"workspace_id": "w2",
|
|
"label": "Cassandra",
|
|
}
|
|
]
|
|
}
|
|
}
|
|
if command[1:3] == ["pane", "list"]:
|
|
return {
|
|
"result": {
|
|
"panes": [
|
|
{
|
|
"pane_id": "w2:p1",
|
|
"tab_id": "w2:t1",
|
|
"workspace_id": "w2",
|
|
"agent": "hermes",
|
|
},
|
|
{
|
|
"pane_id": "w2:p2",
|
|
"tab_id": "w2:t2",
|
|
"workspace_id": "w2",
|
|
"foreground_cwd": str(tmp_path),
|
|
},
|
|
]
|
|
}
|
|
}
|
|
return {"result": {"ok": True}}
|
|
|
|
started = tab_router.route_unmanaged_tabs("w2", fake_run, projects)
|
|
|
|
assert started == ["w2:p2"]
|
|
assert calls[-2][1:4] == ["pane", "run", "w2:p2"]
|
|
assert calls[-2][-1] == f"cd {cassandra.resolve()}"
|
|
assert calls[-1][1:] == [
|
|
"agent",
|
|
"start",
|
|
"cassandra-p2",
|
|
"--kind",
|
|
"hermes",
|
|
"--pane",
|
|
"w2:p2",
|
|
"--timeout",
|
|
"120000",
|
|
]
|
|
|
|
|
|
def test_agent_tab_router_only_targets_coordinator_workspace():
|
|
calls = []
|
|
|
|
def fake_run(command):
|
|
calls.append(command)
|
|
return {
|
|
"result": {
|
|
"workspaces": [
|
|
{"workspace_id": "w2", "label": "coordinator"},
|
|
{"workspace_id": "w3", "label": "codex-worker"},
|
|
]
|
|
}
|
|
}
|
|
|
|
assert tab_router.find_workspace("coordinator", fake_run) == "w2"
|
|
assert calls == [[str(tab_router.HERDR_BIN), "workspace", "list"]]
|
|
|
|
|
|
def test_agent_ttyd_defers_identity_to_owner_only_oauth_boundary():
|
|
deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
|
|
containers = deployment["spec"]["template"]["spec"]["containers"]
|
|
ttyd = next(container for container in containers if container["name"] == "herdr-tui")
|
|
command = ttyd["args"][0]
|
|
|
|
assert "--check-origin" in command
|
|
assert "--auth-header" not in command
|
|
assert "--index /ttyd-index/index.html" in command
|
|
assert "--client-option disableReconnect=true" in command
|
|
|
|
init_names = {
|
|
container["name"]
|
|
for container in deployment["spec"]["template"]["spec"]["initContainers"]
|
|
}
|
|
assert "prepare-ttyd-index" in init_names
|
|
|
|
oauth_documents = [
|
|
document
|
|
for document in yaml.safe_load_all((HERMES / "oauth2-proxy.yaml").read_text())
|
|
if document
|
|
]
|
|
oauth = next(
|
|
document
|
|
for document in oauth_documents
|
|
if document["kind"] == "Deployment"
|
|
and document["metadata"]["name"] == "oauth2-proxy-hermes-agent"
|
|
)
|
|
oauth_args = oauth["spec"]["template"]["spec"]["containers"][0]["args"]
|
|
assert "--authenticated-emails-file=/etc/oauth2-proxy/allowed-emails" in oauth_args
|
|
assert "--proxy-websockets=true" in oauth_args
|
|
|
|
network_documents = [
|
|
document
|
|
for document in yaml.safe_load_all((HERMES / "networkpolicy.yaml").read_text())
|
|
if document
|
|
]
|
|
isolation = next(
|
|
document
|
|
for document in network_documents
|
|
if document["kind"] == "NetworkPolicy"
|
|
and document["metadata"]["name"] == "hermes-agent-isolation"
|
|
)
|
|
ingress = isolation["spec"]["ingress"]
|
|
assert ingress == [
|
|
{
|
|
"from": [{"podSelector": {"matchLabels": {"app": "oauth2-proxy-hermes-agent"}}}],
|
|
"ports": [{"protocol": "TCP", "port": 7681}],
|
|
}
|
|
]
|
|
|
|
|
|
def test_hermes_oauth_boundaries_use_current_hardened_proxy():
|
|
documents = [
|
|
document
|
|
for document in yaml.safe_load_all((HERMES / "oauth2-proxy.yaml").read_text())
|
|
if document
|
|
]
|
|
deployments = [
|
|
document
|
|
for document in documents
|
|
if document["kind"] == "Deployment"
|
|
and document["metadata"]["name"].startswith("oauth2-proxy-hermes-")
|
|
]
|
|
|
|
assert len(deployments) == 3
|
|
for deployment in deployments:
|
|
container = deployment["spec"]["template"]["spec"]["containers"][0]
|
|
args = container["args"]
|
|
assert "v7.15.3@sha256:10a1165743a192e" in container["image"]
|
|
assert "--cookie-csrf-expire=10m" in args
|
|
assert "--cookie-csrf-per-request=true" in args
|
|
assert "--cookie-csrf-per-request-limit=8" in args
|
|
assert "--trusted-proxy-ip=10.42.0.0/16" in args
|
|
|
|
|
|
def test_agent_installs_hermes_integration_before_startup():
|
|
deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
|
|
pod = deployment["spec"]["template"]["spec"]
|
|
init_config = next(
|
|
container for container in pod["initContainers"] if container["name"] == "init-config"
|
|
)
|
|
assert "ln -s /opt/data /opt/data/home/.hermes" in init_config["command"][-1]
|
|
|
|
installer = next(
|
|
container
|
|
for container in pod["initContainers"]
|
|
if container["name"] == "install-herdr-integrations"
|
|
)
|
|
command = installer["command"][-1]
|
|
assert "herdr integration install codex" in command
|
|
assert "herdr integration install claude" in command
|
|
assert "herdr integration install hermes" in command
|
|
assert "|| true" not in command
|
|
|
|
containers = {container["name"]: container for container in pod["containers"]}
|
|
for name in ("herdr-tui", "herdr-server"):
|
|
env = {item["name"]: item["value"] for item in containers[name]["env"]}
|
|
assert "/opt/hermes/.venv/bin" in env["PATH"].split(":")
|
|
|
|
server_env = {
|
|
item["name"]: item["value"] for item in containers["herdr-server"]["env"]
|
|
}
|
|
assert "/opt/data/home/.local/bin" in server_env["PATH"].split(":")
|
|
|
|
for name in ("hermes", "herdr-server"):
|
|
env = {item["name"]: item["value"] for item in containers[name]["env"]}
|
|
assert env["AGENT_BROWSER_EXECUTABLE_PATH"].endswith("/chrome-linux/headless_shell")
|
|
assert "--no-sandbox" in env["AGENT_BROWSER_ARGS"]
|
|
|
|
server_command = containers["herdr-server"]["command"][-1]
|
|
assert 'herdr pane process-info --pane "${pane}"' in server_command
|
|
assert 'herdr workspace close "${workspace}"' in server_command
|
|
assert 'item.get("label") == "coordinator"' in server_command
|
|
assert 'item["workspace_id"] != active' in server_command
|
|
assert 'herdr workspace close "${stale_workspace}"' in server_command
|
|
assert "herdr agent start coordinator" in server_command
|
|
assert "/opt/coordinator/herdr_tab_router.py" in server_command
|
|
assert "--workspace-label coordinator" in server_command
|
|
assert "--kind hermes" in server_command
|
|
assert "--timeout 60000" in server_command
|
|
assert "--env AGENT_BROWSER_EXECUTABLE_PATH=" in server_command
|
|
|
|
tui_command = containers["herdr-tui"]["args"][0]
|
|
assert "while true" in tui_command
|
|
assert "Herdr disconnected" in tui_command
|
|
assert "/opt/data/tools/bin/herdr" in tui_command
|
|
|
|
|
|
def test_agent_mounts_auto_router_into_both_hermes_runtimes():
|
|
deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
|
|
pod = deployment["spec"]["template"]["spec"]
|
|
containers = {container["name"]: container for container in pod["containers"]}
|
|
|
|
for name in ("hermes", "herdr-server"):
|
|
mounts = {
|
|
mount["mountPath"]: mount["name"]
|
|
for mount in containers[name]["volumeMounts"]
|
|
}
|
|
assert mounts["/opt/data/plugins/auto-router"] == "auto-router-plugin"
|
|
|
|
volume = next(
|
|
item for item in pod["volumes"] if item["name"] == "auto-router-plugin"
|
|
)
|
|
assert volume["configMap"]["name"] == "hermes-auto-router-plugin"
|
|
|
|
|
|
def test_agent_has_cassandra_readonly_kubernetes_context():
|
|
deployment = yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
|
|
pod = deployment["spec"]["template"]["spec"]
|
|
containers = {container["name"]: container for container in pod["containers"]}
|
|
|
|
installer = next(
|
|
container
|
|
for container in pod["initContainers"]
|
|
if container["name"] == "install-agent-tools"
|
|
)
|
|
install_command = installer["command"][-1]
|
|
assert "v1.33.3/bin/linux/arm64/kubectl" in install_command
|
|
assert "3d514dbae5dc8c09f773df0ef0f5d449" in install_command
|
|
|
|
for name in ("hermes", "herdr-server"):
|
|
env = {item["name"]: item["value"] for item in containers[name]["env"]}
|
|
assert env["KUBECONFIG"] == "/opt/data/home/.kube/config"
|
|
mounts = {
|
|
mount["mountPath"]: mount for mount in containers[name]["volumeMounts"]
|
|
}
|
|
assert mounts["/opt/data/home/.kube/config"]["name"] == "kubeconfig"
|
|
assert mounts["/opt/data/home/.kube/config"]["readOnly"] is True
|
|
|
|
config = yaml.safe_load((HERMES / "agent-kubeconfig.yaml").read_text())
|
|
assert config["current-context"] == "cassandra-readonly"
|
|
assert config["contexts"][0]["context"]["namespace"] == "cassandra"
|
|
assert config["users"][0]["user"] == {
|
|
"tokenFile": "/var/run/secrets/kubernetes.io/serviceaccount/token"
|
|
}
|
|
|
|
rbac_path = HERMES.parent / "cassandra" / "hermes-agent-rbac.yaml"
|
|
documents = [document for document in yaml.safe_load_all(rbac_path.read_text()) if document]
|
|
role = next(document for document in documents if document["kind"] == "Role")
|
|
binding = next(document for document in documents if document["kind"] == "RoleBinding")
|
|
assert role["metadata"]["namespace"] == "cassandra"
|
|
assert binding["metadata"]["namespace"] == "cassandra"
|
|
assert binding["subjects"] == [
|
|
{"kind": "ServiceAccount", "name": "hermes-agent", "namespace": "hermes"}
|
|
]
|
|
assert "secrets" not in {
|
|
resource for rule in role["rules"] for resource in rule["resources"]
|
|
}
|
|
assert {
|
|
verb for rule in role["rules"] for verb in rule["verbs"]
|
|
} == {"get", "list", "watch"}
|
|
|
|
|
|
def test_agent_reconnect_renders_complete_conversation_history():
|
|
documents = [
|
|
document
|
|
for document in yaml.safe_load_all((HERMES / "agent-configmap.yaml").read_text())
|
|
if document
|
|
]
|
|
config_map = next(
|
|
document
|
|
for document in documents
|
|
if document["kind"] == "ConfigMap"
|
|
and document["metadata"]["name"] == "hermes-agent-config"
|
|
)
|
|
config = yaml.safe_load(config_map["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 display["resume_max_assistant_lines"] >= 1000000
|
|
|
|
|
|
def test_agent_coordinator_has_a_long_running_tool_budget():
|
|
configmap = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())
|
|
config = yaml.safe_load(configmap["data"]["config.yaml"])
|
|
|
|
assert config["agent"]["max_turns"] == 180
|
|
assert config["tool_loop_guardrails"]["hard_stop_enabled"] is True
|
|
for platform in ("cli", "api_server"):
|
|
tools = config["platform_toolsets"][platform]
|
|
assert "browser" in tools
|
|
assert "delegation" in tools
|
|
assert "memory" in tools
|
|
assert "terminal" in tools
|
|
assert "vision" in tools
|