fix: wave 5b review blockers - suppressed memory key, finalizer merge, zero-image guard

backend/.env.example

@@ -72,3 +72,9 @@ SMTP_PASSWORD=
 SMTP_FROM=
 SMTP_USE_TLS=true
 SMTP_USE_SSL=false
+
+# Wave 5b evidence verification (vision fact-check of cited sheet text)
+AGENT_VERIFY_MAX_CHECKS=20
+AGENT_VERIFY_SEVERITIES=critical,high
+AGENT_VERIFY_REASONING_EFFORT=low
+VERIFY_MAX_TOKENS=8192

backend/agents/conflict_critic.py

@@ -60,6 +60,8 @@ class ConflictCriticAgent:
                 model=config.AGENT_CONFLICT_MODEL,
                 usage_tracker=self.usage,
                 usage_stage="agent.conflict",
+                reasoning_effort=config.EXTRACT_REASONING_EFFORT or None,
+                reasoning_max_tokens=config.EXTRACT_REASONING_MAX_TOKENS or None,
             )
             candidates = parsed if isinstance(parsed, list) else (
                 parsed.get("conflicts") if isinstance(parsed, dict) else []

backend/agents/construct_agent.py

@@ -47,6 +47,7 @@ class ConstructabilityAgent:
                 "assertions": dumps(cluster["assertions"]),
                 "clusters": dumps(slim_clusters([cluster])),
                 "conflicts": dumps(scope.payload.get("conflicts") or []),
+                "disputes": dumps(cluster.get("disputed_attributes") or []),
             }
             for key, value in substitutions.items():
                 instruction = instruction.replace("{" + key + "}", value)
@@ -64,7 +65,8 @@ class ConstructabilityAgent:
                 lambda item: validate_issue(item, "constructability"),
             )
             for finding in findings:
-                finding.update(agent=self.name, scope_id=scope.scope_id)
+                finding.update(agent=self.name, scope_id=scope.scope_id,
+                               cluster_key=cluster.get("key"))
             return AgentResult(scope_id=scope.scope_id, artifacts=findings)
         except Exception as exc:
             return failure(scope, exc)

backend/agents/disputes.py

@@ -0,0 +1,55 @@
+"""Deterministic detection of contradictory extracted values within a cluster.
+
+Extraction is a vision pass: quantities and sizes can be misread ("(2) 2x6" vs
+"(5) 2x6"). Cluster members are supposed to describe the same real-world
+element, so two members asserting different values for the same attribute are
+a probable misread. Flag these so downstream text-only stages treat the value
+as unverified instead of reasoning from one reading.
+"""
+
+import re
+from typing import Dict, List
+
+
+def _norm(value) -> str:
+    return re.sub(r"\s+", " ", ("" if value is None else str(value)).strip().lower())
+
+
+def find_disputes(assertions: List[Dict]) -> List[Dict]:
+    """Same attribute with >= 2 distinct normalized values = disputed."""
+    groups: Dict[str, Dict[str, Dict]] = {}
+    for assertion in assertions:
+        attribute = _norm(assertion.get("attribute"))
+        value = _norm(assertion.get("value"))
+        if not attribute or not value:
+            continue
+        # Group on the normalized value, but keep the original (whitespace-
+        # collapsed) text so disputes read like the sheet, not a lowercase munge.
+        original = re.sub(r"\s+", " ", str(assertion.get("value")).strip())
+        bucket = groups.setdefault(attribute, {}).setdefault(
+            value, {"original": original, "ids": set()}
+        )
+        bucket["ids"].add(assertion.get("id"))
+    disputes = []
+    for attribute, values in sorted(groups.items()):
+        if len(values) < 2:
+            continue
+        disputes.append({
+            "attribute": attribute,
+            "values": sorted(v["original"] for v in values.values()),
+            "assertion_ids": sorted(
+                aid for v in values.values() for aid in v["ids"] if aid
+            ),
+        })
+    return disputes
+
+
+def annotate_clusters(clusters: List[Dict]) -> int:
+    """Attach disputed_attributes to each cluster that has any. Returns count."""
+    annotated = 0
+    for cluster in clusters:
+        disputes = find_disputes(cluster.get("assertions") or [])
+        if disputes:
+            cluster["disputed_attributes"] = disputes
+            annotated += 1
+    return annotated

backend/agents/linker.py

@@ -30,9 +30,23 @@ def _family(assertion: Dict) -> str:
     )
 
 
+def _xref_keys(assertion: Dict) -> List[str]:
+    """Cross-level join keys: detail references and member tags."""
+    location = assertion.get("location_key") or {}
+    keys = []
+    ref = re.sub(r"\s+", "", str(location.get("detail_reference") or "")).upper()
+    if ref:
+        keys.append(f"detail:{ref}")
+    tag = re.sub(r"\s+", "", str(location.get("tag") or "")).upper()
+    if re.match(r"^[A-Z]+\d", tag):  # member marks: W12X26, HSS16X4X5/8, ...
+        keys.append(f"tag:{tag}")
+    return keys
+
+
 def build_link_scopes(sheets: List[Dict]) -> List[AgentScope]:
     """Partition facts by level and object/tag family, then enforce a hard cap."""
     buckets: Dict[Tuple[str, str], List[Dict]] = defaultdict(list)
+    xref: Dict[str, List[Dict]] = defaultdict(list)
     for sheet in sheets:
         for assertion in sheet.get("assertions", []):
             enriched = {
@@ -44,6 +58,8 @@ def build_link_scopes(sheets: List[Dict]) -> List[AgentScope]:
             level = str((assertion.get("location_key") or {}).get("level")
                         or sheet.get("level") or "unknown").lower()
             buckets[(level, _family(assertion))].append(enriched)
+            for key in _xref_keys(assertion):
+                xref[key].append(enriched)
 
     scopes: List[AgentScope] = []
     cap = max(2, config.AGENT_LINK_MAX_ASSERTIONS)
@@ -56,6 +72,18 @@ def build_link_scopes(sheets: List[Dict]) -> List[AgentScope]:
                 scope_id=f"{level}:{family}:{offset // cap + 1}",
                 payload={"assertions": chunk, "level": level, "family": family},
             ))
+    for key, assertions in sorted(xref.items()):
+        sheets_present = {a.get("sheet_number") for a in assertions}
+        if len(assertions) < 2 or len(sheets_present) < 2:
+            continue
+        chunk = assertions[:cap]
+        if len({a.get("sheet_number") for a in chunk}) < 2:
+            continue  # cap landed on a single sheet — xref adds nothing
+        scopes.append(AgentScope(
+            scope_id=f"xref:{key}",
+            payload={"assertions": chunk,
+                     "level": "xref", "family": key},
+        ))
     return scopes
 
 

backend/agents/memory.py

@@ -7,7 +7,8 @@ import threading
 from typing import Any, Dict, Iterable, Optional
 
 
-_COLLECTION_KEYS = {"sheets", "clusters", "findings", "decisions", "rfis"}
+_COLLECTION_KEYS = {"sheets", "clusters", "findings", "decisions", "rfis",
+                    "suppressed"}
 _MAPPING_KEYS = {"sheet_index", "jurisdiction", "object_graph"}
 _MEMORY_KEYS = _COLLECTION_KEYS | _MAPPING_KEYS
 

backend/agents/runner.py

@@ -11,6 +11,7 @@ from backend.agents.code_agent import CodeAgent, build_code_scopes
 from backend.agents.completeness import CompletenessAgent, build_sheet_summaries
 from backend.agents.conflict_critic import ConflictCriticAgent
 from backend.agents.construct_agent import ConstructabilityAgent, build_construct_scopes
+from backend.agents.disputes import annotate_clusters
 from backend.agents.extractors import (
     JurisdictionAgent,
     SheetExtractorAgent,
@@ -20,6 +21,9 @@ from backend.agents.linker import LinkerAgent, build_link_scopes, build_object_g
 from backend.agents.memory import ProjectMemory
 from backend.agents.orchestrator import Orchestrator
 from backend.agents.rfi_writer import RFIWriterAgent
+from backend.agents.verifier import (
+    EvidenceVerifierAgent, apply_verdicts, select_findings,
+)
 from backend.llm import reset_cost
 from backend.pipeline.pdf_processor import convert_pdf_to_images
 from backend.pipeline.report import build_report, to_markdown
@@ -113,6 +117,9 @@ def run_agent_pipeline(
         for artifact in result.artifacts
     ][:config.CLUSTER_MAX]
     object_graph = build_object_graph(clusters)
+    disputed_count = annotate_clusters(clusters)
+    if disputed_count:
+        orchestrator.stage(f"[Link] {disputed_count} clusters carry disputed extracted values")
     memory.replace("clusters", clusters)
     memory.replace("object_graph", object_graph)
     memory.dump("03-link.json")
@@ -164,6 +171,41 @@ def run_agent_pipeline(
         for result in code_results + construct_results + completeness_results
         for artifact in result.artifacts
     ]
+
+    orchestrator.stage("Agent wave 5b: evidence verification")
+    sheet_to_page = {str(s.get("sheet_number")): s.get("page_number")
+                     for s in sheets}
+    verify_targets = select_findings(
+        specialist_findings, clusters,
+        max_checks=config.AGENT_VERIFY_MAX_CHECKS,
+        severities=config.AGENT_VERIFY_SEVERITIES,
+    )
+    target_indexes = {id(f): i for i, f in enumerate(specialist_findings)}
+    verify_scopes = []
+    for finding in verify_targets:
+        images = [
+            page_to_b64[sheet_to_page[str(name)]]
+            for name in (finding.get("sheets") or [])[:config.AGENT_CONFLICT_MAX_IMAGES]
+            if sheet_to_page.get(str(name)) in page_to_b64
+        ]
+        if not images:
+            continue  # never judge evidence against images we could not load
+        verify_scopes.append(AgentScope(
+            scope_id=f"verify:{target_indexes[id(finding)]}",
+            payload={
+                "finding_index": target_indexes[id(finding)],
+                "finding": finding,
+                "images_b64": images,
+            },
+        ))
+    verify_results = orchestrator.run_scopes(
+        EvidenceVerifierAgent(usage), verify_scopes, config.AGENT_VERIFY_CONCURRENCY)
+    suppressed = apply_verdicts(specialist_findings, verify_results)
+    if suppressed:
+        suppressed_ids = {id(f) for f in suppressed}
+        specialist_findings = [f for f in specialist_findings if id(f) not in suppressed_ids]
+        memory.replace("suppressed", suppressed)
+
     memory.extend("findings", specialist_findings)
     gap_findings = [
         {
@@ -221,7 +263,7 @@ def run_agent_pipeline(
             "project_intelligence": object_graph,
             "validated_issues": prioritized,
             "rfis": [],
-            "suppressed_issues": [],
+            "suppressed_issues": memory.snapshot().get("suppressed") or [],
         })
         progress = store.progress(queue)
         # Same usage/stats summary block as the wave-7 path (rfis: 0 — they
@@ -297,6 +339,7 @@ def run_agent_pipeline(
         "project_intelligence": object_graph,
         "validated_issues": prioritized,
         "rfis": rfis,
+        "suppressed_issues": memory.snapshot().get("suppressed") or [],
     })
     cost = usage.snapshot()
     orchestrator.stats.calls = cost["calls"]

backend/agents/verifier.py

@@ -0,0 +1,95 @@
+"""Wave 5b: vision fact-check of extracted evidence against cited sheet images."""
+
+from backend import config
+from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure
+from backend.llm import call_json
+from backend.pipeline._serialize import dumps
+from backend.pipeline._stage import collect_list, render
+from backend.prompts import VERIFY_SYSTEM_PROMPT, VERIFY_USER_INSTRUCTION
+
+_SEVERITY_RANK = {"critical": 0, "high": 1, "medium": 2, "low": 3}
+_VERDICTS = ("confirmed", "corrected", "not_found")
+
+
+def select_findings(findings, clusters, max_checks, severities):
+    """Severity-gated selection plus any finding tied to a disputed cluster."""
+    disputed_keys = {c.get("key") for c in clusters if c.get("disputed_attributes")}
+    selected = [f for f in findings
+                if str(f.get("severity") or "").lower() in severities
+                or f.get("cluster_key") in disputed_keys]
+    selected.sort(key=lambda f: _SEVERITY_RANK.get(
+        str(f.get("severity") or "").lower(), 9))
+    return selected[:max_checks]
+
+
+def _valid_verdict(item):
+    if not isinstance(item, dict):
+        return None
+    verdict = str(item.get("verdict") or "").lower()
+    if verdict not in _VERDICTS:
+        return None
+    return {"sheet": item.get("sheet") or "",
+            "source_text": item.get("source_text") or "",
+            "verdict": verdict,
+            "actual_text": item.get("actual_text"),
+            "notes": item.get("notes")}
+
+
+def _status(verdicts):
+    if not verdicts:
+        return "unverified"
+    confirmed = sum(1 for v in verdicts if v["verdict"] == "confirmed")
+    if confirmed == len(verdicts):
+        return "confirmed"
+    if confirmed == 0:
+        return "refuted"
+    return "mixed"
+
+
+class EvidenceVerifierAgent:
+    name = "verify"
+
+    def __init__(self, usage: AgentUsage) -> None:
+        self.usage = usage
+
+    def run(self, scope: AgentScope) -> AgentResult:
+        try:
+            finding = scope.payload["finding"]
+            instruction = render(VERIFY_USER_INSTRUCTION, {"finding": dumps(finding)})
+            parsed = call_json(
+                system_prompt=VERIFY_SYSTEM_PROMPT,
+                user_text=instruction,
+                images_b64=scope.payload.get("images_b64") or [],
+                max_tokens=config.VERIFY_MAX_TOKENS,
+                model=config.AGENT_VERIFY_MODEL,
+                reasoning_effort=config.AGENT_VERIFY_REASONING_EFFORT or None,
+                usage_tracker=self.usage,
+                usage_stage="agent.verify",
+            )
+            verdicts = collect_list(parsed, "verdicts", _valid_verdict)
+            return AgentResult(scope_id=scope.scope_id, artifacts=[{
+                "finding_index": scope.payload["finding_index"],
+                "status": _status(verdicts),
+                "verdicts": verdicts,
+            }])
+        except Exception as exc:
+            return failure(scope, exc)
+
+
+def apply_verdicts(findings, verify_results):
+    """Annotate findings with verification; return refuted ones to suppress."""
+    by_index = {}
+    for result in verify_results:
+        for artifact in result.artifacts:
+            by_index[artifact["finding_index"]] = artifact
+    suppressed = []
+    for index, finding in enumerate(findings):
+        artifact = by_index.get(index)
+        if not artifact:
+            continue
+        finding["verification"] = {"status": artifact["status"],
+                                   "verdicts": artifact["verdicts"]}
+        if artifact["status"] == "refuted":
+            finding["confidence"] = "low"
+            suppressed.append(finding)
+    return suppressed

backend/config.py

@@ -47,6 +47,18 @@ AGENT_CONFLICT_CONCURRENCY   = int(os.getenv("AGENT_CONFLICT_CONCURRENCY", "4"))
 AGENT_SPECIALIST_CONCURRENCY = int(os.getenv("AGENT_SPECIALIST_CONCURRENCY", "4"))
 AGENT_RFI_CONCURRENCY        = int(os.getenv("AGENT_RFI_CONCURRENCY", "4"))
 
+# Wave 5b evidence verification (vision fact-check of cited sheet text)
+AGENT_VERIFY_MODEL         = os.getenv("AGENT_VERIFY_MODEL", "") or MODEL
+AGENT_VERIFY_CONCURRENCY   = int(os.getenv("AGENT_VERIFY_CONCURRENCY", "4"))
+AGENT_VERIFY_MAX_CHECKS    = int(os.getenv("AGENT_VERIFY_MAX_CHECKS", "20"))
+AGENT_VERIFY_SEVERITIES    = {
+    s.strip().lower()
+    for s in os.getenv("AGENT_VERIFY_SEVERITIES", "critical,high").split(",")
+    if s.strip()
+}
+AGENT_VERIFY_REASONING_EFFORT = os.getenv("AGENT_VERIFY_REASONING_EFFORT", "low").strip()
+VERIFY_MAX_TOKENS          = int(os.getenv("VERIFY_MAX_TOKENS", "8192"))
+
 # Agent-mode human-review gate. When on (default), Agent runs stop after the
 # Brain merge and wait for human decisions before RFIs/final report/email go
 # out. AGENT_REVIEW_AUDIT_SAMPLE caps how many clean clusters get added to the

backend/pipeline/_serialize.py

@@ -54,6 +54,8 @@ def slim_clusters(clusters: List[Dict]) -> List[Dict]:
             "location": c.get("location"),
             "disciplines": c.get("disciplines"),
             "kind": c.get("kind"),
+            **({"disputed_attributes": c["disputed_attributes"]}
+               if c.get("disputed_attributes") else {}),
             "assertions": [slim_assertion(a) for a in c.get("assertions", [])],
         }
         for c in clusters

backend/pipeline/conflict_checker.py

@@ -32,6 +32,16 @@ def _evidence_block(cluster: Dict) -> str:
             f"{a.get('attribute','')} = {a.get('value','')} | "
             f"\"{a.get('source_text','')}\""
         )
+    disputes = cluster.get("disputed_attributes") or []
+    if disputes:
+        lines.append("")
+        for d in disputes:
+            lines.append(
+                "DISPUTED VALUE (possible extraction misread): "
+                f"attribute={d.get('attribute','')} "
+                f"values={' | '.join(d.get('values') or [])} "
+                f"(assertions {', '.join(d.get('assertion_ids') or [])})"
+            )
     return "\n".join(lines)
 
 
@@ -83,6 +93,8 @@ def _check_one(cluster: Dict, page_to_b64: Dict[int, str]) -> List[Dict]:
         user_text=user_text,
         images_b64=_images_for(cluster, page_to_b64),
         max_tokens=config.REASON_MAX_TOKENS,
+        reasoning_effort=config.EXTRACT_REASONING_EFFORT or None,
+        reasoning_max_tokens=config.EXTRACT_REASONING_MAX_TOKENS or None,
     )
     if isinstance(parsed, list):
         candidates = parsed

backend/pipeline/constructability.py

@@ -27,6 +27,10 @@ def constructability_review(sheets: List[Dict], clusters: List[Dict],
             "assertions": dumps(slim_sheets(sheets)),
             "clusters": dumps(slim_clusters(clusters)),
             "conflicts": dumps(conflicts),
+            "disputes": dumps([
+                d for cluster in clusters
+                for d in (cluster.get("disputed_attributes") or [])
+            ]),
         },
         max_tokens=config.CONSTRUCT_MAX_TOKENS,
     )

backend/pipeline/runner.py

@@ -33,6 +33,7 @@ from backend.pipeline.normalizer import normalize_assertions, build_project_inte
 from backend.pipeline.clusterer import cluster_by_location
 from backend.pipeline.llm_clusterer import cluster_by_location_llm
 from backend import config
+from backend.agents.disputes import annotate_clusters
 from backend.pipeline.conflict_checker import check_conflicts
 from backend.pipeline.qaqc_review import senior_review
 from backend.pipeline.code_review import code_review
@@ -129,6 +130,10 @@ def _run_stages(
     else:
         clusters = cluster_by_location(sheets)
 
+    disputed_count = annotate_clusters(clusters)
+    if disputed_count:
+        print(f"[Cluster] {disputed_count} cluster(s) carry disputed extracted values")
+
     stage("Reason over clusters (conflicts)")
     conflicts = check_conflicts(clusters, pages)
 

backend/prompts.py

@@ -410,6 +410,9 @@ What is NOT a conflict:
 - Anything not supported with drawing evidence.
 Be conservative:
 - Only flag genuine disagreements.
+- When a value is marked DISPUTED (possible extraction misread), verify it against
+  the sheet images before relying on either reading; if the images do not resolve
+  it, do not assert a conflict from one reading alone.
 - A clean cluster with no contradiction must return an empty conflicts array.
 - missing_element requires evidence that another discipline would reasonably be expected to show the missing item.
 For each conflict:
@@ -447,6 +450,26 @@ Clustered assertions (evidence):
 {evidence}"""
 
 
+# ---------------------------------------------------------------------------
+# Wave 5b - evidence verification (vision fact-check of cited sheet text)
+# ---------------------------------------------------------------------------
+
+VERIFY_SYSTEM_PROMPT = """You are a meticulous construction document checker verifying machine-extracted evidence against the actual drawing sheet images.
+For each evidence item you are given the sheet it was extracted from and the verbatim text the extractor claims appears there.
+Judge each item against the images:
+- confirmed: the text (or an obvious equivalent) appears on the cited sheet and means what the finding claims.
+- corrected: the sheet shows a DIFFERENT value than the extracted text. Give the actual verbatim text.
+- not_found: nothing like the extracted text appears on the cited sheet.
+Be strict about numbers, quantities, and member sizes: "(2) 2x6" and "(5) 2x6" are different values. HSS16x4 and HSS16x16 are different values.
+Use plain ASCII only.
+Respond only with valid JSON."""
+
+VERIFY_USER_INSTRUCTION = """Verify this finding's evidence against the attached sheet images.
+Respond ONLY with a valid JSON object - no markdown fences, no explanation:
+{ "verdicts": [ { "sheet": "string", "source_text": "the evidence text judged", "verdict": "confirmed | corrected | not_found", "actual_text": "verbatim sheet text when corrected, else null", "notes": "string or null" } ] }
+Finding: {finding}"""
+
+
 # ---------------------------------------------------------------------------
 # Stage 6 - senior architect full-set QAQC review  (NOT WIRED YET)
 # ---------------------------------------------------------------------------
@@ -545,6 +568,12 @@ Flag:
 Rules:
 - Only flag issues supported by drawing evidence.
 - Be specific about the location and why it is a constructability risk.
+- Assertions are machine-extracted from sheet images and may contain misread values,
+  especially quantities and member sizes (e.g. "(2) 2x6" vs "(5) 2x6").
+- When the cluster lists disputed_attributes, or two evidence items disagree on a
+  numeric value, do NOT assert a buildability conclusion from one reading. Report the
+  ambiguity itself (category "detail_gap", confidence "low") and state that the value
+  needs verification against the sheet.
 - Use plain ASCII only.
 Respond only with valid JSON."""
 
@@ -554,7 +583,8 @@ Respond ONLY with a valid JSON object - no markdown fences, no explanation:
 If no constructability issues are found, return: { "issues": [] }
 Extracted assertions: {assertions}
 Clusters: {clusters}
-Cross-discipline conflicts already found: {conflicts}"""
+Cross-discipline conflicts already found: {conflicts}
+Disputed extracted values in this cluster (possible vision misreads - treat as unverified): {disputes}"""
 
 
 # ---------------------------------------------------------------------------

backend/review/finalizer.py

@@ -216,7 +216,7 @@ def finalize_review(job_id: str, out_dir: str) -> dict:
     rfis = _draft_rfis(kept)
 
     report["validated_issues"] = kept
-    report["suppressed_issues"] = suppressed
+    report["suppressed_issues"] = (report.get("suppressed_issues") or []) + suppressed
     report["rfis"] = rfis
     summary = report.setdefault("summary", {})
     summary["agent_status"] = "complete"

tests/agents/test_classic_constructability_disputes.py

@@ -0,0 +1,51 @@
+"""Classic pipeline path must also satisfy the {disputes} placeholder added to
+CONSTRUCTABILITY_USER_INSTRUCTION (agent path substitutes it in construct_agent.py;
+the classic stage builds its own subs dict)."""
+
+from unittest.mock import patch
+
+from backend.pipeline._stage import render
+from backend.pipeline.constructability import constructability_review
+from backend.prompts import CONSTRUCTABILITY_USER_INSTRUCTION
+
+
+def _cluster_with_dispute():
+    return {
+        "key": "c1",
+        "assertions": [],
+        "disputed_attributes": [{
+            "attribute": "stud_pack_size",
+            "values": ["(2) 2x6 STUD PACK", "(5) 2x6 STUD PACK"],
+            "assertion_ids": ["a1", "a2"],
+        }],
+    }
+
+
+def test_classic_constructability_supplies_disputes_sub():
+    captured = {}
+
+    def fake_call_stage(system_prompt, user_instruction, subs=None, **kwargs):
+        captured["subs"] = subs or {}
+        return {"issues": []}
+
+    with patch("backend.pipeline.constructability.call_stage", fake_call_stage):
+        constructability_review([], [_cluster_with_dispute()], [])
+
+    assert "disputes" in captured["subs"], "classic path must substitute {disputes}"
+    rendered = render(CONSTRUCTABILITY_USER_INSTRUCTION, captured["subs"])
+    assert "{disputes}" not in rendered
+    assert "(5) 2x6 STUD PACK" in rendered
+
+
+def test_classic_constructability_disputes_defaults_empty():
+    captured = {}
+
+    def fake_call_stage(system_prompt, user_instruction, subs=None, **kwargs):
+        captured["subs"] = subs or {}
+        return {"issues": []}
+
+    with patch("backend.pipeline.constructability.call_stage", fake_call_stage):
+        constructability_review([], [{"key": "c2", "assertions": []}], [])
+
+    rendered = render(CONSTRUCTABILITY_USER_INSTRUCTION, captured["subs"])
+    assert "{disputes}" not in rendered

tests/agents/test_disputes.py

@@ -0,0 +1,69 @@
+from backend.agents.disputes import annotate_clusters, find_disputes
+
+
+def _a(id_, attribute, value):
+    return {"id": id_, "attribute": attribute, "value": value,
+            "source_text": value}
+
+
+def test_find_disputes_flags_same_attribute_different_values():
+    assertions = [
+        _a("a1", "stud_pack_size", "(2) 2x6 STUD PACK"),
+        _a("a2", "stud_pack_size", "(5) 2x6 STUD PACK"),
+        _a("a3", "beam_size", "HSS16X4X5/8"),
+    ]
+    disputes = find_disputes(assertions)
+    assert len(disputes) == 1
+    assert disputes[0]["attribute"] == "stud_pack_size"
+    assert disputes[0]["values"] == ["(2) 2x6 STUD PACK", "(5) 2x6 STUD PACK"]
+    assert disputes[0]["assertion_ids"] == ["a1", "a2"]
+
+
+def test_find_disputes_ignores_agreeing_values_and_blanks():
+    assertions = [
+        _a("a1", "beam_size", "HSS16X4X5/8"),
+        _a("a2", "beam_size", "  hss16x4x5/8 "),   # same after normalize
+        _a("a3", "", "orphan"),                     # no attribute -> skipped
+        _a("a4", "beam_size", ""),                  # no value -> skipped
+    ]
+    assert find_disputes(assertions) == []
+
+
+def test_annotate_clusters_writes_disputed_attributes():
+    clusters = [
+        {"key": "c1", "assertions": [
+            _a("a1", "stud_pack_size", "(2) 2x6"),
+            _a("a2", "stud_pack_size", "(5) 2x6"),
+        ]},
+        {"key": "c2", "assertions": [_a("a3", "x", "1"), _a("a4", "x", "1")]},
+    ]
+    assert annotate_clusters(clusters) == 1
+    assert clusters[0]["disputed_attributes"][0]["attribute"] == "stud_pack_size"
+    assert "disputed_attributes" not in clusters[1]
+
+
+def test_slim_clusters_preserves_disputed_attributes():
+    from backend.pipeline._serialize import slim_clusters
+    cluster = {"key": "c1", "assertions": [],
+               "disputed_attributes": [{"attribute": "a", "values": ["1", "2"],
+                                         "assertion_ids": ["x", "y"]}]}
+    slim = slim_clusters([cluster])[0]
+    assert slim["disputed_attributes"][0]["values"] == ["1", "2"]
+
+
+def test_find_disputes_handles_none_and_zero_values():
+    # None value/attribute -> skipped; numeric 0 is a real value, not blank
+    assertions = [
+        {"id": "a1", "attribute": "count", "value": 0},
+        {"id": "a2", "attribute": "count", "value": 1},
+        {"id": "a3", "attribute": None, "value": "x"},
+        {"id": "a4", "attribute": "count", "value": None},
+    ]
+    disputes = find_disputes(assertions)
+    assert len(disputes) == 1
+    assert disputes[0]["values"] == ["0", "1"]
+
+
+def test_find_disputes_empty_input():
+    assert find_disputes([]) == []
+    assert annotate_clusters([]) == 0

tests/agents/test_linker_xref.py

@@ -0,0 +1,74 @@
+from backend.agents.base import AgentScope
+from backend.agents.linker import build_link_scopes
+
+
+def _sheet(number, page, level, assertions):
+    return {"sheet_number": number, "page_number": page,
+            "discipline": "Structural", "level": level,
+            "assertions": assertions}
+
+
+def _assertion(id_, ref=None, tag=None, level=None):
+    return {"id": id_, "attribute": "stud_pack_size", "value": "(5) 2x6",
+            "source_text": "(5) 2x6 STUD PACK",
+            "location_key": {"detail_reference": ref, "tag": tag,
+                             "level": level}}
+
+
+def test_xref_scope_joins_same_detail_reference_across_levels():
+    sheets = [
+        _sheet("S101", 10, "foundation", [_assertion("a1", ref="A/S205")]),
+        _sheet("S205", 20, "roof", [_assertion("a2", ref="A/S205")]),
+        _sheet("S401", 30, "roof", [_assertion("a3", ref="A/S205")]),
+    ]
+    scopes = build_link_scopes(sheets)
+    xref = [s for s in scopes if s.scope_id.startswith("xref:")]
+    assert xref, "expected a cross-level detail-reference scope"
+    ids = {a["id"] for s in xref for a in s.payload["assertions"]}
+    assert ids == {"a1", "a2", "a3"}
+
+
+def test_xref_scope_requires_two_distinct_sheets():
+    sheets = [
+        _sheet("S401", 30, "roof", [_assertion("a1", ref="A/S205"),
+                                    _assertion("a2", ref="A/S205")]),
+    ]
+    scopes = build_link_scopes(sheets)
+    assert not [s for s in scopes if s.scope_id.startswith("xref:")]
+
+
+def test_xref_scope_joins_shared_member_tag():
+    sheets = [
+        _sheet("S102", 5, "roof", [_assertion("a1", tag="HSS16X4X5/8")]),
+        _sheet("S401", 30, "unknown", [_assertion("a2", tag="HSS16X4X5/8")]),
+    ]
+    scopes = build_link_scopes(sheets)
+    xref = [s for s in scopes if s.scope_id.startswith("xref:")]
+    assert xref
+
+
+def test_xref_scope_joins_single_letter_member_mark():
+    # W-shapes (W12X26) are the most common steel marks and have one leading letter
+    sheets = [
+        _sheet("S102", 5, "roof", [_assertion("a1", tag="W12X26")]),
+        _sheet("S401", 30, "unknown", [_assertion("a2", tag="W12X26")]),
+    ]
+    scopes = build_link_scopes(sheets)
+    xref = [s for s in scopes if s.scope_id.startswith("xref:")]
+    assert xref, "single-letter member marks (W12X26) must join xref scopes"
+
+
+def test_xref_scope_rechecks_sheet_diversity_after_cap(monkeypatch):
+    from backend import config
+    monkeypatch.setattr(config, "AGENT_LINK_MAX_ASSERTIONS", 2)
+    sheets = [
+        _sheet("S401", 30, "roof", [_assertion("a1", ref="A/S205"),
+                                    _assertion("a2", ref="A/S205")]),
+        _sheet("S205", 20, "roof", [_assertion("a3", ref="A/S205")]),
+    ]
+    scopes = build_link_scopes(sheets)
+    xref = [s for s in scopes if s.scope_id.startswith("xref:")]
+    for scope in xref:
+        sheets_in_scope = {a["sheet_number"] for a in scope.payload["assertions"]}
+        assert len(sheets_in_scope) >= 2, \
+            "capped xref scope must still span two sheets"

tests/agents/test_verifier.py

@@ -0,0 +1,68 @@
+from unittest.mock import patch
+
+from backend.agents.base import AgentScope, AgentUsage
+
+from backend.agents.verifier import (
+    EvidenceVerifierAgent, apply_verdicts, select_findings,
+)
+
+
+def _finding(sev="critical", issue_id="i1", sheets=("S401",), cluster_key=None):
+    f = {"issue_id": issue_id, "severity": sev, "confidence": "high",
+         "source_stage": "constructability", "sheets": list(sheets),
+         "description": "HSS16x4 on (2) 2x6 STUD PACK is unbuildable",
+         "evidence": [{"sheet": "S401", "source_text": "(2) 2x6 STUD PACK",
+                       "asserted_value": "3-inch width"}]}
+    if cluster_key:
+        f["cluster_key"] = cluster_key
+    return f
+
+
+def test_select_findings_by_severity_and_dispute():
+    findings = [_finding("critical"), _finding("low", "i2"),
+                _finding("medium", "i3", cluster_key="c9")]
+    clusters = [{"key": "c9", "disputed_attributes": [{"attribute": "a"}]}]
+    selected = select_findings(findings, clusters, max_checks=20,
+                               severities={"critical", "high"})
+    assert [f["issue_id"] for f in selected] == ["i1", "i3"]
+
+
+def test_select_findings_respects_cap():
+    findings = [_finding("critical", f"i{n}") for n in range(30)]
+    selected = select_findings(findings, [], max_checks=5,
+                               severities={"critical"})
+    assert len(selected) == 5
+
+
+def test_run_attaches_verdicts_and_marks_refuted():
+    agent = EvidenceVerifierAgent(usage=AgentUsage())
+    scope = AgentScope(scope_id="verify:0", payload={
+        "finding_index": 0,
+        "finding": _finding(),
+        "images_b64": ["QUJD"],
+    })
+    verdicts = {"verdicts": [
+        {"sheet": "S401", "source_text": "(2) 2x6 STUD PACK",
+         "verdict": "corrected", "actual_text": "(5) 2x6 STUD PACK",
+         "notes": "callout reads (5)"},
+    ]}
+    with patch("backend.agents.verifier.call_json", return_value=verdicts):
+        result = agent.run(scope)
+    assert not result.error
+    artifact = result.artifacts[0]
+    assert artifact["finding_index"] == 0
+    assert artifact["status"] == "refuted"      # no evidence confirmed
+    assert artifact["verdicts"][0]["actual_text"] == "(5) 2x6 STUD PACK"
+
+
+def test_apply_verdicts_annotates_and_suppresses():
+    from backend.agents.base import AgentResult
+    findings = [_finding("critical", "i1"), _finding("high", "i2")]
+    results = [AgentResult(scope_id="verify:0", artifacts=[
+        {"finding_index": 0, "status": "refuted", "verdicts": []},
+        {"finding_index": 1, "status": "confirmed", "verdicts": []},
+    ])]
+    suppressed = apply_verdicts(findings, results)
+    assert suppressed == [findings[0]]
+    assert findings[0]["verification"]["status"] == "refuted"
+    assert findings[1]["verification"]["status"] == "confirmed"

tests/agents/test_wave5b_suppression.py

@@ -0,0 +1,86 @@
+"""Runner-level wave-5b tests: suppression path and zero-image guard."""
+
+import backend.agents.runner as runner_mod
+from backend.agents.base import AgentResult
+from backend.agents.runner import run_agent_pipeline
+
+
+def _finding(sheets):
+    return {
+        "issue_id": "C1", "severity": "critical", "confidence": "high",
+        "source_stage": "constructability", "sheets": sheets,
+        "description": "HSS16x4 on (2) 2x6 STUD PACK is unbuildable",
+        "evidence": [{"sheet": sheets[0], "source_text": "(2) 2x6 STUD PACK"}],
+    }
+
+
+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, finding):
+    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": []},
+    ]))
+    monkeypatch.setattr(runner_mod, "ConflictCriticAgent", _stub_agent([]))
+    monkeypatch.setattr(runner_mod, "CodeAgent", _stub_agent([]))
+    monkeypatch.setattr(runner_mod, "ConstructabilityAgent", _stub_agent([finding]))
+    monkeypatch.setattr(runner_mod, "CompletenessAgent", _stub_agent([]))
+    monkeypatch.setattr(
+        runner_mod, "BrainAgent",
+        lambda usage: type("B", (), {
+            "run": lambda self, findings, sheet_index, jurisdiction:
+                (list(findings), [])})())
+
+
+def test_refuted_finding_is_suppressed_not_crash(monkeypatch, tmp_path):
+    """Regression: memory.replace("suppressed", ...) must not KeyError."""
+    _patch_pipeline(monkeypatch, _finding(["S401"]))
+    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": "callout reads (5)"},
+        ]})
+    pdf = tmp_path / "dummy.pdf"
+    pdf.write_bytes(b"%PDF-1.4\n")
+    report = run_agent_pipeline(str(pdf), out_dir=str(tmp_path),
+                                require_review=False)
+    assert [f["issue_id"] for f in report["suppressed_issues"]] == ["C1"]
+    assert report["suppressed_issues"][0]["verification"]["status"] == "refuted"
+
+
+def test_zero_image_finding_is_not_suppressed(monkeypatch, tmp_path):
+    """A finding whose sheets resolve to no page images must not be judged
+    (and must never be refuted) without pixels."""
+    _patch_pipeline(monkeypatch, _finding(["S999"]))  # no such sheet
+    monkeypatch.setattr(
+        "backend.agents.verifier.call_json",
+        lambda **kwargs: {"verdicts": [
+            {"sheet": "S999", "source_text": "(2) 2x6 STUD PACK",
+             "verdict": "not_found", "actual_text": None, "notes": None},
+        ]})
+    pdf = tmp_path / "dummy.pdf"
+    pdf.write_bytes(b"%PDF-1.4\n")
+    report = run_agent_pipeline(str(pdf), out_dir=str(tmp_path),
+                                require_review=False)
+    assert report["suppressed_issues"] == []
+    validated = report.get("validated_issues") or []
+    assert any(f.get("issue_id") == "C1" for f in validated)

tests/review/test_finalizer.py

@@ -85,6 +85,29 @@ def test_finalize_confirm_keeps_confirmed(monkeypatch, tmp_path):
     assert report["summary"]["agent_status"] == "complete"
 
 
+def test_finalize_preserves_verifier_suppressed(monkeypatch, tmp_path):
+    """Wave-5b (verifier) suppressions must survive review finalization and
+    merge with review-rejected suppressions."""
+    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": "reject", "reason_code": "not_a_contradiction"}],
+    )
+    path = os.path.join(str(tmp_path), "conflicts.json")
+    with open(path, encoding="utf-8") as f:
+        report = json.load(f)
+    report["suppressed_issues"] = [
+        {"issue_id": "C1", "verification": {"status": "refuted"}}]
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(report, f)
+    final = finalize_review("job1", str(tmp_path))
+    ids = [f["issue_id"] for f in final["suppressed_issues"]]
+    assert ids == ["C1", "AGENT-0001"]
+
+
 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: [])