323 lines
13 KiB
Python
323 lines
13 KiB
Python
"""Model worker path, evidence, retention, and readiness contracts."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import sys
|
|
import urllib.error
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
ROOT = Path(__file__).parents[2]
|
|
SCRIPTS = ROOT / "services/hermes/scripts"
|
|
sys.path.insert(0, str(SCRIPTS))
|
|
|
|
import execution_pool_protocol as protocol # noqa: E402
|
|
import execution_pool_worker as worker # noqa: E402
|
|
|
|
|
|
def assignment(**changes):
|
|
value = {
|
|
"board": "metis", "task_id": "t_deadbeef", "run_id": "23",
|
|
"worker_ordinal": 0, "attempt": 1, "protocol_version": 2,
|
|
"workspace": "/workspace/runs/metis/t_deadbeef/23",
|
|
"baseline_sha": "a" * 40,
|
|
"payload": {"context": "safe objective", "assignee": "cli-auto"},
|
|
}
|
|
value.update(changes)
|
|
return value
|
|
|
|
|
|
class Response:
|
|
def __init__(self, value):
|
|
self.value = value
|
|
|
|
def __enter__(self):
|
|
return self
|
|
|
|
def __exit__(self, *_args):
|
|
return False
|
|
|
|
def read(self, _size=-1):
|
|
return self.value
|
|
|
|
|
|
def test_post_client_and_poll_validate_every_response_boundary(monkeypatch):
|
|
monkeypatch.setattr(
|
|
worker.urllib.request, "urlopen",
|
|
lambda *_a, **_k: Response(json.dumps({"safe": True}).encode()),
|
|
)
|
|
assert worker._post("http://mediator", {}) == {"safe": True}
|
|
monkeypatch.setattr(
|
|
worker.urllib.request, "urlopen",
|
|
lambda *_a, **_k: Response(b"x" * (64 * 1024 + 1)),
|
|
)
|
|
with pytest.raises(protocol.ProtocolError, match="wire"):
|
|
worker._post("http://mediator", {})
|
|
monkeypatch.setattr(
|
|
worker.urllib.request, "urlopen", lambda *_a, **_k: Response(b"[]")
|
|
)
|
|
with pytest.raises(protocol.ProtocolError, match="object"):
|
|
worker._post("http://mediator", {})
|
|
|
|
monkeypatch.setattr(worker, "_post", lambda *_a, **_k: {"error": "denied"})
|
|
with pytest.raises(protocol.ProtocolError, match="denied"):
|
|
worker._client("poll")
|
|
monkeypatch.setattr(worker, "_post", lambda *_a, **_k: {"safe": True})
|
|
assert worker._client("poll") == {"safe": True}
|
|
monkeypatch.setattr(worker, "ORDINAL", 0)
|
|
monkeypatch.setattr(worker, "_client", lambda *_a, **_k: {"assignment": None})
|
|
assert worker._poll() is None
|
|
for value in ([], assignment(worker_ordinal=1), assignment(protocol_version=1)):
|
|
monkeypatch.setattr(worker, "_client", lambda *_a, value=value, **_k: {"assignment": value})
|
|
with pytest.raises(protocol.ProtocolError, match="foreign"):
|
|
worker._poll()
|
|
monkeypatch.setattr(worker, "_client", lambda *_a, **_k: {"assignment": assignment()})
|
|
assert worker._poll()["task_id"] == "t_deadbeef"
|
|
assert worker._binding(assignment()) == {
|
|
"board": "metis", "task_id": "t_deadbeef", "run_id": "23",
|
|
"worker_ordinal": 0, "attempt": 1,
|
|
}
|
|
|
|
|
|
def test_state_path_rejects_traversal_root_and_leaf_symlinks(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(worker, "ROOT", tmp_path)
|
|
path = worker._state_path(assignment())
|
|
assert path == tmp_path / "session-state/metis/t_deadbeef/23.json"
|
|
with pytest.raises(protocol.ProtocolError, match="invalid"):
|
|
worker._state_path(assignment(task_id="../bad"))
|
|
|
|
state_root = tmp_path / "session-state"
|
|
outside = tmp_path / "outside"
|
|
path.parent.rmdir()
|
|
path.parent.parent.rmdir()
|
|
state_root.rmdir()
|
|
outside.mkdir()
|
|
state_root.symlink_to(outside, target_is_directory=True)
|
|
with pytest.raises(protocol.ProtocolError, match="root"):
|
|
worker._state_path(assignment())
|
|
state_root.unlink()
|
|
path = worker._state_path(assignment())
|
|
path.symlink_to("/etc/passwd")
|
|
with pytest.raises(protocol.ProtocolError, match="symlink"):
|
|
worker._state_path(assignment())
|
|
|
|
|
|
def prepare_provider_roots(tmp_path, monkeypatch):
|
|
worker_root = tmp_path / "worker"
|
|
data_root = tmp_path / "data"
|
|
codex = tmp_path / "runtime/codex"
|
|
claude = tmp_path / "runtime/claude"
|
|
(worker_root / "provider-state").mkdir(parents=True)
|
|
data_root.mkdir()
|
|
codex.mkdir(parents=True)
|
|
claude.mkdir(parents=True)
|
|
monkeypatch.setattr(worker, "ROOT", worker_root)
|
|
monkeypatch.setattr(worker.cli_lane_runner, "DATA_ROOT", data_root)
|
|
monkeypatch.setenv("CODEX_HOME", str(codex))
|
|
monkeypatch.setenv("CLAUDE_CONFIG_DIR", str(claude))
|
|
return worker_root, data_root, codex, claude
|
|
|
|
|
|
def test_provider_session_binding_replaces_only_owned_symlinks(tmp_path, monkeypatch):
|
|
worker_root, data_root, codex, claude = prepare_provider_roots(tmp_path, monkeypatch)
|
|
old = tmp_path / "old"
|
|
old.mkdir()
|
|
(data_root / "home").symlink_to(old, target_is_directory=True)
|
|
(codex / "sessions").symlink_to(old, target_is_directory=True)
|
|
worker._bind_provider_sessions(assignment())
|
|
assert "metis/t_deadbeef/23" in str((data_root / "home").resolve())
|
|
assert "metis/t_deadbeef/23" in str((codex / "sessions").resolve())
|
|
settings = worker_root / "provider-state/metis/t_deadbeef/23/home/.claude/settings.json"
|
|
assert json.loads(settings.read_text()) == {}
|
|
worker._bind_provider_sessions(assignment())
|
|
|
|
with pytest.raises(protocol.ProtocolError, match="binding"):
|
|
worker._bind_provider_sessions(assignment(run_id="../bad"))
|
|
monkeypatch.setattr(worker, "ROOT", tmp_path / "missing")
|
|
with pytest.raises(protocol.ProtocolError, match="unavailable"):
|
|
worker._bind_provider_sessions(assignment())
|
|
|
|
|
|
def test_provider_session_binding_rejects_durable_and_runtime_tampering(
|
|
tmp_path, monkeypatch
|
|
):
|
|
worker_root, data_root, codex, _claude = prepare_provider_roots(tmp_path, monkeypatch)
|
|
run_parent = worker_root / "provider-state/metis"
|
|
run_parent.symlink_to(tmp_path, target_is_directory=True)
|
|
with pytest.raises(protocol.ProtocolError, match="session path"):
|
|
worker._bind_provider_sessions(assignment())
|
|
run_parent.unlink()
|
|
home = worker_root / "provider-state/metis/t_deadbeef/23/home"
|
|
home.parent.mkdir(parents=True)
|
|
home.symlink_to(tmp_path, target_is_directory=True)
|
|
with pytest.raises(protocol.ProtocolError, match="HOME"):
|
|
worker._bind_provider_sessions(assignment())
|
|
home.unlink()
|
|
(data_root / "home").write_text("not-owned")
|
|
with pytest.raises(protocol.ProtocolError, match="task-bound"):
|
|
worker._bind_provider_sessions(assignment())
|
|
(data_root / "home").unlink()
|
|
(codex / "sessions").write_text("not-owned")
|
|
with pytest.raises(protocol.ProtocolError, match="not a symlink"):
|
|
worker._bind_provider_sessions(assignment())
|
|
(codex / "sessions").unlink()
|
|
durable_sessions = (
|
|
worker_root / "provider-state/metis/t_deadbeef/23/codex/sessions"
|
|
)
|
|
durable_sessions.rmdir()
|
|
durable_sessions.symlink_to(tmp_path, target_is_directory=True)
|
|
with pytest.raises(protocol.ProtocolError, match="session path"):
|
|
worker._bind_provider_sessions(assignment())
|
|
|
|
|
|
def test_prompt_activity_git_and_result_bounding(tmp_path, monkeypatch):
|
|
text = worker._prompt("objective", tmp_path, worker._binding(assignment()))
|
|
assert "no Kubernetes identity" in text and "objective" in text
|
|
missing = tmp_path / "missing.log"
|
|
assert worker._read_activity(missing, 7) == ("", 7)
|
|
log = tmp_path / "worker.log"
|
|
log.write_text("abc")
|
|
assert worker._read_activity(log, 0) == ("abc", 3)
|
|
assert worker._read_activity(log, 99) == ("abc", 3)
|
|
fifo = tmp_path / "fifo"
|
|
os.mkfifo(fifo)
|
|
with pytest.raises(protocol.ProtocolError, match="regular"):
|
|
worker._read_activity(fifo, 0)
|
|
|
|
repo = tmp_path / "repo"
|
|
subprocess.run(["git", "init", "-q", str(repo)], check=True)
|
|
assert worker._git(repo, "status", "--porcelain") == ""
|
|
with pytest.raises(RuntimeError, match="ambiguous"):
|
|
worker._git(repo, "rev-parse", "missing")
|
|
|
|
value = {
|
|
"status": "completed", "summary": "s" * 20_000,
|
|
"changed_files": ["x" * 3000] * 100,
|
|
"tests_run": "not-list", "artifacts": [], "findings": [], "blockers": [],
|
|
}
|
|
bounded = worker._bounded_result(value)
|
|
assert len(protocol.canonical_json(bounded)) <= 32 * 1024
|
|
assert bounded["tests_run"] == [] and len(bounded["summary"]) <= 8000
|
|
|
|
|
|
def test_refresh_assignment_requires_exact_binding(monkeypatch):
|
|
exact = worker._binding(assignment())
|
|
monkeypatch.setattr(worker, "_poll", lambda: assignment())
|
|
assert worker._refresh_assignment(exact)["run_id"] == "23"
|
|
for value in (None, assignment(attempt=2)):
|
|
monkeypatch.setattr(worker, "_poll", lambda value=value: value)
|
|
with pytest.raises(protocol.ProtocolError, match="changed"):
|
|
worker._refresh_assignment(exact)
|
|
|
|
|
|
def write_gc_state(root, task_id, workspace, terminal_at):
|
|
path = root / f"session-state/metis/{task_id}/23.json"
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps({"terminal_at": terminal_at, "workspace": str(workspace)}))
|
|
return path
|
|
|
|
|
|
def init_clean_repo(path):
|
|
subprocess.run(["git", "init", "-q", str(path)], check=True)
|
|
subprocess.run(["git", "-C", str(path), "config", "user.email", "a@b.c"], check=True)
|
|
subprocess.run(["git", "-C", str(path), "config", "user.name", "Test"], check=True)
|
|
(path / "tracked").write_text("safe")
|
|
subprocess.run(["git", "-C", str(path), "add", "tracked"], check=True)
|
|
subprocess.run(["git", "-C", str(path), "commit", "-qm", "initial"], check=True)
|
|
|
|
|
|
def test_retention_skips_young_outside_dirty_and_symlink_workspaces(tmp_path, monkeypatch):
|
|
monkeypatch.setattr(worker, "ROOT", tmp_path)
|
|
monkeypatch.setattr(worker, "RETENTION_SECONDS", 3600)
|
|
run_root = tmp_path / "runs/metis"
|
|
clean = run_root / "clean/23"
|
|
dirty = run_root / "dirty/23"
|
|
clean.mkdir(parents=True)
|
|
dirty.mkdir(parents=True)
|
|
init_clean_repo(clean)
|
|
init_clean_repo(dirty)
|
|
(dirty / "untracked").write_text("dirty")
|
|
now = 10_000
|
|
write_gc_state(tmp_path, "clean", clean, 1)
|
|
dirty_state = write_gc_state(tmp_path, "dirty", dirty, 1)
|
|
young = write_gc_state(tmp_path, "young", clean, now)
|
|
outside = write_gc_state(tmp_path, "outside", tmp_path / "missing", 1)
|
|
assert worker.garbage_collect(now=now) == 1
|
|
assert not clean.exists() and dirty.exists()
|
|
assert dirty_state.exists() and young.exists() and outside.exists()
|
|
|
|
link = run_root / "link/23"
|
|
link.parent.mkdir()
|
|
link.symlink_to(dirty, target_is_directory=True)
|
|
link_state = write_gc_state(tmp_path, "link", link, 1)
|
|
assert worker.garbage_collect(now=now) == 0
|
|
assert link_state.exists()
|
|
|
|
|
|
def test_readiness_checks_ordinal_paths_credentials_and_mediator(tmp_path, monkeypatch):
|
|
worker_root = tmp_path / "worker"
|
|
data_root = tmp_path / "data"
|
|
claude_token = tmp_path / "claude-oauth/token"
|
|
for path in (worker_root, worker_root / "provider-state", data_root):
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
claude_token.parent.mkdir()
|
|
claude_token.write_text("setup-token")
|
|
schema = tmp_path / "schema/result.json"
|
|
monkeypatch.setattr(worker, "ORDINAL", 0)
|
|
monkeypatch.setattr(worker, "ROOT", worker_root)
|
|
monkeypatch.setattr(worker.cli_lane_runner, "DATA_ROOT", data_root)
|
|
monkeypatch.setattr(worker.cli_lane_runner, "RESULT_SCHEMA_PATH", schema)
|
|
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN_FILE", str(claude_token))
|
|
polled = []
|
|
monkeypatch.setattr(worker, "_poll", lambda: polled.append(True))
|
|
worker.readiness()
|
|
assert polled and json.loads(schema.read_text()) == worker.cli_lane_runner.RESULT_SCHEMA
|
|
|
|
monkeypatch.setattr(worker, "ORDINAL", 3)
|
|
with pytest.raises(protocol.ProtocolError, match="ordinal"):
|
|
worker.readiness()
|
|
monkeypatch.setattr(worker, "ORDINAL", 0)
|
|
monkeypatch.setattr(worker, "ROOT", tmp_path / "missing")
|
|
with pytest.raises(protocol.ProtocolError, match="path"):
|
|
worker.readiness()
|
|
monkeypatch.setattr(worker, "ROOT", worker_root)
|
|
claude_token.unlink()
|
|
with pytest.raises(protocol.ProtocolError, match="credential"):
|
|
worker.readiness()
|
|
|
|
|
|
def test_readiness_defers_transient_ordinal_lease_conflict(tmp_path, monkeypatch):
|
|
worker_root = tmp_path / "worker"
|
|
data_root = tmp_path / "data"
|
|
claude_token = tmp_path / "claude-oauth/token"
|
|
for path in (worker_root, worker_root / "provider-state", data_root):
|
|
path.mkdir(parents=True, exist_ok=True)
|
|
claude_token.parent.mkdir()
|
|
claude_token.write_text("setup-token")
|
|
monkeypatch.setattr(worker, "ORDINAL", 1)
|
|
monkeypatch.setattr(worker, "ROOT", worker_root)
|
|
monkeypatch.setattr(worker.cli_lane_runner, "DATA_ROOT", data_root)
|
|
monkeypatch.setattr(
|
|
worker.cli_lane_runner, "RESULT_SCHEMA_PATH", tmp_path / "schema/result.json"
|
|
)
|
|
monkeypatch.setenv("CLAUDE_CODE_OAUTH_TOKEN_FILE", str(claude_token))
|
|
|
|
def conflict():
|
|
raise urllib.error.HTTPError("http://mediator", 409, "Conflict", {}, None)
|
|
|
|
monkeypatch.setattr(worker, "_poll", conflict)
|
|
worker.readiness()
|
|
|
|
def unauthorized():
|
|
raise urllib.error.HTTPError("http://mediator", 401, "Unauthorized", {}, None)
|
|
|
|
monkeypatch.setattr(worker, "_poll", unauthorized)
|
|
with pytest.raises(urllib.error.HTTPError) as error:
|
|
worker.readiness()
|
|
assert error.value.code == 401
|