"""Wave 6.5 Brain-directed clarification: planning unit + runner integration.""" import backend.agents.brain as brain_mod import backend.agents.runner as runner_mod from backend import config from backend.agents.base import AgentResult from backend.agents.base import AgentUsage from backend.agents.brain import BrainAgent from backend.agents.runner import run_agent_pipeline # --- plan_clarifications unit tests --------------------------------------- def _finding(issue_id, **kw): base = {"issue_id": issue_id, "severity": "medium", "confidence": "low", "source_stage": "conflict", "description": "d", "evidence": [{"sheet": "A1", "source_text": "x"}]} base.update(kw) return base def test_plan_caps_and_filters_unknown_ids(monkeypatch): monkeypatch.setattr(config, "BRAIN_CLARIFY_MAX_REQUESTS", 2) monkeypatch.setattr(brain_mod, "call_json", lambda **k: {"requests": [ {"issue_id": "A", "request_type": "verify_evidence", "reason": "thin"}, {"issue_id": "GHOST", "request_type": "verify_evidence", "reason": "x"}, {"issue_id": "B", "request_type": "verify_evidence", "reason": "amb"}, {"issue_id": "C", "request_type": "verify_evidence", "reason": "over cap"}, ]}) prioritized = [_finding("A"), _finding("B"), _finding("C")] reqs = BrainAgent(AgentUsage()).plan_clarifications(prioritized) ids = [r["issue_id"] for r in reqs] assert ids == ["A", "B"] # GHOST filtered, capped at 2 def test_plan_skips_already_verified(monkeypatch): monkeypatch.setattr(config, "BRAIN_CLARIFY_MAX_REQUESTS", 8) captured = {} def fake(**kwargs): captured["user_text"] = kwargs["user_text"] return {"requests": [ {"issue_id": "A", "request_type": "verify_evidence", "reason": "y"}, ]} monkeypatch.setattr(brain_mod, "call_json", fake) prioritized = [ _finding("A"), _finding("V", verification={"status": "confirmed", "verdicts": []}), ] reqs = BrainAgent(AgentUsage()).plan_clarifications(prioritized) assert [r["issue_id"] for r in reqs] == ["A"] # The already-verified finding must not even be offered to the model. assert '"V"' not in captured["user_text"] def test_plan_empty_on_call_failure(monkeypatch): monkeypatch.setattr(config, "BRAIN_CLARIFY_MAX_REQUESTS", 8) def boom(**k): raise RuntimeError("brain down") monkeypatch.setattr(brain_mod, "call_json", boom) assert BrainAgent(AgentUsage()).plan_clarifications([_finding("A")]) == [] def test_plan_no_requests_returns_empty(monkeypatch): monkeypatch.setattr(config, "BRAIN_CLARIFY_MAX_REQUESTS", 8) monkeypatch.setattr(brain_mod, "call_json", lambda **k: {"requests": []}) assert BrainAgent(AgentUsage()).plan_clarifications([_finding("A")]) == [] # --- runner-level integration --------------------------------------------- def _stub_agent(artifacts): return lambda usage: type("S", (), { "name": "stub", "run": lambda self, scope: AgentResult( scope_id=scope.scope_id, artifacts=list(artifacts)), })() def _patch_pipeline(monkeypatch, brain_finding, plan_requests): monkeypatch.setattr( runner_mod, "convert_pdf_to_images", lambda path: [{"page_number": 1, "base64": "QUJD"}]) monkeypatch.setattr(runner_mod, "SheetExtractorAgent", _stub_agent([ {"sheet_number": "S401", "page_number": 1, "level": "roof", "discipline": "S", "assertions": [ {"text": "(2) 2x6 STUD PACK", "object_type": "framing"}, {"text": "HSS16X4 beam", "object_type": "framing"}, ]}, ])) monkeypatch.setattr(runner_mod, "SheetIndexAgent", _stub_agent([{}])) monkeypatch.setattr(runner_mod, "JurisdictionAgent", _stub_agent([{}])) monkeypatch.setattr(runner_mod, "LinkerAgent", _stub_agent([ {"key": "c1", "location": "roof beam pocket", "assertions": []}, ])) # A filler conflict finding so memory["findings"] is non-empty and wave 6 # actually invokes Brain.run (which our stub replaces with brain_finding). monkeypatch.setattr(runner_mod, "ConflictCriticAgent", _stub_agent([ {"issue_id": "FILLER", "severity": "low", "confidence": "low", "source_stage": "conflict", "sheets": [], "description": "filler", "evidence": []}, ])) monkeypatch.setattr(runner_mod, "CodeAgent", _stub_agent([])) monkeypatch.setattr(runner_mod, "ConstructabilityAgent", _stub_agent([])) monkeypatch.setattr(runner_mod, "CompletenessAgent", _stub_agent([])) monkeypatch.setattr(runner_mod, "DrawingIntegrityAgent", _stub_agent([])) # Brain.run returns our finding; plan_clarifications returns the requests. monkeypatch.setattr( runner_mod, "BrainAgent", lambda usage: type("B", (), { "run": lambda self, findings, si, ju: ([dict(brain_finding)], []), "plan_clarifications": lambda self, prioritized: list(plan_requests), })()) def test_brain_clarify_refutes_and_suppresses(monkeypatch, tmp_path): """Brain flags a MEDIUM finding wave-5b's severity gate skipped; the clarification verifier refutes it, so it moves to suppressed_issues.""" monkeypatch.setattr(config, "ENABLE_BRAIN_CLARIFY", True) finding = { "issue_id": "M1", "severity": "medium", "confidence": "low", "source_stage": "conflict", "sheets": ["S401"], "description": "beam bears on (2) 2x6 stud pack", "evidence": [{"sheet": "S401", "source_text": "(2) 2x6 STUD PACK"}], } _patch_pipeline(monkeypatch, finding, [ {"issue_id": "M1", "request_type": "verify_evidence", "reason": "misread?"}, ]) # The clarification verifier returns a 'corrected' verdict -> refuted. monkeypatch.setattr( "backend.agents.verifier.call_json", lambda **kwargs: {"verdicts": [ {"sheet": "S401", "source_text": "(2) 2x6 STUD PACK", "verdict": "corrected", "actual_text": "(5) 2x6 STUD PACK", "notes": "reads (5)"}, ]}) pdf = tmp_path / "d.pdf" pdf.write_bytes(b"%PDF-1.4\n") report = run_agent_pipeline(str(pdf), out_dir=str(tmp_path), require_review=False) validated = report.get("validated_issues") or [] assert all(f.get("issue_id") != "M1" for f in validated) # dropped assert [f["issue_id"] for f in report["suppressed_issues"]] == ["M1"] assert report["suppressed_issues"][0]["verification"]["status"] == "refuted" def test_brain_clarify_confirms_keeps_finding(monkeypatch, tmp_path): monkeypatch.setattr(config, "ENABLE_BRAIN_CLARIFY", True) finding = { "issue_id": "M2", "severity": "medium", "confidence": "low", "source_stage": "conflict", "sheets": ["S401"], "description": "beam bears on (5) 2x6 stud pack", "evidence": [{"sheet": "S401", "source_text": "(5) 2x6 STUD PACK"}], } _patch_pipeline(monkeypatch, finding, [ {"issue_id": "M2", "request_type": "verify_evidence", "reason": "check"}, ]) monkeypatch.setattr( "backend.agents.verifier.call_json", lambda **kwargs: {"verdicts": [ {"sheet": "S401", "source_text": "(5) 2x6 STUD PACK", "verdict": "confirmed", "actual_text": None, "notes": None}, ]}) pdf = tmp_path / "d.pdf" pdf.write_bytes(b"%PDF-1.4\n") report = run_agent_pipeline(str(pdf), out_dir=str(tmp_path), require_review=False) validated = report.get("validated_issues") or [] kept = [f for f in validated if f.get("issue_id") == "M2"] assert len(kept) == 1 assert kept[0]["verification"]["status"] == "confirmed" assert report["suppressed_issues"] == [] def test_brain_clarify_disabled_is_noop(monkeypatch, tmp_path): monkeypatch.setattr(config, "ENABLE_BRAIN_CLARIFY", False) finding = { "issue_id": "M3", "severity": "medium", "confidence": "low", "source_stage": "conflict", "sheets": ["S401"], "description": "d", "evidence": [{"sheet": "S401", "source_text": "t"}], } # plan_clarifications should never be consulted; give it a bomb to prove it. def _bomb(self, prioritized): raise AssertionError("plan_clarifications must not run when disabled") _patch_pipeline(monkeypatch, finding, []) monkeypatch.setattr( runner_mod, "BrainAgent", lambda usage: type("B", (), { "run": lambda self, findings, si, ju: ([dict(finding)], []), "plan_clarifications": _bomb})()) pdf = tmp_path / "d.pdf" pdf.write_bytes(b"%PDF-1.4\n") report = run_agent_pipeline(str(pdf), out_dir=str(tmp_path), require_review=False) validated = report.get("validated_issues") or [] assert any(f.get("issue_id") == "M3" for f in validated)