orchestrator/agent-team/tests/test_review_loop.py

"""Unit tests for agent_team.nodes.review_loop (design §3.3, §7.1 P2)."""

from __future__ import annotations

from typing import Any

import pytest

from agent_team.nodes import review_loop
from agent_team.nodes.review_loop import (
    DEFAULT_MAX_REVIEW_ROUNDS,
    ReviewOutcome,
    ReviewResult,
    ReviewVerdict,
    build_review_prompt,
    parse_verdict,
    review_node,
    route_after_review,
    set_review_invoker,
)
from agent_team.task_model import Phase, PipelineState, TaskStatus


@pytest.fixture(autouse=True)
def _restore_invoker():
    """Restore the module review invoker after each test."""
    original = review_loop._review_invoker
    yield
    review_loop._review_invoker = original


@pytest.fixture(autouse=True)
def _clear_env(monkeypatch: pytest.MonkeyPatch):
    """Clear review-loop env knobs so tests are hermetic."""
    monkeypatch.delenv("AGENT_TEAM_MAX_REVIEW_ROUNDS", raising=False)
    monkeypatch.delenv("AGENT_TEAM_ORCHESTRATOR_RUN_PY", raising=False)


def _state(**overrides: Any) -> PipelineState:
    """Build a minimal PipelineState with a plan present."""
    base: PipelineState = {
        "thread_id": "t1",
        "plan": {"phases": ["P1", "P2"]},
        "review_verdicts": [],
    }
    base.update(overrides)  # type: ignore[typeddict-item]
    return base


def _invoker_returning(text: str):
    """Return a review invoker that always yields ``text`` and records calls."""
    calls: list[dict[str, Any]] = []

    def invoker(prompt: str, **kw: Any) -> str:
        calls.append({"prompt": prompt, "kw": kw})
        return text

    invoker.calls = calls  # type: ignore[attr-defined]
    return invoker


# --------------------------------------------------------------------------- #
# parse_verdict
# --------------------------------------------------------------------------- #


def test_parse_verdict_approve() -> None:
    assert parse_verdict("VERDICT: APPROVE\nlooks good") is ReviewVerdict.APPROVE
    assert parse_verdict("LGTM, ship it") is ReviewVerdict.APPROVE


def test_parse_verdict_request_changes() -> None:
    assert (
        parse_verdict("VERDICT: REQUEST CHANGES\nmissing rollback")
        is ReviewVerdict.REQUEST_CHANGES
    )
    assert parse_verdict("BLOCK: unsafe IAM policy") is ReviewVerdict.REQUEST_CHANGES


def test_parse_verdict_is_case_insensitive() -> None:
    assert parse_verdict("verdict: approve") is ReviewVerdict.APPROVE


def test_parse_verdict_fails_closed_on_ambiguous() -> None:
    # Neither token present -> REQUEST_CHANGES (fail closed).
    assert parse_verdict("hmm, not sure") is ReviewVerdict.REQUEST_CHANGES
    assert parse_verdict("") is ReviewVerdict.REQUEST_CHANGES
    assert parse_verdict(None) is ReviewVerdict.REQUEST_CHANGES  # type: ignore[arg-type]


def test_parse_verdict_request_changes_wins_on_conflict() -> None:
    # Both tokens present -> REQUEST_CHANGES wins (fail closed).
    text = "Some phases APPROVE-able but VERDICT: REQUEST CHANGES overall"
    assert parse_verdict(text) is ReviewVerdict.REQUEST_CHANGES


def test_parse_verdict_approve_with_no_blockers_prose() -> None:
    # Regression: "no blockers" / "no blocking" prose inside an APPROVE must not
    # trip the BLOCK change-token (substring false-positive). Word-boundary
    # matching keeps these as APPROVE.
    assert parse_verdict("Approved, no blockers.") is ReviewVerdict.APPROVE
    assert (
        parse_verdict("VERDICT: APPROVE — no blocking issues found")
        is ReviewVerdict.APPROVE
    )
    assert parse_verdict("LGTM, found no blockers") is ReviewVerdict.APPROVE


def test_parse_verdict_real_block_token_requests_changes() -> None:
    # A real, standalone BLOCK verdict token (rubric vocabulary) -> REQUEST_CHANGES.
    assert parse_verdict("BLOCK: unsafe IAM policy") is ReviewVerdict.REQUEST_CHANGES
    assert (
        parse_verdict("VERDICT: REQUEST CHANGES\nthis is a BLOCK")
        is ReviewVerdict.REQUEST_CHANGES
    )


def test_parse_verdict_bare_no_blockers_prose_fails_closed() -> None:
    # "no blockers" with NO explicit APPROVE/LGTM token is genuinely ambiguous
    # and must fail closed (the dropped NO BLOCKERS approve token is unreachable).
    # Note these inflected words ("blockers"/"blocking") are NOT change tokens.
    assert parse_verdict("no blockers") is ReviewVerdict.REQUEST_CHANGES
    assert parse_verdict("no blocking issues") is ReviewVerdict.REQUEST_CHANGES


# --------------------------------------------------------------------------- #
# build_review_prompt
# --------------------------------------------------------------------------- #


def test_build_review_prompt_embeds_plan() -> None:
    prompt = build_review_prompt(_state(plan={"phases": ["alpha"]}))
    assert "alpha" in prompt
    assert "VERDICT: APPROVE" in prompt
    assert "VERDICT: REQUEST CHANGES" in prompt


def test_build_review_prompt_includes_prior_findings() -> None:
    state = _state(
        review_verdicts=[{"verdict": "request_changes", "findings": "rollback missing"}]
    )
    prompt = build_review_prompt(state)
    assert "rollback missing" in prompt
    assert "REVISED" in prompt


# --------------------------------------------------------------------------- #
# review_node — APPROVE path
# --------------------------------------------------------------------------- #


def test_review_node_approve_advances_to_build() -> None:
    set_review_invoker(_invoker_returning("VERDICT: APPROVE\nsolid plan"))
    update = review_node(_state())

    assert update["current_phase"] == Phase.BUILD.value
    assert update["status"] == TaskStatus.ACTIVE.value
    assert update["updated_at"]

    verdicts = update["review_verdicts"]
    assert len(verdicts) == 1
    assert verdicts[0]["verdict"] == ReviewVerdict.APPROVE.value
    assert verdicts[0]["outcome"] == ReviewOutcome.APPROVED.value
    assert verdicts[0]["round_index"] == 1
    assert verdicts[0]["reviewer"] == "cross_reviewer"


# --------------------------------------------------------------------------- #
# review_node — REQUEST_CHANGES loop-back vs escalate
# --------------------------------------------------------------------------- #


def test_review_node_request_changes_loops_back_under_cap() -> None:
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES\nfix it"))
    update = review_node(_state(), config={"max_review_rounds": 3})

    assert update["current_phase"] == Phase.PLAN.value
    assert update["status"] == TaskStatus.ACTIVE.value
    assert update["review_verdicts"][-1]["outcome"] == ReviewOutcome.LOOP_BACK.value


def test_review_node_escalates_at_round_cap() -> None:
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES\nstill broken"))
    # Two prior REQUEST_CHANGES rounds already recorded; cap is 3 -> this is
    # round 3 -> escalate.
    state = _state(
        review_verdicts=[
            {"verdict": "request_changes", "outcome": "loop_back"},
            {"verdict": "request_changes", "outcome": "loop_back"},
        ]
    )
    update = review_node(state, config={"max_review_rounds": 3})

    assert update["current_phase"] == Phase.PARKED.value
    assert update["status"] == TaskStatus.PARKED.value
    last = update["review_verdicts"][-1]
    assert last["outcome"] == ReviewOutcome.ESCALATE.value
    assert last["round_index"] == 3


def test_review_node_approve_at_cap_still_advances() -> None:
    # Even at the round cap, an APPROVE advances to build (cap only bounds
    # REQUEST_CHANGES looping).
    set_review_invoker(_invoker_returning("VERDICT: APPROVE"))
    state = _state(
        review_verdicts=[
            {"verdict": "request_changes", "outcome": "loop_back"},
            {"verdict": "request_changes", "outcome": "loop_back"},
        ]
    )
    update = review_node(state, config={"max_review_rounds": 3})
    assert update["current_phase"] == Phase.BUILD.value
    assert update["review_verdicts"][-1]["outcome"] == ReviewOutcome.APPROVED.value


def test_review_node_appends_to_prior_verdicts() -> None:
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES"))
    state = _state(
        review_verdicts=[{"verdict": "request_changes", "outcome": "loop_back"}]
    )
    update = review_node(state, config={"max_review_rounds": 5})
    assert len(update["review_verdicts"]) == 2
    assert update["review_verdicts"][-1]["round_index"] == 2


# --------------------------------------------------------------------------- #
# review_node — human-gate verdicts excluded from the round cap (LOGIC-04)
# --------------------------------------------------------------------------- #


def test_review_round_index_excludes_human_plan_gate_verdicts() -> None:
    # A human request_changes folds a synthetic verdict tagged
    # reviewer == "human_plan_gate" into review_verdicts. It must NOT count as a
    # reviewer round: with one reviewer round + two human-gate verdicts present,
    # the next reviewer round is still round 2 (not round 4).
    state = _state(
        review_verdicts=[
            {"verdict": "request_changes", "outcome": "loop_back"},
            {"verdict": "request_changes", "reviewer": "human_plan_gate"},
            {"verdict": "request_changes", "reviewer": "human_plan_gate"},
        ]
    )
    assert review_loop._review_round_index(state) == 2


def test_review_node_human_gate_verdict_grants_fresh_round() -> None:
    # After a human request_changes (synthetic human_plan_gate verdict), the
    # revised plan's review gets a fresh round rather than immediate escalation,
    # even though len(review_verdicts) would otherwise be at the cap.
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES\nstill rough"))
    state = _state(
        review_verdicts=[
            {"verdict": "request_changes", "outcome": "loop_back"},  # reviewer r1
            {"verdict": "request_changes", "outcome": "loop_back"},  # reviewer r2
            {"verdict": "request_changes", "reviewer": "human_plan_gate"},  # human
        ]
    )
    update = review_node(state, config={"max_review_rounds": 3})

    # Only 2 reviewer rounds counted, so this is reviewer round 3 == cap: it
    # escalates here, NOT one round early because of the human verdict.
    last = update["review_verdicts"][-1]
    assert last["round_index"] == 3
    assert last["outcome"] == ReviewOutcome.ESCALATE.value


def test_review_node_only_human_verdicts_is_fresh_first_round() -> None:
    # With NO prior reviewer verdicts but a human-gate verdict present, the very
    # first reviewer pass is round 1 and loops back under the cap (the bug made
    # this escalate immediately at cap 1).
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES\nredo"))
    state = _state(
        review_verdicts=[
            {"verdict": "request_changes", "reviewer": "human_plan_gate"},
        ]
    )
    update = review_node(state, config={"max_review_rounds": 2})

    assert update["current_phase"] == Phase.PLAN.value
    assert update["status"] == TaskStatus.ACTIVE.value
    assert update["review_verdicts"][-1]["round_index"] == 1


# --------------------------------------------------------------------------- #
# review_node — invoker wiring & errors
# --------------------------------------------------------------------------- #


def test_review_node_passes_prompt_and_config_to_invoker() -> None:
    invoker = _invoker_returning("VERDICT: APPROVE")
    set_review_invoker(invoker)
    cfg = {"max_review_rounds": 2, "orchestrator_run_py": "/tmp/run.py"}
    review_node(_state(plan={"phases": ["zeta"]}), config=cfg)

    assert len(invoker.calls) == 1  # type: ignore[attr-defined]
    call = invoker.calls[0]  # type: ignore[attr-defined]
    assert "zeta" in call["prompt"]
    assert call["kw"]["run_py"] == "/tmp/run.py"
    assert call["kw"]["config"] is cfg


def test_review_node_requires_plan() -> None:
    set_review_invoker(_invoker_returning("VERDICT: APPROVE"))
    with pytest.raises(ValueError):
        review_node({"thread_id": "t1", "review_verdicts": []})


def test_review_node_rejects_bad_round_cap() -> None:
    set_review_invoker(_invoker_returning("VERDICT: APPROVE"))
    with pytest.raises(ValueError):
        review_node(_state(), config={"max_review_rounds": 0})
    with pytest.raises(ValueError):
        review_node(_state(), config={"max_review_rounds": "lots"})


def test_review_node_uses_env_round_cap(monkeypatch: pytest.MonkeyPatch) -> None:
    monkeypatch.setenv("AGENT_TEAM_MAX_REVIEW_ROUNDS", "1")
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES"))
    # Cap 1 from env -> first REQUEST_CHANGES round escalates immediately.
    update = review_node(_state())
    assert update["status"] == TaskStatus.PARKED.value


def test_review_node_default_cap_is_three() -> None:
    assert DEFAULT_MAX_REVIEW_ROUNDS == 3
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES"))
    # Rounds 1 and 2 loop back under the default cap of 3.
    state = _state(
        review_verdicts=[{"verdict": "request_changes", "outcome": "loop_back"}]
    )
    update = review_node(state)  # round 2
    assert update["current_phase"] == Phase.PLAN.value


def test_review_node_coerces_non_string_invoker_output() -> None:
    set_review_invoker(lambda prompt, **kw: 12345)  # type: ignore[return-value]
    # No verdict token in "12345" -> fails closed to REQUEST_CHANGES.
    update = review_node(_state(), config={"max_review_rounds": 3})
    assert (
        update["review_verdicts"][-1]["verdict"] == ReviewVerdict.REQUEST_CHANGES.value
    )


# --------------------------------------------------------------------------- #
# route_after_review
# --------------------------------------------------------------------------- #


def test_route_after_review_approved_to_build() -> None:
    state: PipelineState = {"review_verdicts": [{"outcome": "approved"}]}
    assert route_after_review(state) == "build"


def test_route_after_review_loop_back_to_plan() -> None:
    state: PipelineState = {"review_verdicts": [{"outcome": "loop_back"}]}
    assert route_after_review(state) == "plan"


def test_route_after_review_escalate_to_parked() -> None:
    state: PipelineState = {"review_verdicts": [{"outcome": "escalate"}]}
    assert route_after_review(state) == "parked"


def test_route_after_review_no_verdict_parks_fail_closed() -> None:
    assert route_after_review({"review_verdicts": []}) == "parked"
    assert route_after_review({}) == "parked"


def test_route_after_review_unexpected_outcome_parks() -> None:
    state: PipelineState = {"review_verdicts": [{"outcome": "weird"}]}
    assert route_after_review(state) == "parked"


def test_route_uses_most_recent_verdict() -> None:
    state: PipelineState = {
        "review_verdicts": [{"outcome": "loop_back"}, {"outcome": "approved"}]
    }
    assert route_after_review(state) == "build"


# --------------------------------------------------------------------------- #
# node + router integration (the full loop decision)
# --------------------------------------------------------------------------- #


def test_node_then_route_round_trip_approve() -> None:
    set_review_invoker(_invoker_returning("VERDICT: APPROVE"))
    update = review_node(_state())
    # Merge update back into state (LangGraph reducer would do this).
    state = {**_state(), **update}
    assert route_after_review(state) == "build"


def test_node_then_route_round_trip_loop_back() -> None:
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES"))
    update = review_node(_state(), config={"max_review_rounds": 3})
    state = {**_state(), **update}
    assert route_after_review(state) == "plan"


def test_node_then_route_round_trip_escalate() -> None:
    set_review_invoker(_invoker_returning("VERDICT: REQUEST CHANGES"))
    update = review_node(_state(), config={"max_review_rounds": 1})
    state = {**_state(), **update}
    assert route_after_review(state) == "parked"


# --------------------------------------------------------------------------- #
# ReviewResult serialization
# --------------------------------------------------------------------------- #


def test_review_result_to_dict_round_trips_fields() -> None:
    result = ReviewResult(
        verdict=ReviewVerdict.APPROVE,
        round_index=2,
        outcome=ReviewOutcome.APPROVED,
        findings="all good",
        created_at="2026-06-17T00:00:00+00:00",
    )
    d = result.to_dict()
    assert d == {
        "verdict": "approve",
        "round_index": 2,
        "outcome": "approved",
        "findings": "all good",
        "reviewer": "cross_reviewer",
        "created_at": "2026-06-17T00:00:00+00:00",
    }


# --------------------------------------------------------------------------- #
# default invoker (orchestrator shell-out) — error surface only, no real call
# --------------------------------------------------------------------------- #


def test_default_invoker_missing_run_py_raises() -> None:
    with pytest.raises(FileNotFoundError):
        review_loop._orchestrator_invoker("prompt", run_py="/nonexistent/path/run.py")


def test_default_invoker_passes_timeout_to_subprocess(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # The default shell-out must pass a bounded timeout to subprocess.run.
    captured: dict[str, Any] = {}

    class _Completed:
        returncode = 0
        stdout = "VERDICT: APPROVE"
        stderr = ""

    def _fake_run(args: list[str], **kw: Any) -> _Completed:
        captured["kw"] = kw
        return _Completed()

    monkeypatch.setattr(review_loop.subprocess, "run", _fake_run)
    monkeypatch.setattr(review_loop.os.path, "exists", lambda _p: True)
    out = review_loop._orchestrator_invoker(
        "prompt", run_py="/tmp/run.py", config={"review_timeout_seconds": 12}
    )
    assert "APPROVE" in out
    assert captured["kw"]["timeout"] == 12.0


def test_default_invoker_timeout_fails_closed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # A hung run.py (TimeoutExpired) must fail CLOSED: return text that parses to
    # REQUEST_CHANGES rather than raising and crashing review_node.
    import subprocess as _sp

    def _raise_timeout(args: list[str], **kw: Any):
        raise _sp.TimeoutExpired(cmd=args, timeout=kw.get("timeout", 1))

    monkeypatch.setattr(review_loop.subprocess, "run", _raise_timeout)
    monkeypatch.setattr(review_loop.os.path, "exists", lambda _p: True)
    out = review_loop._orchestrator_invoker("prompt", run_py="/tmp/run.py")
    assert parse_verdict(out) is ReviewVerdict.REQUEST_CHANGES


def test_review_node_survives_timeout_fail_closed(
    monkeypatch: pytest.MonkeyPatch,
) -> None:
    # End-to-end: a hung default invoker makes review_node loop back / escalate,
    # never approve, and never raise.
    import subprocess as _sp

    def _raise_timeout(args: list[str], **kw: Any):
        raise _sp.TimeoutExpired(cmd=args, timeout=kw.get("timeout", 1))

    monkeypatch.setattr(review_loop.subprocess, "run", _raise_timeout)
    monkeypatch.setattr(review_loop.os.path, "exists", lambda _p: True)
    # Use the real default invoker (not a test fake).
    set_review_invoker(review_loop._orchestrator_invoker)
    update = review_node(_state(), config={"max_review_rounds": 3})
    assert (
        update["review_verdicts"][-1]["verdict"] == ReviewVerdict.REQUEST_CHANGES.value
    )
    assert update["current_phase"] == Phase.PLAN.value