393 lines
13 KiB
Python
393 lines
13 KiB
Python
"""Kanban claim and lifecycle contracts for Hermes CLI lanes."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import sys
|
|
from contextlib import nullcontext
|
|
from pathlib import Path
|
|
from types import SimpleNamespace
|
|
|
|
import pytest
|
|
|
|
from testing.tests.test_hermes_cli_lanes_support import (
|
|
lanes,
|
|
)
|
|
|
|
|
|
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 == []
|