orchestrator/agent-team/tests/test_verifier.py

"""Unit tests for agent_team.nodes.verifier — the VERIFY node (§3.3, §3.3.2)."""

from __future__ import annotations

import pytest

from agent_team.ci_gate import GateDecision, GateResult
from agent_team.nodes import verifier as verifier_mod
from agent_team.nodes.verifier import (
    DEFAULT_MAX_BUILD_LOOPS,
    VerifierConfig,
    set_fix_advisor,
    verifier_node,
)
from agent_team.state_store import compute_content_hash
from agent_team.task_model import Phase, PipelineState, TaskStatus


def _diff_for(*paths: str) -> str:
    chunks = []
    for p in paths:
        chunks.append(f"diff --git a/{p} b/{p}\n@@ -1 +1 @@\n-old\n+new\n")
    return "".join(chunks)


def _hash(diff: str) -> str:
    return compute_content_hash(diff.encode("utf-8"))


def _state(diff: str, ci: dict | None) -> PipelineState:
    return {
        "thread_id": "t1",
        "status": TaskStatus.ACTIVE.value,
        "current_phase": Phase.VERIFY.value,
        "candidate_diff": diff,
        "diff_hash": _hash(diff),
        "ci_results": ci,
    }


@pytest.fixture(autouse=True)
def _reset_advisor():
    """Restore the default null advisor after each test."""
    yield
    set_fix_advisor(verifier_mod._null_advisor)


# --------------------------------------------------------------------------- #
# PASS path
# --------------------------------------------------------------------------- #


def test_pass_advances_to_done() -> None:
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
    out = verifier_node(_state(diff, ci), VerifierConfig(expected_run_id="r1"))
    assert out["status"] == TaskStatus.DONE.value
    assert out["current_phase"] == Phase.DONE.value
    assert out["ci_results"]["gate_decision"] == "pass"


def test_pass_does_not_consult_advisor() -> None:
    calls: list = []

    def advisor(result, state):  # pragma: no cover - asserted not called
        calls.append(result)
        return "hint"

    set_fix_advisor(advisor)
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
    verifier_node(_state(diff, ci), VerifierConfig(expected_run_id="r1"))
    assert calls == []  # the LLM is never asked whether it passed


# --------------------------------------------------------------------------- #
# FAIL path — loop back to BUILD
# --------------------------------------------------------------------------- #


def test_fail_loops_back_to_build() -> None:
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
    out = verifier_node(
        _state(diff, ci), VerifierConfig(expected_run_id="r1", build_loops=0)
    )
    assert out["status"] == TaskStatus.ACTIVE.value
    assert out["current_phase"] == Phase.BUILD.value


def test_fail_consults_advisor_for_hint() -> None:
    def advisor(result: GateResult, state) -> str:
        assert result.decision is GateDecision.FAIL
        return "bump the pinned version"

    set_fix_advisor(advisor)
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
    out = verifier_node(_state(diff, ci), VerifierConfig(expected_run_id="r1"))
    assert out["review_verdicts"][0]["fix_hint"] == "bump the pinned version"


def test_fail_parks_when_build_loops_exhausted() -> None:
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
    cfg = VerifierConfig(
        expected_run_id="r1",
        build_loops=DEFAULT_MAX_BUILD_LOOPS - 1,
        max_build_loops=DEFAULT_MAX_BUILD_LOOPS,
    )
    out = verifier_node(_state(diff, ci), cfg)
    assert out["status"] == TaskStatus.PARKED.value
    assert out["current_phase"] == Phase.PARKED.value
    assert any("max build loops" in r for r in out["review_verdicts"][0]["reasons"])


def test_fail_increments_build_loops_in_verdict() -> None:
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
    out = verifier_node(
        _state(diff, ci), VerifierConfig(expected_run_id="r1", build_loops=1)
    )
    assert out["review_verdicts"][0]["build_loops"] == 2


# --------------------------------------------------------------------------- #
# Durable per-task build-loop budget (LOGIC-RACE-01)
# --------------------------------------------------------------------------- #


def test_fail_reads_loop_count_from_state_not_config() -> None:
    # The count that bounds the loop is DURABLE per-task state. A shared config
    # with build_loops=0 must NOT mask a per-task state["build_loops"] of 2.
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
    state = _state(diff, ci)
    state["build_loops"] = 2
    out = verifier_node(state, VerifierConfig(expected_run_id="r1", build_loops=0))
    # next = state(2) + 1 = 3 == DEFAULT_MAX_BUILD_LOOPS -> park (not a vacuous
    # loop driven by the always-zero shared config).
    assert out["status"] == TaskStatus.PARKED.value
    assert out["review_verdicts"][0]["build_loops"] == 3


def test_fail_writes_incremented_loop_count_into_state() -> None:
    # On a recoverable FAIL the node persists the incremented count into the
    # returned partial state so the NEXT VERIFY (after BUILD->DISPATCH) sees it.
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
    state = _state(diff, ci)
    state["build_loops"] = 0
    out = verifier_node(state, VerifierConfig(expected_run_id="r1"))
    assert out["current_phase"] == Phase.BUILD.value
    assert out["build_loops"] == 1


def test_repeated_fail_parks_after_exactly_max_build_loops_via_state() -> None:
    # Simulate the durable BUILD->DISPATCH->VERIFY loop: the incremented
    # build_loops the node returns is fed back into the next call's state (the
    # checkpointer carries it). A perpetually-FAILing task must reach PARKED
    # after EXACTLY max_build_loops iterations, never loop unbounded.
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "failure", "diff_hash": _hash(diff)}
    # A single SHARED config (build_loops stays 0) — exactly the wiring-time
    # reality that made the budget dead before the fix.
    shared_cfg = VerifierConfig(expected_run_id="r1", max_build_loops=3, build_loops=0)

    state = _state(diff, ci)
    state["build_loops"] = 0

    statuses: list[str] = []
    for _ in range(10):  # bound the harness so a regression can't hang the test
        out = verifier_node(state, shared_cfg)
        statuses.append(out["status"])
        if out["status"] == TaskStatus.PARKED.value:
            break
        # Carry the durable count forward, as the checkpointer would.
        state["build_loops"] = out["build_loops"]

    # FAILs at counts 1, 2 loop back to BUILD; the 3rd (next==3==max) parks.
    assert statuses == [
        TaskStatus.ACTIVE.value,
        TaskStatus.ACTIVE.value,
        TaskStatus.PARKED.value,
    ]
    assert out["current_phase"] == Phase.PARKED.value


# --------------------------------------------------------------------------- #
# BLOCK path — park for human + GPT cross-review
# --------------------------------------------------------------------------- #


def test_block_on_denylist_parks() -> None:
    diff = _diff_for(".github/workflows/ci.yml")
    ci = {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
    out = verifier_node(_state(diff, ci), VerifierConfig(expected_run_id="r1"))
    assert out["status"] == TaskStatus.PARKED.value
    assert out["current_phase"] == Phase.PARKED.value
    assert out["ci_results"]["gate_decision"] == "block"


def test_block_on_hash_mismatch_parks() -> None:
    diff = _diff_for("src/foo.py")
    state = _state(diff, {"run_id": "r1", "conclusion": "success"})
    state["diff_hash"] = "tampered"
    out = verifier_node(state, VerifierConfig(expected_run_id="r1"))
    assert out["status"] == TaskStatus.PARKED.value


def test_block_never_advances_to_done_even_with_advisor() -> None:
    set_fix_advisor(lambda result, state: "irrelevant")
    diff = _diff_for(".github/workflows/ci.yml")
    ci = {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
    out = verifier_node(_state(diff, ci), VerifierConfig(expected_run_id="r1"))
    assert out["status"] != TaskStatus.DONE.value


def test_missing_diff_parks() -> None:
    state: PipelineState = {
        "thread_id": "t1",
        "candidate_diff": None,
        "diff_hash": None,
        "ci_results": None,
    }
    out = verifier_node(state, VerifierConfig(expected_run_id="r1"))
    assert out["status"] == TaskStatus.PARKED.value
    assert any("no candidate_diff" in r for r in out["review_verdicts"][0]["reasons"])


# --------------------------------------------------------------------------- #
# Provenance / partial-update shape
# --------------------------------------------------------------------------- #


def test_verdict_records_provenance() -> None:
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r9", "conclusion": "success", "diff_hash": _hash(diff)}
    out = verifier_node(_state(diff, ci), VerifierConfig(expected_run_id="r9"))
    verdict = out["review_verdicts"][0]
    assert verdict["stage"] == "verify"
    assert verdict["run_id"] == "r9"
    assert verdict["diff_hash"] == _hash(diff)
    assert verdict["ci_conclusion"] == "success"
    assert "at" in verdict


def test_returns_partial_update_only() -> None:
    diff = _diff_for("src/foo.py")
    ci = {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
    out = verifier_node(_state(diff, ci), VerifierConfig(expected_run_id="r1"))
    # Node returns only the keys it writes (LangGraph reducer merges the rest).
    assert set(out) == {
        "status",
        "current_phase",
        "review_verdicts",
        "ci_results",
        "updated_at",
    }


def test_allowed_scope_threaded_to_gate() -> None:
    diff = _diff_for("src/foo.py", "elsewhere/bar.py")
    ci = {"run_id": "r1", "conclusion": "success", "diff_hash": _hash(diff)}
    cfg = VerifierConfig(expected_run_id="r1", allowed_scope=["src/"])
    out = verifier_node(_state(diff, ci), cfg)
    assert out["status"] == TaskStatus.PARKED.value  # out-of-scope -> BLOCK -> park


# --------------------------------------------------------------------------- #
# Per-task run-id binding (design §4 Decision 4)
# --------------------------------------------------------------------------- #


def test_state_run_id_overrides_static_config_run_id() -> None:
    # The gate must bind to the run id THIS task dispatched (state["run_id"]),
    # not the static config constant. A CI conclusion keyed to the per-task
    # run id passes even though config carries a different (stale) run id.
    diff = _diff_for("src/foo.py")
    state = _state(
        diff, {"run_id": "task-run", "conclusion": "success", "diff_hash": _hash(diff)}
    )
    state["run_id"] = "task-run"
    # config.expected_run_id is a DIFFERENT, stale value — state must win.
    out = verifier_node(state, VerifierConfig(expected_run_id="stale-wiring-run"))
    assert out["status"] == TaskStatus.DONE.value
    assert out["review_verdicts"][0]["run_id"] == "task-run"


def test_substituted_run_id_is_rejected() -> None:
    # Anti-substitution: a CI result whose run_id != state["run_id"] is a BLOCK,
    # even with a success conclusion (someone tried to graft a passing run from
    # another task onto this one).
    diff = _diff_for("src/foo.py")
    state = _state(
        diff,
        {"run_id": "other-task-run", "conclusion": "success", "diff_hash": _hash(diff)},
    )
    state["run_id"] = "my-task-run"
    out = verifier_node(state, VerifierConfig(expected_run_id=None))
    assert out["status"] == TaskStatus.PARKED.value
    assert out["ci_results"]["gate_decision"] == "block"
    assert any("run-id mismatch" in r for r in out["review_verdicts"][0]["reasons"])


def test_two_concurrent_tasks_each_gate_against_own_run_id() -> None:
    # Two tasks share ONE VerifierConfig but each gates against its OWN
    # state["run_id"]. Task A's CI matches A's run id (PASS); task B's CI is
    # keyed to A's run id (substitution) so B BLOCKs.
    shared_cfg = VerifierConfig(expected_run_id=None)

    diff_a = _diff_for("src/a.py")
    state_a = _state(
        diff_a, {"run_id": "run-A", "conclusion": "success", "diff_hash": _hash(diff_a)}
    )
    state_a["run_id"] = "run-A"

    diff_b = _diff_for("src/b.py")
    # B's fetched CI is wrongly keyed to run-A (a leaked/substituted run).
    state_b = _state(
        diff_b, {"run_id": "run-A", "conclusion": "success", "diff_hash": _hash(diff_b)}
    )
    state_b["run_id"] = "run-B"

    out_a = verifier_node(state_a, shared_cfg)
    out_b = verifier_node(state_b, shared_cfg)

    assert out_a["status"] == TaskStatus.DONE.value
    assert out_a["review_verdicts"][0]["run_id"] == "run-A"
    assert out_b["status"] == TaskStatus.PARKED.value
    assert out_b["review_verdicts"][0]["run_id"] == "run-B"


def test_none_run_id_at_gate_time_blocks_never_passes() -> None:
    # No per-task run_id in state AND no config fallback -> BLOCK (park), never a
    # vacuous pass, even when CI reports success.
    diff = _diff_for("src/foo.py")
    state = _state(
        diff, {"run_id": "", "conclusion": "success", "diff_hash": _hash(diff)}
    )
    # state has no "run_id" key; config fallback is None.
    out = verifier_node(state, VerifierConfig(expected_run_id=None))
    assert out["status"] == TaskStatus.PARKED.value
    assert out["ci_results"]["gate_decision"] == "block"