r"""Hermetic module-level tests for the cross-review protocol (issue #11). No WebSocket, no Redis, no model calls — exercises orchestrator/cross_review.py purely. The core scenario: two reviewers give DIFFERENT verdicts on the same artifact, aggregation detects the disagreement and arbitrates, structured evidence + a rework target are produced, and the rework is attributed and classified for the benchmark P_rework input. Run (from the repo root, agent_swarm_v6): pip install -r orchestrator/requirements.txt # cross_review has no extra deps; stdlib only ..\.venv\Scripts\python.exe scripts\test-cross-review.py (The pip step is only needed for a fresh env; cross_review.py itself imports only the stdlib.) """ import sys from pathlib import Path sys.path.insert(0, str(Path(__file__).resolve().parents[1])) from orchestrator.cross_review import ( ReviewDecision, ReworkCategory, aggregate_reviews, build_rework_attributions, classify_rework, review_started_payload, review_decision_made_payload, rework_requested_payload, rework_completed_payload, ) failures = [] def check(name, cond): print(("PASS" if cond else "FAIL"), "-", name) if not cond: failures.append(name) def test_single_reviewer_rejected(): """A single reviewer is a Supervisor Retry, not a cross-review — must be refused.""" only = ReviewDecision(verdict="pass", reviewer_agent_id="r1") try: aggregate_reviews([only]) check("single reviewer rejected", False) except ValueError: check("single reviewer rejected", True) def test_disagreement_detected_and_arbitrated(): """Two reviewers, DIFFERENT verdicts on the same artifact -> disagreement + arbitration.""" passer = ReviewDecision( verdict="pass", reviewer_agent_id="reviewer-impl", evidence=["add() returns sum for valid input"], summary="implementation looks correct", confidence=0.6, ) failer = ReviewDecision( verdict="fail", reviewer_agent_id="reviewer-test", evidence=["tests assert ValueError but impl raises ZeroDivisionError"], failed_criteria=["error semantics consistent across specialists"], affected_tasks=["swarm-x-testing"], recommended_rework=["swarm-x-testing"], summary="conflicting error semantics between impl and tests", confidence=0.9, ) # Majority of 2 with a 1/1 split -> tie -> safety-biased reject. verdict = aggregate_reviews([passer, failer], method="majority") check("disagreement detected", verdict.disagreement is True) check("split vote counted (1 pass / 1 fail)", verdict.pass_votes == 1 and verdict.fail_votes == 1) check("majority tie arbitrates to reject", verdict.accepted is False) check("rework target produced from failing reviewer", verdict.rework_targets == ["swarm-x-testing"]) check("structured evidence preserved on decisions", any("ZeroDivisionError" in e for d in verdict.decisions for e in d.evidence)) check("arbitration method recorded", verdict.method == "majority") # Weighted arbitration: the failer is more confident/weighty -> still reject. weighted = aggregate_reviews([passer, failer], method="weighted") check("weighted arbitration rejects when fail side heavier", weighted.accepted is False) check("weighted method recorded", weighted.method == "weighted") def test_unanimous_pass_accepts(): a = ReviewDecision(verdict="pass", reviewer_agent_id="r1", summary="ok") b = ReviewDecision(verdict="pass", reviewer_agent_id="r2", summary="ok") verdict = aggregate_reviews([a, b]) check("unanimous pass accepted", verdict.accepted is True) check("unanimous has no disagreement", verdict.disagreement is False) check("accepted verdict has no rework targets", verdict.rework_targets == []) def test_majority_fail_three_reviewers(): a = ReviewDecision(verdict="fail", reviewer_agent_id="r1", recommended_rework=["t-impl"], summary="impl wrong") b = ReviewDecision(verdict="fail", reviewer_agent_id="r2", recommended_rework=["t-impl"], summary="impl wrong") c = ReviewDecision(verdict="pass", reviewer_agent_id="r3", summary="fine") verdict = aggregate_reviews([a, b, c]) check("3-reviewer majority fail rejects", verdict.accepted is False) check("majority disagreement detected", verdict.disagreement is True) check("union of rework targets de-duped", verdict.rework_targets == ["t-impl"]) def test_rework_attribution_and_classification(): """Disagreement-driven rework is attributed to a source task/agent and classified.""" failer = ReviewDecision( verdict="fail", reviewer_agent_id="reviewer-test", failed_criteria=["test framework consistency"], recommended_rework=["swarm-x-testing"], summary="docs say unittest but tests use pytest — inconsistent across specialists", confidence=0.8, ) passer = ReviewDecision(verdict="pass", reviewer_agent_id="reviewer-impl", summary="impl ok") verdict = aggregate_reviews([failer, passer], method="majority") owners = {"swarm-x-testing": "agent-tester-7"} attributions = build_rework_attributions(verdict, task_owner=owners) check("one attribution per rework target", len(attributions) == 1) att = attributions[0] check("attribution targets the reopened task", att.target_task_id == "swarm-x-testing") check("attribution records introducing agent", att.introduced_by_agent_id == "agent-tester-7") check("attribution records detecting reviewer", att.detected_by_agent_id == "reviewer-test") check("attribution carries a reason", bool(att.rework_reason)) # Disagreement + framework/consistency wording -> COLLABORATION root cause. check("rework classified as collaboration", att.root_cause == ReworkCategory.COLLABORATION) # Direct classifier checks across categories. check("impl defect classified", classify_rework("implementation logic bug") == ReworkCategory.IMPLEMENTATION) check("test defect classified", classify_rework("pytest assertion wrong") == ReworkCategory.TEST) check("doc drift classified", classify_rework("readme documentation outdated") == ReworkCategory.DOC) check("requirement miss classified", classify_rework("misunderstood the objective scope") == ReworkCategory.REQUIREMENT) check("no signal -> unknown (not fabricated)", classify_rework("") == ReworkCategory.UNKNOWN) check("empty reason + disagreement -> collaboration", classify_rework("", disagreement=True) == ReworkCategory.COLLABORATION) def test_event_payload_builders(): failer = ReviewDecision(verdict="fail", reviewer_agent_id="r-test", recommended_rework=["t1"], summary="bad") passer = ReviewDecision(verdict="pass", reviewer_agent_id="r-impl", summary="ok") verdict = aggregate_reviews([failer, passer]) att = build_rework_attributions(verdict, task_owner={"t1": "agent-1"})[0] started = review_started_payload("swarm-x", reviewer_agent_ids=["r-test", "r-impl"], artifact_task_ids=["t1"], cycle=1) check("review.started carries reviewer count", started["reviewer_count"] == 2) check("review.started marks cross_review kind", started["review_kind"] == "cross_review") decided = review_decision_made_payload("swarm-x", verdict, cycle=1) check("review.decision_made carries disagreement flag", decided["disagreement"] is True) check("review.decision_made carries rework targets", decided["rework_targets"] == ["t1"]) check("review.decision_made embeds per-reviewer reviews", len(decided["reviews"]) == 2) requested = rework_requested_payload("swarm-x", att, cycle=1) check("rework.requested carries task_id", requested["task_id"] == "t1") check("rework.requested carries root_cause", "root_cause" in requested) check("rework.requested carries introducing agent", requested["introduced_by_agent_id"] == "agent-1") completed = rework_completed_payload("swarm-x", att, cycle=1, succeeded=True) check("rework.completed marks status", completed["status"] == "completed") check("rework.completed carries task_id", completed["task_id"] == "t1") def test_invalid_verdict_rejected(): try: ReviewDecision(verdict="maybe", reviewer_agent_id="r1") check("invalid verdict rejected", False) except ValueError: check("invalid verdict rejected", True) try: ReviewDecision(verdict="pass", reviewer_agent_id="") check("missing reviewer id rejected", False) except ValueError: check("missing reviewer id rejected", True) def main(): test_single_reviewer_rejected() test_disagreement_detected_and_arbitrated() test_unanimous_pass_accepts() test_majority_fail_three_reviewers() test_rework_attribution_and_classification() test_event_payload_builders() test_invalid_verdict_rejected() print() if failures: print(f"{len(failures)} check(s) FAILED: {failures}") sys.exit(1) print("all cross-review checks passed") if __name__ == "__main__": main()