389 lines
12 KiB
Python
389 lines
12 KiB
Python
"""Provider routing and process-isolation contracts for Hermes CLI lanes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import signal
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
|
|
from testing.tests.test_hermes_cli_lanes_support import (
|
|
_SwitchyardResponse,
|
|
lanes,
|
|
)
|
|
|
|
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_explicit_auth_failure_survives_restart_health_gap(tmp_path: Path, monkeypatch):
|
|
paths = {
|
|
"codex": tmp_path / "codex.json",
|
|
"claude": tmp_path / "claude.json",
|
|
}
|
|
paths["codex"].write_text(
|
|
'{"state":"available","authenticated":true}\n', encoding="utf-8"
|
|
)
|
|
paths["claude"].write_text(
|
|
'{"state":"unavailable","authenticated":false}\n', encoding="utf-8"
|
|
)
|
|
monkeypatch.setattr(lanes, "PROVIDER_HEALTH_PATHS", paths)
|
|
|
|
stale_during_rollout = paths["claude"].stat().st_mtime + 10 * 60
|
|
assert lanes.fresh_unavailable_provider(now=stale_during_rollout) == "claude"
|
|
|
|
paths["claude"].write_text(
|
|
'{"state":"unavailable","authenticated":true}\n', encoding="utf-8"
|
|
)
|
|
transient_stale = paths["claude"].stat().st_mtime + 10 * 60
|
|
assert lanes.fresh_unavailable_provider(now=transient_stale) 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)]
|