atlas-iac/testing/tests/test_hermes_cli_review_lane.py

"""Lane-level goal-loop behaviour for review and implementation cards."""

from __future__ import annotations

import importlib.util
import json
import sys
from contextlib import nullcontext
from pathlib import Path
from types import SimpleNamespace


SCRIPTS = Path(__file__).parents[2] / "services/hermes/scripts"
sys.path.insert(0, str(SCRIPTS))


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")
# The single-claim loop lives in the runner today and in ``cli_lane_execution``
# once the lane modules are split. Patch whichever module owns it so this file
# describes lane behaviour rather than one file layout.
execution = sys.modules.get("cli_lane_execution", lanes)
goal = sys.modules["cli_lane_goal"]

REVIEW_CARD = """# Kanban task t_review: Independent read-only review of the worker pool

## Body
Perform a fresh independent read-only release review of the pool implementation.
Do not edit files, comment, commit, push, merge, publish, or deploy.
Return strict SHIP or BLOCK with exact file/line and reproducible inputs.
"""
IMPLEMENTATION_CARD = """# Kanban task t_impl: Repair the pool blockers

## Body
Repair every reported blocker, run the focused suites, and open a draft PR.
Return exact pushed SHA and evidence.
"""
BLOCK_REVIEW = {
    "status": "completed",
    "summary": (
        "Independent read-only review of PR #18 at head 2000252. Verdict: BLOCK. "
        "Five reproducible P0 defects fire on the coordinator's first tick and the "
        "review worktree was left pristine."
    ),
    "changed_files": [],
    "tests_run": ["pytest testing/tests/test_pool.py: 82 passed"],
    "artifacts": [],
    "findings": ["P0-1 BLOCKER - coordinator.py:248 compares int to str run ids."],
    "blockers": [],
}


class _Connection:
    def close(self) -> None:
        return None


class _Lane:
    """One instrumented direct-CLI-lane claim over a fake Kanban board."""

    def __init__(self, tmp_path: Path, monkeypatch, *, card: str, task, reports):
        self.calls: list[tuple[str, dict]] = []
        self.comments: list[str] = []
        self.prompts: list[str] = []
        self.routes: list[tuple[str, dict]] = []
        self.heartbeats: list[tuple[str, object]] = []
        self.reports = list(reports)
        self.task = task
        self.state_root = tmp_path / "cli-lanes"
        self.workspace = tmp_path
        self.db = SimpleNamespace(
            scoped_current_board=lambda _board: nullcontext(),
            connect=lambda board=None, **_kwargs: _Connection(),
            get_task=lambda _conn, _task_id: self.task,
            worker_log_path=lambda _task_id, board: tmp_path / "worker.log",
            _resolve_worktree_workspace=lambda _task, board: (tmp_path, "wt/t_card"),
            set_branch_name=lambda *_args: None,
            set_workspace_path=lambda *_args: None,
            build_worker_context=lambda *_args: card,
            heartbeat_worker=lambda _conn, _task_id, *, note, expected_run_id: (
                self.heartbeats.append((note, expected_run_id)) or True
            ),
            add_comment=lambda _conn, _task_id, _author, body: self.comments.append(body),
            complete_task=self._complete_task,
            block_task=self._block_task,
            reclaim_task=self._reclaim_task,
        )
        monkeypatch.setitem(sys.modules, "hermes_cli", SimpleNamespace(kanban_db=self.db))
        # The lane derives durable journal identity from ``STATE_ROOT``, so
        # relocate the root itself rather than stubbing ``state_path``.
        for module in list(sys.modules.values()):
            if getattr(module, "__name__", "").startswith("cli_lane") and hasattr(
                module, "STATE_ROOT"
            ):
                monkeypatch.setattr(module, "STATE_ROOT", self.state_root)
        monkeypatch.setattr(execution, "fresh_unavailable_provider", lambda: None)
        monkeypatch.setattr(execution, "select_route", self._select_route)
        monkeypatch.setattr(execution, "run_provider", self._run_provider)

    def _complete_task(
        self,
        _conn,
        task_id,
        *,
        result,
        summary,
        metadata,
        expected_run_id=None,
        replay_ended_run_id=None,
    ):
        self.calls.append(
            (
                "complete",
                {
                    "task_id": task_id,
                    "result": result,
                    "summary": summary,
                    "metadata": metadata,
                    "expected_run_id": expected_run_id or replay_ended_run_id,
                },
            )
        )
        self.task = SimpleNamespace(
            **{
                **vars(self.task),
                "status": "done",
                "result": result,
                "completed_run_id": expected_run_id or replay_ended_run_id,
            }
        )
        return True

    def _block_task(self, _conn, task_id, *, reason, kind, expected_run_id=None):
        self.calls.append(
            (
                "block",
                {
                    "task_id": task_id,
                    "reason": reason,
                    "kind": kind,
                    "expected_run_id": expected_run_id,
                },
            )
        )
        return True

    def _reclaim_task(self, _conn, _task_id, *, reason=None, expected_run_id=None):
        return True

    @property
    def state_file(self) -> Path:
        return self.state_root / "titan-iac/t_card.json"

    @staticmethod
    def _route(provider: str, effort: str = "xhigh"):
        return lanes.Route(
            provider,
            f"{provider}-model",
            effort,
            f"{provider}-{effort}",
            "switchyard",
            "routed",
            1,
            (),
        )

    def _select_route(self, _prompt, assignee, **kwargs):
        self.routes.append((assignee, kwargs))
        if assignee.startswith("cli-codex"):
            return self._route("codex")
        return self._route("claude")

    def _run_provider(self, *args, **_kwargs):
        self.prompts.append(args[1])
        args[6]("working")
        return self.reports.pop(0)

    @property
    def terminal(self) -> tuple[str, dict]:
        assert len(self.calls) == 1, self.calls
        return self.calls[0]

    @property
    def state(self) -> dict:
        return json.loads(self.state_file.read_text(encoding="utf-8"))

    def rejections(self) -> list[str]:
        return [item for item in self.comments if "Goal completion rejected" in item]


def _task(**overrides):
    value = {
        "id": "t_card",
        "status": "running",
        "current_run_id": 23,
        "completed_run_id": None,
        "result": None,
        "assignee": "cli-claude-xhigh",
        "max_runtime_seconds": 600,
        "goal_mode": True,
        "goal_max_turns": 8,
    }
    value.update(overrides)
    return SimpleNamespace(**value)


def _result(**overrides):
    value = dict(BLOCK_REVIEW)
    value.update(overrides)
    return lanes.ProcessResult(0, "review turn", value, False)


def _no_judge(*_args, **_kwargs):
    raise AssertionError("the model judge must not be consulted for a review card")


def test_completed_block_review_finalizes_on_its_first_turn(tmp_path, monkeypatch):
    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
    lane = _Lane(
        tmp_path,
        monkeypatch,
        card=REVIEW_CARD,
        task=_task(),
        reports=[_result()],
    )

    execution.execute_claim("titan-iac", "t_card")

    action, kwargs = lane.terminal
    assert action == "complete"
    assert kwargs["expected_run_id"] == 23
    assert json.loads(kwargs["result"])["findings"] == BLOCK_REVIEW["findings"]
    assert kwargs["metadata"]["goal_turn"] == 1
    assert "BLOCK verdict with 1 finding(s)" in kwargs["metadata"]["goal_judge_reason"]
    assert lane.rejections() == []
    assert len(lane.prompts) == 1
    assert lane.reports == []


def test_review_verdict_reaches_kanban_without_hidden_mutation(tmp_path, monkeypatch):
    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
    lane = _Lane(
        tmp_path,
        monkeypatch,
        card=REVIEW_CARD,
        task=_task(),
        reports=[_result()],
    )

    execution.execute_claim("titan-iac", "t_card")

    _action, kwargs = lane.terminal
    stored = json.loads(kwargs["result"])
    assert stored["status"] == "completed"
    assert stored["summary"] == BLOCK_REVIEW["summary"]
    assert stored["blockers"] == []
    assert kwargs["summary"] == BLOCK_REVIEW["summary"]
    assert kwargs["metadata"]["findings"] == BLOCK_REVIEW["findings"]
    assert len(kwargs["metadata"]["goal_judge_reason"]) <= goal.JUDGE_REASON_LIMIT


def test_reviewer_provider_fallback_still_finalizes_one_verdict(tmp_path, monkeypatch):
    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
    lane = _Lane(
        tmp_path,
        monkeypatch,
        card=REVIEW_CARD,
        task=_task(assignee="cli-auto"),
        reports=[
            lanes.ProcessResult(1, "usage limit reached", None, True),
            _result(),
        ],
    )

    execution.execute_claim("titan-iac", "t_card")

    action, kwargs = lane.terminal
    assert action == "complete"
    assert kwargs["metadata"]["provider"] == "codex"
    assert kwargs["metadata"]["goal_turn"] == 1
    assert any("Provider fallback: claude -> codex" in item for item in lane.comments)
    assert lane.rejections() == []
    assert len(lane.prompts) == 2


def test_review_restart_resumes_the_next_turn_and_completes_once(tmp_path, monkeypatch):
    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
    lane = _Lane(
        tmp_path,
        monkeypatch,
        card=REVIEW_CARD,
        task=_task(),
        reports=[_result()],
    )
    lane.state_file.parent.mkdir(parents=True, exist_ok=True)
    lane.state_file.write_text(
        json.dumps({"goal_turn": 3, "current_route": {"provider": "codex"}}),
        encoding="utf-8",
    )

    execution.execute_claim("titan-iac", "t_card")

    action, kwargs = lane.terminal
    assert action == "complete"
    assert kwargs["metadata"]["goal_turn"] == 4
    assert lane.state["goal_turn"] == 4
    assert any("Restart-time provider change" in item for item in lane.comments)


def test_malformed_review_is_resumed_without_demanding_an_edit(tmp_path, monkeypatch):
    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
    lane = _Lane(
        tmp_path,
        monkeypatch,
        card=REVIEW_CARD,
        task=_task(goal_max_turns=2),
        reports=[
            _result(summary="Review done.", findings=[]),
            _result(),
        ],
    )

    execution.execute_claim("titan-iac", "t_card")

    action, kwargs = lane.terminal
    assert action == "complete"
    assert kwargs["metadata"]["goal_turn"] == 2
    assert len(lane.rejections()) == 1
    assert goal.READ_ONLY_GUARD in lane.rejections()[0]
    assert "do not modify the reviewed implementation" in lane.prompts[1]
    assert "Goal-loop continuation" in lane.prompts[1]


def test_exhausted_review_budget_blocks_with_a_read_only_reason(tmp_path, monkeypatch):
    monkeypatch.setattr(goal.urllib.request, "urlopen", _no_judge)
    lane = _Lane(
        tmp_path,
        monkeypatch,
        card=REVIEW_CARD,
        task=_task(goal_max_turns=1),
        reports=[_result(summary="Review done.", findings=[])],
    )

    execution.execute_claim("titan-iac", "t_card")

    action, kwargs = lane.terminal
    assert action == "block"
    assert kwargs["expected_run_id"] == 23
    assert goal.READ_ONLY_GUARD in kwargs["reason"]


def test_implementation_goal_loop_is_preserved(tmp_path, monkeypatch):
    verdicts = iter([(False, "the remote head was never verified"), (True, "all criteria hold")])
    contexts: list[str] = []

    def judge(objective, *_args, **_kwargs):
        contexts.append(objective)
        return next(verdicts)

    monkeypatch.setattr(goal, "judge_goal_completion", judge)
    implementation = {
        "status": "completed",
        "summary": "Focused tests passed and the branch was pushed.",
        "changed_files": ["src/a.py"],
        "tests_run": ["pytest focused: passed"],
        "artifacts": [],
        "findings": [],
        "blockers": [],
    }
    lane = _Lane(
        tmp_path,
        monkeypatch,
        card=IMPLEMENTATION_CARD,
        task=_task(assignee="cli-auto", goal_max_turns=3),
        reports=[
            lanes.ProcessResult(0, "turn one", dict(implementation), False),
            lanes.ProcessResult(0, "turn two", dict(implementation), False),
        ],
    )

    execution.execute_claim("titan-iac", "t_card")

    action, kwargs = lane.terminal
    assert action == "complete"
    assert kwargs["metadata"]["goal_turn"] == 2
    assert len(lane.rejections()) == 1
    assert "the remote head was never verified" in lane.rejections()[0]
    assert "prior rejected reports" in contexts[1]


def test_incomplete_implementation_never_finalizes(tmp_path, monkeypatch):
    lane = _Lane(
        tmp_path,
        monkeypatch,
        card=IMPLEMENTATION_CARD,
        task=_task(assignee="cli-auto", goal_max_turns=1),
        reports=[
            lanes.ProcessResult(
                0,
                "turn one",
                {
                    "status": "incomplete",
                    "summary": "The focused suite is still running.",
                    "changed_files": ["src/a.py"],
                    "tests_run": [],
                    "artifacts": [],
                    "findings": [],
                    "blockers": [],
                },
                False,
            )
        ],
    )

    execution.execute_claim("titan-iac", "t_card")

    action, kwargs = lane.terminal
    assert action == "block"
    assert "The focused suite is still running." in kwargs["reason"]