1905 lines
67 KiB
Python
1905 lines
67 KiB
Python
"""Focused tests for Agent Hermes' direct Codex and Claude Kanban lanes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import importlib.util
|
|
import json
|
|
import os
|
|
import signal
|
|
import sys
|
|
from contextlib import nullcontext
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
import yaml
|
|
|
|
|
|
SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
HERMES = Path(__file__).parents[2] / "services/hermes"
|
|
KEYCLOAK = Path(__file__).parents[2] / "services/keycloak"
|
|
FLUX_HERMES = (
|
|
Path(__file__).parents[2]
|
|
/ "clusters/atlas/flux-system/applications/hermes/kustomization.yaml"
|
|
)
|
|
|
|
|
|
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
|
|
|
|
|
|
lanes = _load("cli_lane_runner")
|
|
policy = _load("claude_command_policy")
|
|
migration = _load("migrate_herdr_state")
|
|
auth_patch = _load("patch_hermes_auth")
|
|
tui_gateway_patch = _load("patch_tui_gateway")
|
|
codex_runtime_patch = _load("patch_codex_runtime")
|
|
ttyd_patch = _load("patch_ttyd_index")
|
|
client_config = _load("configure_agent_clients")
|
|
|
|
|
|
def _agent_deployment() -> dict:
|
|
return yaml.safe_load((HERMES / "agent-deployment.yaml").read_text())
|
|
|
|
|
|
def _services() -> dict[str, dict]:
|
|
return {
|
|
item["metadata"]["name"]: item
|
|
for item in yaml.safe_load_all((HERMES / "service.yaml").read_text())
|
|
if item
|
|
}
|
|
|
|
|
|
def _oauth_deployment(name: str) -> dict:
|
|
documents = [
|
|
item
|
|
for item in yaml.safe_load_all((HERMES / "oauth2-proxy.yaml").read_text())
|
|
if item
|
|
]
|
|
return next(
|
|
item
|
|
for item in documents
|
|
if item["kind"] == "Deployment" and item["metadata"]["name"] == name
|
|
)
|
|
|
|
|
|
class _SwitchyardResponse:
|
|
"""Minimal context-managed response used by routing contract tests."""
|
|
|
|
def __init__(self, selected: str, rationale: str = "local classifier vote"):
|
|
self.headers = {
|
|
"x-model-router-selected-model": selected,
|
|
"x-model-router-rationale": rationale,
|
|
}
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self):
|
|
return b"{}"
|
|
|
|
|
|
def test_auto_lane_uses_switchyard_worker_decision():
|
|
observed = {}
|
|
|
|
def route(request, timeout):
|
|
observed["payload"] = json.loads(request.data)
|
|
observed["timeout"] = timeout
|
|
return _SwitchyardResponse("worker/claude/claude-opus-5/xhigh")
|
|
|
|
route = lanes.select_route(
|
|
"Perform the security review.",
|
|
"cli-auto",
|
|
open_request=route,
|
|
)
|
|
|
|
assert route.provider == "claude"
|
|
assert route.model == "claude-opus-5"
|
|
assert route.effort == "xhigh"
|
|
assert route.classifier == "switchyard-classifier"
|
|
assert observed["payload"]["model"] == "atlas/worker/auto/maximum"
|
|
assert observed["timeout"] == 60
|
|
|
|
|
|
def test_manual_lane_is_still_enforced_by_switchyard():
|
|
observed = {}
|
|
|
|
def route(request, timeout):
|
|
observed["payload"] = json.loads(request.data)
|
|
return _SwitchyardResponse(
|
|
"worker/claude/claude-sonnet-5/high", "manual route"
|
|
)
|
|
|
|
route = lanes.select_route(
|
|
"Implement it.",
|
|
"cli-claude-high",
|
|
open_request=route,
|
|
)
|
|
|
|
assert route.provider == "claude"
|
|
assert route.effort == "high"
|
|
assert route.classifier == "switchyard-manual"
|
|
assert route.reason == "manual route"
|
|
assert observed["payload"]["model"] == "atlas/worker/manual/claude/high"
|
|
|
|
|
|
def test_cross_provider_retry_passes_failed_provider_to_switchyard():
|
|
observed = {}
|
|
|
|
def route(request, timeout):
|
|
observed["payload"] = json.loads(request.data)
|
|
return _SwitchyardResponse("worker/claude/claude-sonnet-5/high")
|
|
|
|
route = lanes.select_route(
|
|
"Retry after capacity exhaustion.",
|
|
"cli-auto",
|
|
exclude_provider="codex",
|
|
open_request=route,
|
|
)
|
|
|
|
assert route.provider == "claude"
|
|
assert route.classifier == "switchyard-classifier"
|
|
content = observed["payload"]["messages"][0]["content"]
|
|
assert "codex provider failed or exhausted capacity" in content
|
|
|
|
|
|
def test_classifier_cannot_select_a_freshly_excluded_provider():
|
|
payloads = []
|
|
|
|
def route(request, timeout):
|
|
payload = json.loads(request.data)
|
|
payloads.append(payload)
|
|
if len(payloads) == 1:
|
|
return _SwitchyardResponse("worker/claude/opus/xhigh")
|
|
return _SwitchyardResponse("worker/codex/sol/xhigh", "healthy route")
|
|
|
|
selected = lanes.select_route(
|
|
"Perform a consequential review.",
|
|
"cli-auto",
|
|
exclude_provider="claude",
|
|
exclude_reason="is unavailable according to fresh native health",
|
|
open_request=route,
|
|
)
|
|
|
|
assert selected.provider == "codex"
|
|
assert selected.effort == "xhigh"
|
|
assert selected.classifier == "switchyard-classifier-health-guard"
|
|
assert payloads[0]["model"] == "atlas/worker/auto/maximum"
|
|
assert payloads[1]["model"] == "atlas/worker/manual/codex/xhigh"
|
|
assert "claude provider is unavailable" in payloads[0]["messages"][0]["content"]
|
|
|
|
|
|
def test_fresh_native_health_excludes_only_one_proven_down_provider(
|
|
tmp_path: Path, monkeypatch
|
|
):
|
|
paths = {
|
|
"codex": tmp_path / "codex.json",
|
|
"claude": tmp_path / "claude.json",
|
|
}
|
|
paths["codex"].write_text('{"state":"available"}\n', encoding="utf-8")
|
|
paths["claude"].write_text('{"state":"unavailable"}\n', encoding="utf-8")
|
|
monkeypatch.setattr(lanes, "PROVIDER_HEALTH_PATHS", paths)
|
|
|
|
now = max(path.stat().st_mtime for path in paths.values())
|
|
assert lanes.fresh_unavailable_provider(now=now) == "claude"
|
|
|
|
paths["codex"].write_text('{"state":"unavailable"}\n', encoding="utf-8")
|
|
now = max(path.stat().st_mtime for path in paths.values())
|
|
assert lanes.fresh_unavailable_provider(now=now) is None
|
|
|
|
|
|
def test_worker_environment_preserves_vault_backed_cli_homes(monkeypatch):
|
|
"""Kanban workers must not fall back to credential-free persistent homes."""
|
|
monkeypatch.setenv("CODEX_HOME", "/runtime-access/codex")
|
|
monkeypatch.setenv("CLAUDE_CONFIG_DIR", "/runtime-access/claude")
|
|
|
|
env = lanes._base_env()
|
|
|
|
assert env["HOME"] == str(lanes.DATA_ROOT / "home")
|
|
assert env["CODEX_HOME"] == "/runtime-access/codex"
|
|
assert env["CLAUDE_CONFIG_DIR"] == "/runtime-access/claude"
|
|
assert env["GIT_TERMINAL_PROMPT"] == "0"
|
|
|
|
|
|
def test_claude_session_is_reserved_before_first_process(tmp_path: Path, monkeypatch):
|
|
state: dict = {}
|
|
state_file = tmp_path / "state.json"
|
|
observed = {}
|
|
|
|
def fake_stream(command, **kwargs):
|
|
observed["command"] = command
|
|
observed["state"] = json.loads(state_file.read_text())
|
|
return lanes.ProcessResult(0, "", {"status": "completed"}, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", fake_stream)
|
|
route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
|
|
|
|
lanes.run_provider(
|
|
route,
|
|
"Do the work.",
|
|
tmp_path,
|
|
state,
|
|
state_file,
|
|
tmp_path / "worker.log",
|
|
lambda _note: True,
|
|
60,
|
|
)
|
|
|
|
reserved = observed["state"]["claude_session_id"]
|
|
assert reserved
|
|
assert ["--session-id", reserved] == observed["command"][-4:-2]
|
|
|
|
|
|
def test_claude_recovery_uses_verified_error_state_machine(tmp_path: Path, monkeypatch):
|
|
state = {"claude_session_id": "d4f91f59-88cb-43d0-a64f-f3284688fe9e", "claude_started": True}
|
|
state_file = tmp_path / "state.json"
|
|
calls = []
|
|
|
|
def fake_stream(command, **kwargs):
|
|
calls.append(command)
|
|
if len(calls) == 1:
|
|
return lanes.ProcessResult(
|
|
1,
|
|
"No conversation found with session ID: d4f91f59-88cb-43d0-a64f-f3284688fe9e",
|
|
None,
|
|
False,
|
|
)
|
|
return lanes.ProcessResult(0, "", {"status": "completed"}, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", fake_stream)
|
|
route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
|
|
lanes.run_provider(route, "Continue.", tmp_path, state, state_file, tmp_path / "log", lambda _: True, 60)
|
|
|
|
assert "--resume" in calls[0]
|
|
assert "--session-id" in calls[1]
|
|
assert calls[0][calls[0].index("--resume") + 1] == calls[1][calls[1].index("--session-id") + 1]
|
|
|
|
|
|
def test_claude_collision_never_double_starts_a_reserved_session(tmp_path: Path, monkeypatch):
|
|
state = {"claude_session_id": "24927cee-f3da-4bcf-b624-602c22860155"}
|
|
state_file = tmp_path / "state.json"
|
|
calls = []
|
|
|
|
def fake_stream(command, **kwargs):
|
|
calls.append(command)
|
|
if len(calls) == 1:
|
|
return lanes.ProcessResult(1, "Session ID already in use", None, False)
|
|
return lanes.ProcessResult(0, "", {"status": "completed"}, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", fake_stream)
|
|
route = lanes.Route("claude", "claude-opus-5", "high", "claude-high", "jetson", "vote", 1, ())
|
|
lanes.run_provider(route, "Continue.", tmp_path, state, state_file, tmp_path / "log", lambda _: True, 60)
|
|
|
|
assert "--session-id" in calls[0]
|
|
assert "--resume" in calls[1]
|
|
assert calls[0][calls[0].index("--session-id") + 1] == calls[1][calls[1].index("--resume") + 1]
|
|
|
|
|
|
def test_codex_thread_started_is_persisted_before_completion(tmp_path: Path):
|
|
state: dict = {}
|
|
state_file = tmp_path / "state.json"
|
|
|
|
lanes._event_payload(
|
|
"codex",
|
|
json.dumps({"type": "thread.started", "thread_id": "thread-123"}),
|
|
state,
|
|
state_file,
|
|
)
|
|
|
|
assert json.loads(state_file.read_text())["codex_thread_id"] == "thread-123"
|
|
|
|
|
|
def test_codex_missing_server_thread_restarts_fresh(tmp_path: Path, monkeypatch):
|
|
state = {"codex_thread_id": "failed-thread"}
|
|
state_file = tmp_path / "state.json"
|
|
calls = []
|
|
|
|
def fake_stream(command, **kwargs):
|
|
calls.append(command)
|
|
if len(calls) == 1:
|
|
return lanes.ProcessResult(
|
|
1,
|
|
"thread/resume failed: no rollout found for thread id failed-thread",
|
|
None,
|
|
False,
|
|
)
|
|
return lanes.ProcessResult(0, "", {"status": "completed"}, False)
|
|
|
|
monkeypatch.setattr(lanes, "stream_process", fake_stream)
|
|
route = lanes.Route(
|
|
"codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
|
|
)
|
|
|
|
result = lanes.run_provider(
|
|
route,
|
|
"Continue.",
|
|
tmp_path,
|
|
state,
|
|
state_file,
|
|
tmp_path / "log",
|
|
lambda _: True,
|
|
60,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert "resume" in calls[0]
|
|
assert "resume" not in calls[1]
|
|
assert "codex_thread_id" not in json.loads(state_file.read_text())
|
|
|
|
|
|
def test_successful_process_text_cannot_masquerade_as_capacity_failure(tmp_path: Path):
|
|
result = lanes.stream_process(
|
|
[sys.executable, "-c", "print('authentication work completed')"],
|
|
provider="codex",
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
log_path=tmp_path / "worker.log",
|
|
state={},
|
|
state_file=tmp_path / "state.json",
|
|
heartbeat=lambda _note: True,
|
|
max_runtime=60,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert result.capacity_failure is False
|
|
|
|
|
|
def test_worker_process_uses_isolated_process_session(monkeypatch, tmp_path: Path):
|
|
original_popen = lanes.subprocess.Popen
|
|
calls = []
|
|
|
|
def recording_popen(*args, **kwargs):
|
|
calls.append(kwargs)
|
|
return original_popen(*args, **kwargs)
|
|
|
|
monkeypatch.setattr(lanes.subprocess, "Popen", recording_popen)
|
|
result = lanes.stream_process(
|
|
[sys.executable, "-c", "print('done')"],
|
|
provider="codex",
|
|
cwd=tmp_path,
|
|
env=dict(os.environ),
|
|
log_path=tmp_path / "worker.log",
|
|
state={},
|
|
state_file=tmp_path / "state.json",
|
|
heartbeat=lambda _note: True,
|
|
max_runtime=60,
|
|
)
|
|
|
|
assert result.returncode == 0
|
|
assert calls[0]["start_new_session"] is True
|
|
|
|
|
|
def test_worker_process_group_is_killed_after_leader_exits(monkeypatch):
|
|
class FinishedProcess:
|
|
pid = 4321
|
|
|
|
@staticmethod
|
|
def poll():
|
|
return 0
|
|
|
|
signals = []
|
|
monkeypatch.setattr(lanes.os, "killpg", lambda pid, sig: signals.append((pid, sig)))
|
|
|
|
lanes._terminate_worker_process(FinishedProcess())
|
|
|
|
assert signals == [(4321, signal.SIGKILL)]
|
|
|
|
|
|
def test_worker_terminal_descendants_in_separate_groups_are_killed(monkeypatch):
|
|
class RunningProcess:
|
|
pid = 4321
|
|
running = True
|
|
|
|
@classmethod
|
|
def poll(cls):
|
|
return None if cls.running else 0
|
|
|
|
@classmethod
|
|
def wait(cls, timeout):
|
|
assert timeout == 10
|
|
cls.running = False
|
|
return 0
|
|
|
|
group_signals = []
|
|
process_signals = []
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"_descendant_processes",
|
|
lambda _pid: {5000: (5000, 12345)},
|
|
)
|
|
monkeypatch.setattr(lanes, "_process_identity_matches", lambda _pid, _start: True)
|
|
monkeypatch.setattr(
|
|
lanes.os,
|
|
"killpg",
|
|
lambda process_group, sig: group_signals.append((process_group, sig)),
|
|
)
|
|
monkeypatch.setattr(
|
|
lanes.os,
|
|
"kill",
|
|
lambda pid, sig: process_signals.append((pid, sig)),
|
|
)
|
|
|
|
lanes._terminate_worker_process(RunningProcess())
|
|
|
|
assert set(group_signals) == {
|
|
(4321, signal.SIGTERM),
|
|
(5000, signal.SIGTERM),
|
|
(4321, signal.SIGKILL),
|
|
(5000, signal.SIGKILL),
|
|
}
|
|
assert process_signals == [(5000, signal.SIGTERM), (5000, signal.SIGKILL)]
|
|
|
|
|
|
def test_unassigned_ready_task_is_persistently_routed_to_auto_lane(monkeypatch):
|
|
task = SimpleNamespace(id="t_auto", assignee=None, status="ready")
|
|
assigned = []
|
|
|
|
class Connection:
|
|
def close(self):
|
|
return None
|
|
|
|
def assign_task(_conn, task_id, profile):
|
|
assigned.append((task_id, profile))
|
|
task.assignee = profile
|
|
return True
|
|
|
|
fake_db = SimpleNamespace(
|
|
list_boards=lambda include_archived=False: [{"slug": "cassandra"}],
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=lambda board: Connection(),
|
|
recompute_ready=lambda _conn: None,
|
|
list_tasks=lambda _conn: [task],
|
|
assign_task=assign_task,
|
|
get_task=lambda _conn, _task_id: task,
|
|
claim_task=lambda _conn, _task_id, **_kwargs: task,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
|
|
|
|
assert lanes.claim_ready(set(), 1) == [("cassandra", "t_auto")]
|
|
assert assigned == [("t_auto", "cli-auto")]
|
|
|
|
|
|
def test_corrupt_board_is_quarantined_without_stopping_healthy_lanes(monkeypatch, capsys):
|
|
class CorruptBoardError(Exception):
|
|
pass
|
|
|
|
task = SimpleNamespace(id="t_healthy", assignee="cli-auto", status="ready")
|
|
|
|
class Connection:
|
|
def close(self):
|
|
return None
|
|
|
|
def connect(*, board):
|
|
if board == "cassandra":
|
|
raise CorruptBoardError("integrity_check failed")
|
|
return Connection()
|
|
|
|
fake_db = SimpleNamespace(
|
|
KanbanDbCorruptError=CorruptBoardError,
|
|
list_boards=lambda include_archived=False: [
|
|
{"slug": "cassandra"},
|
|
{"slug": "healthy"},
|
|
],
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=connect,
|
|
recompute_ready=lambda _conn: None,
|
|
list_tasks=lambda _conn: [task],
|
|
claim_task=lambda _conn, _task_id, **_kwargs: task,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
|
|
lanes.BOARD_CORRUPTION_ERRORS.clear()
|
|
|
|
assert lanes.claim_ready(set(), 1) == [("healthy", "t_healthy")]
|
|
assert "temporarily skipping Kanban board 'cassandra'" in capsys.readouterr().err
|
|
|
|
|
|
def test_transient_board_scan_failure_does_not_stop_healthy_lanes(monkeypatch, capsys):
|
|
task = SimpleNamespace(id="t_healthy", assignee="cli-auto", status="ready")
|
|
|
|
class Connection:
|
|
def __init__(self, board):
|
|
self.board = board
|
|
|
|
def close(self):
|
|
return None
|
|
|
|
def recompute_ready(connection):
|
|
if connection.board == "cassandra":
|
|
raise lanes.sqlite3.OperationalError("disk I/O error")
|
|
|
|
fake_db = SimpleNamespace(
|
|
list_boards=lambda include_archived=False: [
|
|
{"slug": "cassandra"},
|
|
{"slug": "healthy"},
|
|
],
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=lambda board: Connection(board),
|
|
recompute_ready=recompute_ready,
|
|
list_tasks=lambda _conn: [task],
|
|
claim_task=lambda _conn, _task_id, **_kwargs: task,
|
|
)
|
|
monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=fake_db))
|
|
lanes.BOARD_CORRUPTION_ERRORS.clear()
|
|
|
|
assert lanes.claim_ready(set(), 1) == [("healthy", "t_healthy")]
|
|
error = capsys.readouterr().err
|
|
assert "temporarily skipping Kanban board 'cassandra'" in error
|
|
assert "storage OperationalError: disk I/O error" in error
|
|
|
|
|
|
def test_board_call_retries_storage_faults_on_fresh_connections():
|
|
connections = []
|
|
|
|
class Connection:
|
|
def __init__(self):
|
|
self.closed = False
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
def connect(*, board):
|
|
assert board == "cassandra"
|
|
connection = Connection()
|
|
connections.append(connection)
|
|
return connection
|
|
|
|
attempts = []
|
|
|
|
def operation(_connection):
|
|
attempts.append(1)
|
|
if len(attempts) < 3:
|
|
raise lanes.sqlite3.OperationalError("disk I/O error")
|
|
return "healthy"
|
|
|
|
fake_db = SimpleNamespace(
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=connect,
|
|
)
|
|
lanes.BOARD_CORRUPTION_ERRORS.clear()
|
|
|
|
assert lanes._board_call(fake_db, "cassandra", operation) == "healthy"
|
|
assert len(connections) == 3
|
|
assert all(connection.closed for connection in connections)
|
|
|
|
|
|
@pytest.mark.parametrize(
|
|
("result", "expected_action"),
|
|
[
|
|
(lanes.ProcessResult(0, "plain text only", None, False), "block"),
|
|
(
|
|
lanes.ProcessResult(
|
|
0,
|
|
"",
|
|
{
|
|
"status": "completed",
|
|
"summary": "done",
|
|
"changed_files": ["src/a.py"],
|
|
"tests_run": ["pytest -q"],
|
|
"artifacts": ["reports/result.json"],
|
|
"blockers": [],
|
|
},
|
|
False,
|
|
),
|
|
"complete",
|
|
),
|
|
(
|
|
lanes.ProcessResult(
|
|
0,
|
|
"",
|
|
{
|
|
"status": "completed",
|
|
"summary": "The full test suite is still running.",
|
|
"changed_files": ["src/a.py"],
|
|
"tests_run": ["pytest -q — in progress"],
|
|
"artifacts": [],
|
|
"blockers": [],
|
|
},
|
|
False,
|
|
),
|
|
"block",
|
|
),
|
|
],
|
|
)
|
|
def test_claim_requires_structured_evidence_and_surfaces_artifacts(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
result,
|
|
expected_action,
|
|
):
|
|
task = SimpleNamespace(
|
|
id="t_worker",
|
|
current_run_id=4,
|
|
assignee="cli-auto",
|
|
max_runtime_seconds=60,
|
|
)
|
|
calls = []
|
|
heartbeats = []
|
|
connections = []
|
|
artifact = tmp_path / "reports/result.json"
|
|
artifact.parent.mkdir()
|
|
artifact.write_text("{}\n", encoding="utf-8")
|
|
|
|
class Connection:
|
|
def __init__(self):
|
|
self.closed = False
|
|
|
|
def close(self):
|
|
self.closed = True
|
|
|
|
def connect(*, board):
|
|
assert board == "cassandra"
|
|
connection = Connection()
|
|
connections.append(connection)
|
|
return connection
|
|
|
|
fake_db = SimpleNamespace(
|
|
scoped_current_board=lambda _board: nullcontext(),
|
|
connect=connect,
|
|
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_worker"),
|
|
set_branch_name=lambda *_args: None,
|
|
set_workspace_path=lambda *_args: None,
|
|
build_worker_context=lambda *_args: "bounded objective",
|
|
heartbeat_worker=lambda _conn, _task_id, *, note, expected_run_id: (
|
|
heartbeats.append((note, expected_run_id)) or True
|
|
),
|
|
add_comment=lambda *_args: None,
|
|
complete_task=lambda *_args, **kwargs: calls.append(("complete", kwargs)),
|
|
block_task=lambda *_args, **kwargs: calls.append(("block", 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")
|
|
monkeypatch.setattr(
|
|
lanes,
|
|
"select_route",
|
|
lambda *_args, **_kwargs: lanes.Route(
|
|
"codex", "gpt-5.6-sol", "high", "codex-high", "jetson", "vote", 1, ()
|
|
),
|
|
)
|
|
def run_provider(*args, **_kwargs):
|
|
assert connections[0].closed
|
|
before_heartbeat = len(connections)
|
|
assert args[6]("working") is True
|
|
assert len(connections) == before_heartbeat + 1
|
|
assert connections[-1].closed
|
|
return result
|
|
|
|
monkeypatch.setattr(lanes, "run_provider", run_provider)
|
|
|
|
lanes.execute_claim("cassandra", "t_worker")
|
|
|
|
assert calls[0][0] == expected_action
|
|
assert heartbeats == [("working", 4)]
|
|
if expected_action == "complete":
|
|
assert calls[0][1]["metadata"]["artifacts"] == [str(artifact)]
|
|
assert calls[0][1]["metadata"]["tests_run"] == ["pytest -q"]
|
|
else:
|
|
assert calls[0][1]["kind"] == "capability"
|
|
assert all(connection.closed for connection in connections)
|
|
|
|
|
|
def test_goal_card_continues_after_local_judge_rejects_progress(
|
|
tmp_path: Path,
|
|
monkeypatch,
|
|
):
|
|
task = SimpleNamespace(
|
|
id="t_goal",
|
|
current_run_id=12,
|
|
assignee="cli-auto",
|
|
max_runtime_seconds=300,
|
|
goal_mode=True,
|
|
goal_max_turns=3,
|
|
)
|
|
calls = []
|
|
comments = []
|
|
|
|
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_goal"),
|
|
set_branch_name=lambda *_args: None,
|
|
set_workspace_path=lambda *_args: None,
|
|
build_worker_context=lambda *_args: "Run tests, commit, push, and verify remote HEAD.",
|
|
heartbeat_worker=lambda *_args, **_kwargs: True,
|
|
add_comment=lambda _conn, _task_id, _author, body: comments.append(body),
|
|
complete_task=lambda *_args, **kwargs: calls.append(("complete", kwargs)),
|
|
block_task=lambda *_args, **kwargs: calls.append(("block", 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",
|
|
)
|
|
claude_low = lanes.Route(
|
|
"claude", "claude-fable-5", "low", "claude-low", "jetson", "vote", 1, ()
|
|
)
|
|
codex_low = lanes.Route(
|
|
"codex", "gpt-5.6-luna", "low", "codex-low", "manual", "fallback", 1, ()
|
|
)
|
|
codex_xhigh = lanes.Route(
|
|
"codex", "gpt-5.6-sol", "xhigh", "codex-xhigh", "jetson", "escalated", 1, ()
|
|
)
|
|
route_calls = []
|
|
|
|
def select_route(_prompt, assignee, **kwargs):
|
|
route_calls.append((assignee, kwargs))
|
|
if assignee == "cli-codex-low":
|
|
return codex_low
|
|
if len(route_calls) == 1:
|
|
return claude_low
|
|
return codex_xhigh
|
|
|
|
monkeypatch.setattr(lanes, "select_route", select_route)
|
|
monkeypatch.setattr(lanes, "fresh_unavailable_provider", lambda: None)
|
|
reports = [
|
|
lanes.ProcessResult(1, "authentication expired", None, True),
|
|
lanes.ProcessResult(
|
|
0,
|
|
"first turn",
|
|
{
|
|
"status": "completed",
|
|
"summary": "Focused tests passed.",
|
|
"changed_files": ["src/a.py"],
|
|
"tests_run": ["pytest focused: passed"],
|
|
"artifacts": [],
|
|
"blockers": [],
|
|
},
|
|
False,
|
|
),
|
|
lanes.ProcessResult(
|
|
0,
|
|
"second turn",
|
|
{
|
|
"status": "completed",
|
|
"summary": "Full tests passed; commit pushed and remote HEAD verified.",
|
|
"changed_files": ["src/a.py"],
|
|
"tests_run": ["pytest full: passed"],
|
|
"artifacts": [],
|
|
"blockers": [],
|
|
},
|
|
False,
|
|
),
|
|
]
|
|
monkeypatch.setattr(lanes, "run_provider", lambda *_args, **_kwargs: reports.pop(0))
|
|
verdicts = iter(
|
|
[
|
|
(False, "commit, push, and remote verification are missing"),
|
|
(True, "all explicit acceptance criteria have evidence"),
|
|
]
|
|
)
|
|
judge_contexts = []
|
|
|
|
def judge_goal_completion(objective, *_args, **_kwargs):
|
|
judge_contexts.append(objective)
|
|
return next(verdicts)
|
|
|
|
monkeypatch.setattr(
|
|
lanes.cli_lane_goal,
|
|
"judge_goal_completion",
|
|
judge_goal_completion,
|
|
)
|
|
|
|
lanes.execute_claim("cassandra", "t_goal")
|
|
|
|
assert calls[0][0] == "complete"
|
|
assert calls[0][1]["metadata"]["goal_turn"] == 2
|
|
assert any("Goal completion rejected; continuing turn 2/3" in item for item in comments)
|
|
assert any("Goal route 2/3: codex/gpt-5.6-sol at xhigh" in item for item in comments)
|
|
assert route_calls[2][1]["exclude_provider"] == "claude"
|
|
assert "prior rejected reports" in judge_contexts[1]
|
|
assert "commit, push, and remote verification are missing" in judge_contexts[1]
|
|
assert reports == []
|
|
|
|
|
|
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"]
|
|
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_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"
|
|
|
|
|
|
def test_broker_services_survive_sibling_container_readiness_loss():
|
|
services = _services()
|
|
|
|
for name in (
|
|
"hermes-image-broker",
|
|
"hermes-codex-broker",
|
|
"hermes-local-image",
|
|
"hermes-claude-broker",
|
|
):
|
|
assert services[name]["spec"]["publishNotReadyAddresses"] is True
|
|
|
|
|
|
def test_agent_dashboard_reconnects_all_transient_websockets():
|
|
dockerfile = (
|
|
HERMES.parents[1] / "dockerfiles/Dockerfile.hermes-agent"
|
|
).read_text(encoding="utf-8")
|
|
assert "eventsRetryAttempt.current" in dockerfile
|
|
assert "if (!unmounting) setVersion((v) => v + 1);" in dockerfile
|
|
assert "events feed rejected (${ev.code}) — reload the page" in dockerfile
|
|
assert 'url = await api.buildWsUrl("/api/pty", params);' in dockerfile
|
|
assert 'url = await buildWsUrl("/api/events", { channel });' in dockerfile
|
|
assert dockerfile.count("' await api.getSessions(1, 0,") == 2
|
|
assert ".then(() => gw.connect())" in dockerfile
|
|
assert 'api.getSessions(1, 0, profile ?? "")' in dockerfile
|
|
assert "dashboard token rotated by a server restart" in dockerfile
|
|
|
|
|
|
def test_agent_refreshes_routes_after_restoring_cli_logins():
|
|
deployment = _agent_deployment()
|
|
init_containers = {
|
|
item["name"]: item
|
|
for item in deployment["spec"]["template"]["spec"]["initContainers"]
|
|
}
|
|
configure = init_containers["configure-agent-clients"]
|
|
command = configure["command"][-1]
|
|
assert "configure_agent_clients.py" in command
|
|
assert command.index("configure_agent_clients.py") < command.index(
|
|
"hermes_coordinator.py --once"
|
|
)
|
|
env = {item["name"]: item["value"] for item in configure["env"]}
|
|
assert env["HERMES_AUTH_FILE"] == "/runtime-access/hermes-auth.json"
|
|
assert env["PYTHONPATH"] == "/opt/hermes"
|
|
for name in ("bootstrap-coordinator", "configure-agent-clients"):
|
|
route_env = {
|
|
item["name"]: item["value"]
|
|
for item in init_containers[name]["env"]
|
|
}
|
|
assert route_env["CODEX_HOME"] == "/runtime-access/codex"
|
|
assert route_env["CLAUDE_CONFIG_DIR"] == "/runtime-access/claude"
|
|
assert "/opt/data/tools/bin" in route_env["PATH"]
|
|
|
|
assert "patch-web-session-activity" in init_containers
|
|
web_patch = init_containers["patch-web-session-activity"]
|
|
assert "/opt/coordinator/patch_web_session_activity.py" in web_patch["command"]
|
|
|
|
containers = {
|
|
item["name"]: item
|
|
for item in deployment["spec"]["template"]["spec"]["containers"]
|
|
}
|
|
steward_env = {
|
|
item["name"]: item["value"]
|
|
for item in containers["model-steward"]["env"]
|
|
}
|
|
assert steward_env["CODEX_HOME"] == "/runtime-access/codex"
|
|
assert steward_env["CLAUDE_CONFIG_DIR"] == "/runtime-access/claude"
|
|
assert "/opt/data/tools/bin" in steward_env["PATH"]
|
|
hermes_mounts = {
|
|
(item["name"], item["mountPath"], item.get("subPath"))
|
|
for item in containers["hermes"]["volumeMounts"]
|
|
}
|
|
assert (
|
|
"web-server-patch",
|
|
"/opt/hermes/hermes_cli/web_server.py",
|
|
"web_server.py",
|
|
) in hermes_mounts
|
|
|
|
|
|
def test_flux_health_checks_follow_the_owner_oauth_sidecar():
|
|
flux = yaml.safe_load(FLUX_HERMES.read_text())
|
|
checks = {
|
|
(item["kind"], item["name"])
|
|
for item in flux["spec"]["healthChecks"]
|
|
}
|
|
assert ("Deployment", "hermes-agent") in checks
|
|
assert ("DaemonSet", "hermes-node-ssh-access") in checks
|
|
assert ("Deployment", "oauth2-proxy-hermes-agent") not in checks
|
|
|
|
|
|
def test_agent_auth_is_bstein_group_and_email_bounded():
|
|
deployment = _agent_deployment()
|
|
oauth = next(
|
|
item for item in deployment["spec"]["template"]["spec"]["containers"]
|
|
if item["name"] == "oauth2-proxy"
|
|
)
|
|
args = oauth["args"]
|
|
assert "--user-id-claim=sub" in args
|
|
assert "--oidc-groups-claim=groups" in args
|
|
assert "--allowed-group=/hermes-owner" in args
|
|
assert "--authenticated-emails-file=/etc/oauth2-proxy/allowed-emails" in args
|
|
script = (KEYCLOAK / "scripts/hermes_access_oidc_ensure.sh").read_text()
|
|
assert 'group_name="hermes-owner"' in script
|
|
assert "username=bstein&exact=true" in script
|
|
assert '"full.path":"true"' in script
|
|
|
|
|
|
def test_agent_network_boundary_allows_only_authenticated_and_metrics_surfaces():
|
|
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-agent-isolation")
|
|
assert isolation["spec"]["ingress"] == [
|
|
{
|
|
"from": [
|
|
{
|
|
"namespaceSelector": {
|
|
"matchLabels": {
|
|
"kubernetes.io/metadata.name": "traefik"
|
|
}
|
|
},
|
|
"podSelector": {
|
|
"matchLabels": {"app.kubernetes.io/name": "traefik"}
|
|
},
|
|
}
|
|
],
|
|
"ports": [{"protocol": "TCP", "port": 4180}],
|
|
},
|
|
{
|
|
"from": [
|
|
{
|
|
"podSelector": {
|
|
"matchLabels": {"app": "hermes-chat-tenant"}
|
|
}
|
|
}
|
|
],
|
|
"ports": [
|
|
{"protocol": "TCP", "port": 9002},
|
|
{"protocol": "TCP", "port": 9003},
|
|
],
|
|
},
|
|
{
|
|
"from": [
|
|
{
|
|
"podSelector": {
|
|
"matchLabels": {"app": "hermes-switchyard"}
|
|
}
|
|
}
|
|
],
|
|
"ports": [
|
|
{"protocol": "TCP", "port": 9003},
|
|
{"protocol": "TCP", "port": 9006},
|
|
],
|
|
},
|
|
{
|
|
"from": [
|
|
{
|
|
"namespaceSelector": {
|
|
"matchLabels": {
|
|
"kubernetes.io/metadata.name": "monitoring"
|
|
}
|
|
},
|
|
"podSelector": {"matchLabels": {"app": "server"}},
|
|
}
|
|
],
|
|
"ports": [{"protocol": "TCP", "port": 9010}],
|
|
},
|
|
]
|
|
assert isolation["spec"]["egress"] == [{}]
|
|
|
|
|
|
def test_owner_agent_has_cluster_admin_kubernetes_context():
|
|
config = yaml.safe_load((HERMES / "agent-kubeconfig.yaml").read_text())
|
|
assert config["current-context"] == "atlas-owner"
|
|
assert config["contexts"][0]["context"]["namespace"] == "default"
|
|
rbac_path = HERMES / "agent-rbac.yaml"
|
|
documents = [item for item in yaml.safe_load_all(rbac_path.read_text()) if item]
|
|
binding = next(item for item in documents if item["kind"] == "ClusterRoleBinding")
|
|
assert binding["roleRef"] == {
|
|
"apiGroup": "rbac.authorization.k8s.io",
|
|
"kind": "ClusterRole",
|
|
"name": "cluster-admin",
|
|
}
|
|
assert binding["subjects"] == [
|
|
{"kind": "ServiceAccount", "name": "hermes-agent", "namespace": "hermes"}
|
|
]
|
|
|
|
|
|
def test_owner_agent_has_pinned_dedicated_node_ssh_access():
|
|
deployment = _agent_deployment()
|
|
annotations = deployment["spec"]["template"]["metadata"]["annotations"]
|
|
assert annotations[
|
|
"vault.hashicorp.com/agent-inject-secret-node-ssh-private-key"
|
|
] == "kv/data/atlas/hermes/developer-ssh"
|
|
assert annotations[
|
|
"vault.hashicorp.com/agent-inject-secret-node-ssh-config"
|
|
] == "kv/data/atlas/hermes/developer-ssh"
|
|
assert annotations[
|
|
"vault.hashicorp.com/agent-inject-secret-node-ssh-known-hosts"
|
|
] == "kv/data/atlas/hermes/developer-ssh"
|
|
|
|
init = next(
|
|
item
|
|
for item in deployment["spec"]["template"]["spec"]["initContainers"]
|
|
if item["name"] == "init-config"
|
|
)
|
|
command = init["command"][2]
|
|
assert "ln -s /runtime-access/node-ssh-config /opt/data/home/.ssh/config" in command
|
|
assert (
|
|
"ln -s /runtime-access/node-ssh-known-hosts /opt/data/home/.ssh/known_hosts"
|
|
in command
|
|
)
|
|
assert "ln -s home/.ssh /opt/data/.ssh" in command
|
|
assert "chmod 0700 /opt/data/home/.ssh" in command
|
|
assert (
|
|
"ln -s /runtime-access/node-ssh-private-key "
|
|
"/opt/data/home/.ssh/id_ed25519_atlas_nodes"
|
|
) in command
|
|
|
|
config = yaml.safe_load((HERMES / "agent-configmap.yaml").read_text())["data"]
|
|
assert "ssh_config" not in config
|
|
assert "ssh_known_hosts" not in config
|
|
assert "ssh-ed25519" not in (HERMES / "agent-configmap.yaml").read_text()
|
|
|
|
resources = yaml.safe_load((HERMES / "kustomization.yaml").read_text())[
|
|
"resources"
|
|
]
|
|
assert "node-ssh-access.yaml" in resources
|
|
access = [
|
|
item
|
|
for item in yaml.safe_load_all((HERMES / "node-ssh-access.yaml").read_text())
|
|
if item
|
|
]
|
|
service_account = next(item for item in access if item["kind"] == "ServiceAccount")
|
|
assert service_account["metadata"]["name"] == "hermes-node-ssh-access"
|
|
provider = next(item for item in access if item["kind"] == "SecretProviderClass")
|
|
assert provider["metadata"]["name"] == "hermes-node-ssh-access"
|
|
assert provider["spec"]["provider"] == "vault"
|
|
parameters = provider["spec"]["parameters"]
|
|
assert parameters["roleName"] == "hermes-node-ssh"
|
|
assert 'secretPath: "kv/data/atlas/hermes/developer-ssh"' in parameters["objects"]
|
|
assert 'secretKey: "public_key"' in parameters["objects"]
|
|
daemonset = next(item for item in access if item["kind"] == "DaemonSet")
|
|
pod = daemonset["spec"]["template"]["spec"]
|
|
assert pod["serviceAccountName"] == "hermes-node-ssh-access"
|
|
assert pod["automountServiceAccountToken"] is True
|
|
host_home = next(item for item in pod["volumes"] if item["name"] == "host-home")
|
|
assert host_home["hostPath"] == {"path": "/home", "type": "Directory"}
|
|
vault_secrets = next(
|
|
item for item in pod["volumes"] if item["name"] == "vault-secrets"
|
|
)
|
|
assert vault_secrets["csi"]["driver"] == "secrets-store.csi.k8s.io"
|
|
assert vault_secrets["csi"]["volumeAttributes"] == {
|
|
"secretProviderClass": "hermes-node-ssh-access"
|
|
}
|
|
reconciler = pod["containers"][0]["args"][0]
|
|
assert "cat /vault/secrets/node-ssh-public-key" in reconciler
|
|
assert "grep -qxF" in reconciler
|
|
assert "for user in atlas oceanus" in reconciler
|
|
assert "/host-etc/passwd" in reconciler
|
|
assert "chown \"${uid}:${gid}\"" in reconciler
|
|
host_passwd = next(
|
|
item for item in pod["volumes"] if item["name"] == "host-passwd"
|
|
)
|
|
assert host_passwd["hostPath"] == {"path": "/etc/passwd", "type": "File"}
|
|
|
|
|
|
def test_owner_agent_tracks_no_ssh_identity_or_host_key_material():
|
|
"""Vault references may be tracked; SSH identities and trust data may not."""
|
|
forbidden = (
|
|
"BEGIN OPENSSH PRIVATE KEY",
|
|
"ssh-ed25519 AAAA",
|
|
"ssh-rsa AAAA",
|
|
"IdentityFile ",
|
|
"UserKnownHostsFile ",
|
|
"StrictHostKeyChecking ",
|
|
"ssh_config:",
|
|
"ssh_known_hosts:",
|
|
)
|
|
text_suffixes = {
|
|
".conf",
|
|
".json",
|
|
".md",
|
|
".py",
|
|
".sh",
|
|
".toml",
|
|
".yaml",
|
|
".yml",
|
|
}
|
|
tracked = "\n".join(
|
|
path.read_text(encoding="utf-8")
|
|
for path in HERMES.rglob("*")
|
|
if path.is_file() and path.suffix in text_suffixes
|
|
)
|
|
for marker in forbidden:
|
|
assert marker not in tracked
|
|
|
|
|
|
def test_switchyard_has_a_dedicated_non_owner_identity_and_read_only_catalog():
|
|
"""Routing must not inherit the owner agent's cluster-admin capability."""
|
|
service_accounts = [
|
|
item
|
|
for item in yaml.safe_load_all(
|
|
(HERMES / "vault-serviceaccount.yaml").read_text()
|
|
)
|
|
if item
|
|
]
|
|
assert any(
|
|
item["kind"] == "ServiceAccount"
|
|
and item["metadata"]["name"] == "hermes-switchyard"
|
|
for item in service_accounts
|
|
)
|
|
|
|
switchyard = yaml.safe_load(
|
|
(HERMES / "switchyard-deployment.yaml").read_text()
|
|
)
|
|
switchyard_pod = switchyard["spec"]["template"]["spec"]
|
|
assert switchyard_pod["serviceAccountName"] == "hermes-switchyard"
|
|
agent_pod = _agent_deployment()["spec"]["template"]["spec"]
|
|
for pod, container_name in (
|
|
(switchyard_pod, "worker-route-broker"),
|
|
(agent_pod, "claude-broker"),
|
|
):
|
|
container = next(item for item in pod["containers"] if item["name"] == container_name)
|
|
catalog = next(
|
|
item
|
|
for item in container["volumeMounts"]
|
|
if item["mountPath"] == "/routing-catalog"
|
|
)
|
|
assert catalog["readOnly"] is True
|
|
|
|
rbac = [
|
|
item
|
|
for item in yaml.safe_load_all((HERMES / "agent-rbac.yaml").read_text())
|
|
if item
|
|
]
|
|
binding = next(item for item in rbac if item["kind"] == "ClusterRoleBinding")
|
|
assert binding["subjects"] == [
|
|
{"kind": "ServiceAccount", "name": "hermes-agent", "namespace": "hermes"}
|
|
]
|
|
|
|
|
|
def test_switchyard_active_state_uses_a_relocatable_rwx_claim():
|
|
"""A stale node attachment must not strand the routing authority."""
|
|
claims = [
|
|
item
|
|
for item in yaml.safe_load_all((HERMES / "switchyard-pvc.yaml").read_text())
|
|
if item
|
|
]
|
|
active_claim = next(
|
|
item
|
|
for item in claims
|
|
if item["metadata"]["name"] == "hermes-switchyard-state-rwx"
|
|
)
|
|
assert active_claim["spec"]["accessModes"] == ["ReadWriteMany"]
|
|
|
|
deployment = yaml.safe_load((HERMES / "switchyard-deployment.yaml").read_text())
|
|
strategy = deployment["spec"]["strategy"]
|
|
assert strategy == {
|
|
"type": "RollingUpdate",
|
|
"rollingUpdate": {"maxSurge": 1, "maxUnavailable": 0},
|
|
}
|
|
pod = deployment["spec"]["template"]["spec"]
|
|
state = next(item for item in pod["volumes"] if item["name"] == "state")
|
|
assert state["persistentVolumeClaim"]["claimName"] == active_claim["metadata"][
|
|
"name"
|
|
]
|
|
|
|
|
|
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 rule.get("ports") == [{"protocol": "TCP", "port": 9005}]
|
|
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
|
|
assert "go1.26.5.linux-arm64.tar.gz" in script
|
|
assert (
|
|
"fe4789e92b1f33358680864bbe8704289e7bb5fc207d80623c308935bd696d49"
|
|
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_only_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 "agent.hermes.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>")
|