"""Non-LLM tests for the review finalizer: decisions, reruns, final artifacts.""" import json import os import pytest from backend.agents.base import AgentResult from backend.review.finalizer import ( apply_decisions, finalize_review, rerun_clarified_scopes, ) from backend.review.store import ReviewStore def test_reject_suppresses_with_reason(): prioritized = [{"issue_id": "AGENT-0001", "severity": "high"}] decisions = {"finding:AGENT-0001": {"decision": "reject", "reason_code": "duplicate"}} kept, suppressed = apply_decisions(prioritized, decisions) assert kept == [] assert suppressed[0]["review_state"] == "rejected" assert suppressed[0]["reason_code"] == "duplicate" def test_unsure_is_kept_but_flagged(): prioritized = [{"issue_id": "AGENT-0002", "severity": "medium"}] decisions = {"finding:AGENT-0002": {"decision": "unsure"}} kept, suppressed = apply_decisions(prioritized, decisions) assert kept[0]["review_state"] == "unsure" assert suppressed == [] def _write_job(out_dir, prioritized, queue, decisions=None, memory=None): """Hand-written review-mode artifacts (conflicts.json + agent/memory.json).""" os.makedirs(os.path.join(out_dir, "agent"), exist_ok=True) report = { "source": "set.pdf", "generated_at": "2026-07-28T00:00:00+00:00", "summary": { "sheets_analyzed": 0, "disciplines": [], "assertions_extracted": 0, "clusters_checked": 0, "conflicts_found": 0, "by_severity": {"high": 0, "medium": 0, "low": 0}, "by_category": {}, "pipeline_mode": "agent", "agent_status": "needs_review", "review": {"required": 1, "completed": 0, "remaining": 1, "total": 1}, "by_stage": {"validated": len(prioritized), "rfis": 0}, }, "conflicts": [], "sheets": [], "validated_issues": prioritized, "suppressed_issues": [], "rfis": [], } with open(os.path.join(out_dir, "conflicts.json"), "w", encoding="utf-8") as f: json.dump(report, f) with open(os.path.join(out_dir, "agent", "memory.json"), "w", encoding="utf-8") as f: json.dump(memory or {}, f) store = ReviewStore(out_dir) store.write_queue(queue) for decision in decisions or []: store.append_decision(decision) def _blocking_item(issue_id): return {"review_item_id": f"finding:{issue_id}", "kind": "finding", "blocking": True, "reasons": ["high_severity"], "payload": {}} def test_finalize_confirm_keeps_confirmed(monkeypatch, tmp_path): monkeypatch.setattr("backend.review.finalizer._draft_rfis", lambda kept: []) _write_job( str(tmp_path), prioritized=[{"issue_id": "AGENT-0001", "severity": "high"}], queue=[_blocking_item("AGENT-0001")], decisions=[{"review_item_id": "finding:AGENT-0001", "decision": "confirm"}], ) report = finalize_review("job1", str(tmp_path)) assert report["validated_issues"][0]["review_state"] == "confirmed" assert report["suppressed_issues"] == [] assert report["summary"]["agent_status"] == "complete" def test_finalize_no_decision_keeps_unreviewed(monkeypatch, tmp_path): """Non-blocking (audit) items don't need a decision; issue stays unreviewed.""" monkeypatch.setattr("backend.review.finalizer._draft_rfis", lambda kept: []) item = {**_blocking_item("AGENT-0001"), "blocking": False, "kind": "audit_finding"} _write_job( str(tmp_path), prioritized=[{"issue_id": "AGENT-0001", "severity": "medium"}], queue=[item], ) report = finalize_review("job1", str(tmp_path)) assert report["validated_issues"][0]["review_state"] == "unreviewed" def test_finalize_clarification_replacement_marked_clarified(monkeypatch, tmp_path): monkeypatch.setattr("backend.review.finalizer._draft_rfis", lambda kept: []) replacement = {"issue_id": "AGENT-0001-R1", "severity": "medium", "clarification_of": "AGENT-0001"} monkeypatch.setattr( "backend.review.finalizer.rerun_clarified_scopes", lambda snapshot, decisions, prioritized=None: [replacement], ) _write_job( str(tmp_path), prioritized=[{"issue_id": "AGENT-0001", "severity": "high"}], queue=[_blocking_item("AGENT-0001")], decisions=[{"review_item_id": "finding:AGENT-0001", "decision": "needs_clarification", "clarification_answer": "Ceiling is 9'-0\" AFF."}], ) report = finalize_review("job1", str(tmp_path)) kept = report["validated_issues"] assert [issue["issue_id"] for issue in kept] == ["AGENT-0001-R1"] assert kept[0]["review_state"] == "clarified" def test_finalize_failed_clarification_flagged(monkeypatch, tmp_path): monkeypatch.setattr("backend.review.finalizer._draft_rfis", lambda kept: []) monkeypatch.setattr( "backend.review.finalizer.rerun_clarified_scopes", lambda snapshot, decisions, prioritized=None: [], ) _write_job( str(tmp_path), prioritized=[{"issue_id": "AGENT-0001", "severity": "high"}], queue=[_blocking_item("AGENT-0001")], decisions=[{"review_item_id": "finding:AGENT-0001", "decision": "needs_clarification", "clarification_answer": "Ceiling is 9'-0\" AFF."}], ) report = finalize_review("job1", str(tmp_path)) assert report["validated_issues"][0]["review_state"] == "clarification_failed" def test_finalize_incomplete_review_raises(monkeypatch, tmp_path): monkeypatch.setattr("backend.review.finalizer._draft_rfis", lambda kept: []) _write_job( str(tmp_path), prioritized=[{"issue_id": "AGENT-0001", "severity": "high"}], queue=[_blocking_item("AGENT-0001")], ) with pytest.raises(ValueError, match="incomplete review"): finalize_review("job1", str(tmp_path)) def test_finalize_writes_final_artifacts(monkeypatch, tmp_path): monkeypatch.setattr("backend.review.finalizer._draft_rfis", lambda kept: [{"issue_id": kept[0]["issue_id"], "question": "?"}]) _write_job( str(tmp_path), prioritized=[{"issue_id": "AGENT-0001", "severity": "high"}], queue=[_blocking_item("AGENT-0001")], decisions=[{"review_item_id": "finding:AGENT-0001", "decision": "confirm"}], ) report = finalize_review("job1", str(tmp_path)) assert report["summary"]["by_stage"]["validated"] == 1 assert report["summary"]["by_stage"]["rfis"] == 1 for name in ("conflicts.json", "validated_issues.json", "suppressed_issues.json", "rfis.json", "report.md"): assert os.path.isfile(os.path.join(str(tmp_path), name)), name with open(os.path.join(str(tmp_path), "validated_issues.json"), encoding="utf-8") as f: assert json.load(f)[0]["review_state"] == "confirmed" def test_finalize_reject_rebuilds_conflicts_and_counts(monkeypatch, tmp_path): """Rejected conflict-stage findings must not survive into the final report's conflicts / headline counts; suppressed_issues keeps them.""" monkeypatch.setattr("backend.review.finalizer._draft_rfis", lambda kept: []) kept_finding = { "issue_id": "AGENT-0001", "source_stage": "conflict", "category": "note_or_spec_contradiction", "severity": "high", "location": "Grid A", "disciplines": ["A", "S"], "sheets": ["A-1"], "description": "kept finding", "evidence": [], "recommended_resolution": "fix", "confidence": "high", } rejected_finding = { **kept_finding, "issue_id": "AGENT-0002", "severity": "medium", "description": "rejected finding", } _write_job( str(tmp_path), prioritized=[kept_finding, rejected_finding], queue=[_blocking_item("AGENT-0001"), _blocking_item("AGENT-0002")], decisions=[ {"review_item_id": "finding:AGENT-0001", "decision": "confirm"}, {"review_item_id": "finding:AGENT-0002", "decision": "reject", "reason_code": "not_a_contradiction"}, ], ) # Simulate the pre-review candidate values the finalizer must overwrite. candidate_path = os.path.join(str(tmp_path), "conflicts.json") with open(candidate_path, encoding="utf-8") as f: candidate = json.load(f) candidate["conflicts"] = [{"description": "kept finding", "severity": "high", "category": "note_or_spec_contradiction"}, {"description": "rejected finding", "severity": "medium", "category": "note_or_spec_contradiction"}] candidate["summary"]["conflicts_found"] = 2 candidate["summary"]["by_severity"] = {"high": 1, "medium": 1, "low": 0} candidate["summary"]["by_category"] = {"note_or_spec_contradiction": 2} with open(candidate_path, "w", encoding="utf-8") as f: json.dump(candidate, f) report = finalize_review("job1", str(tmp_path)) assert [c["description"] for c in report["conflicts"]] == ["kept finding"] assert report["summary"]["conflicts_found"] == 1 assert report["summary"]["by_severity"] == {"high": 1, "medium": 0, "low": 0} assert report["summary"]["by_category"] == {"note_or_spec_contradiction": 1} suppressed = report["suppressed_issues"] assert [s["issue_id"] for s in suppressed] == ["AGENT-0002"] assert suppressed[0]["review_state"] == "rejected" assert suppressed[0]["reason_code"] == "not_a_contradiction" with open(os.path.join(str(tmp_path), "report.md"), encoding="utf-8") as f: assert "rejected finding" not in f.read() def test_rerun_missing_cluster_degrades_to_analysis_gap(): snapshot = {"findings": [{"issue_id": "AGENT-0001", "scope_id": "conflict:link:1"}], "clusters": []} decisions = {"finding:AGENT-0001": { "decision": "needs_clarification", "clarification_answer": "9'-0\" AFF"}} findings = rerun_clarified_scopes(snapshot, decisions) assert len(findings) == 1 assert findings[0]["category"] == "analysis_gap" assert findings[0]["source_stage"] == "qaqc" assert findings[0]["severity"] == "low" assert findings[0]["confidence"] == "high" def test_rerun_non_conflict_scope_noted_as_analysis_gap(): """v1 only reruns conflict scopes; other scopes get a visible gap, no raise.""" snapshot = {"findings": [{"issue_id": "AGENT-0002", "scope_id": "code: egress"}], "clusters": []} decisions = {"finding:AGENT-0002": { "decision": "needs_clarification", "clarification_answer": "Corridor is 44 in."}} findings = rerun_clarified_scopes(snapshot, decisions) assert len(findings) == 1 assert findings[0]["category"] == "analysis_gap" def test_rerun_successful_scope_prepends_clarification_and_tags(monkeypatch): """Real rerun path (non-LLM): cluster lookup, pseudo-assertion injection, and clarification_of tagging through the real rerun_clarified_scopes.""" captured = {} class FakeCritic: name = "conflict_critic" def __init__(self, usage): pass def run(self, scope): captured["scope"] = scope return AgentResult( scope_id=scope.scope_id, artifacts=[{"issue_id": "AGENT-0001-R1", "severity": "medium"}], ) monkeypatch.setattr("backend.review.finalizer.ConflictCriticAgent", FakeCritic) snapshot = { "findings": [{"issue_id": "AGENT-0001", "scope_id": "conflict:link:1"}], "clusters": [{"key": "link:1", "assertions": [{"attribute": "height", "value": "10'-0\""}]}], } decisions = {"finding:AGENT-0001": { "decision": "needs_clarification", "clarification_answer": "9'-0\" AFF"}} findings = rerun_clarified_scopes(snapshot, decisions) assert len(findings) == 1 assert findings[0]["issue_id"] == "AGENT-0001-R1" assert findings[0]["clarification_of"] == "AGENT-0001" payload = captured["scope"].payload assert payload["page_to_b64"] == {} assertions = payload["cluster"]["assertions"] # Prepended at index 0 so front-truncation can't drop the clarification. assert assertions[0]["discipline"] == "Reviewer" assert assertions[0]["attribute"] == "clarification" assert assertions[0]["value"] == "9'-0\" AFF" assert assertions[1]["attribute"] == "height" def test_rerun_ignores_other_decisions(): decisions = {"finding:AGENT-0001": {"decision": "confirm"}} assert rerun_clarified_scopes({}, decisions) == []