Add required human review gate to the Agent pipeline.

Agent web jobs now stop after Brain consolidation and enter needs_review
with a persisted review queue (blocking: high-severity, low-confidence,
sensitive-category findings; audit sample of clean clusters). Humans
decide confirm/reject/unsure/needs_clarification via new review API and
frontend queue; a finalizer applies decisions (rejections suppressed with
reason codes), performs bounded targeted reruns for clarifications,
drafts RFIs only for kept issues, and only then marks the job done and
sends the final email. Two-phase email (review-required, then final
report), per-decision feedback labels with redacted aggregate metrics,
restart recovery from job artifacts, and CLI --no-review bypass.
Classic pipeline unchanged. 65 non-LLM tests.

backend/.env.example

@@ -27,6 +27,13 @@ AGENT_CONFLICT_CONCURRENCY=4
 AGENT_SPECIALIST_CONCURRENCY=4
 AGENT_RFI_CONCURRENCY=4
 
+# Agent-mode human-review gate (pipeline stops after Brain until a human reviews)
+AGENT_REQUIRE_REVIEW=true
+# Max clean clusters added to the review queue as non-blocking spot-checks
+AGENT_REVIEW_AUDIT_SAMPLE=5
+# Allow future cross-job review-feedback aggregation to include source_text/images/comments
+REVIEW_AGGREGATE_INCLUDE_TEXT=false
+
 # Pipeline tuning
 PDF_DPI=100
 MAX_PAGES=60

backend/agents/runner.py

@@ -23,6 +23,8 @@ from backend.agents.rfi_writer import RFIWriterAgent
 from backend.pipeline.pdf_processor import convert_pdf_to_images
 from backend.pipeline.report import build_report, to_markdown
 from backend.pipeline.sheet_index import derive_project_meta_from_cover
+from backend.review.gate import build_review_queue
+from backend.review.store import ReviewStore
 
 
 def run_agent_pipeline(
@@ -31,6 +33,7 @@ def run_agent_pipeline(
     on_stage: Optional[Callable[[str], None]] = None,
     project_input: Optional[Dict] = None,
     source_name: Optional[str] = None,
+    require_review: bool = True,
 ) -> Dict:
     """Run all scoped specialist waves and return a Classic-compatible report."""
     if not os.path.isfile(pdf_path):
@@ -192,6 +195,71 @@ def run_agent_pipeline(
         1 for decision in decisions if decision.get("action") == "merged"
     )
 
+    if require_review:
+        orchestrator.stage("Agent review gate: build human-review queue")
+        memory_snapshot = memory.snapshot()
+        queue = build_review_queue(memory_snapshot, prioritized, decisions,
+                                   limit=config.AGENT_REVIEW_AUDIT_SAMPLE)
+        store = ReviewStore(out_dir)
+        store.write_queue(queue)
+        candidate_conflicts = [_finding_as_conflict(item) for item in conflict_findings]
+        report = build_report(
+            conflicts=candidate_conflicts,
+            sheets=sheets,
+            clusters=clusters,
+            source=source_name or os.path.basename(pdf_path),
+        )
+        report.update({
+            "project_input": merged_input,
+            "jurisdiction": jurisdiction,
+            "sheet_index": sheet_index,
+            "project_intelligence": object_graph,
+            "validated_issues": prioritized,
+            "rfis": [],
+            "suppressed_issues": [],
+        })
+        progress = store.progress(queue)
+        # Same usage/stats summary block as the wave-7 path (rfis: 0 — they
+        # are drafted only after human review finalizes the run).
+        cost = usage.snapshot()
+        orchestrator.stats.calls = cost["calls"]
+        stats = orchestrator.stats.as_dict()
+        report["summary"].update({
+            "pipeline_mode": "agent",
+            "agent_status": "needs_review",
+            "review": progress,
+            "agent_stats": stats,
+            "by_stage": {
+                "conflicts": len(conflict_findings),
+                "qaqc": sum(
+                    1 for item in specialist_findings
+                    if item.get("source_stage") == "qaqc"
+                ),
+                "code": sum(
+                    1 for item in specialist_findings
+                    if item.get("source_stage") == "code"
+                ),
+                "constructability": sum(
+                    1 for item in specialist_findings
+                    if item.get("source_stage") == "constructability"
+                ),
+                "validated": len(prioritized),
+                "rfis": 0,
+            },
+            "cost_usd": round(cost["usd"], 4),
+            "llm_calls": cost["calls"],
+            "cached_calls": cost["cached"],
+            "cost_by_stage": cost["by_stage"],
+            "text_backend": "openrouter",
+            "models_used": cost["models"],
+        })
+        if out_dir:
+            _dump(out_dir, "conflicts.json", report)
+            _dump(out_dir, "validated_issues.json", prioritized)
+        # Snapshot for the review finalizer's targeted clarification reruns.
+        memory.dump("memory.json")
+        return report
+
     orchestrator.stage("Agent wave 7: per-finding RFI writers")
     rfi_scopes = [
         AgentScope(

backend/config.py

@@ -47,6 +47,17 @@ 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"))
 
+# 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
+# queue as non-blocking spot-checks. REVIEW_AGGREGATE_INCLUDE_TEXT controls
+# whether future cross-job review feedback aggregation may include verbatim
+# source_text/images/comments (off by default = privacy-preserving).
+AGENT_REQUIRE_REVIEW = os.getenv("AGENT_REQUIRE_REVIEW", "true").strip().lower() in ("1", "true", "yes")
+AGENT_REVIEW_AUDIT_SAMPLE = int(os.getenv("AGENT_REVIEW_AUDIT_SAMPLE", "5"))
+# NOTE: currently unwired - reserved for future cross-job aggregation tooling.
+REVIEW_AGGREGATE_INCLUDE_TEXT = os.getenv("REVIEW_AGGREGATE_INCLUDE_TEXT", "false").strip().lower() in ("1", "true", "yes")
+
 # -- Hybrid (local text LLM) ----------------------------------------
 # Optional OpenAI-compatible local endpoint (e.g. a vLLM box) for the text-only
 # QAQC stages. Vision stages ALWAYS use OpenRouter. The user picks hybrid per

backend/email_sender.py

@@ -38,6 +38,22 @@ def _send(msg: EmailMessage) -> bool:
         return False
 
 
+def send_review_required(recipient_email: str, report: dict, review_url: str) -> bool:
+    if not recipient_email or not _smtp_ready():
+        return False
+    msg = EmailMessage()
+    msg["Subject"] = f"Conflict Checker - review required - {report.get('source', 'drawing set')}"
+    msg["From"] = config.SMTP_FROM or config.SMTP_USER
+    msg["To"] = recipient_email
+    review = report.get("summary", {}).get("review", {})
+    msg.set_content(
+        "Agent analysis is complete and waiting for human review.\n\n"
+        f"Required review items: {review.get('required', 0)}\n"
+        f"Review URL: {review_url}\n"
+    )
+    return _send(msg)
+
+
 def send_conflict_report(
     recipient_email: str,
     report: Dict,

backend/jobs.py

@@ -11,6 +11,7 @@ to outputs/<job_id>/ so results survive a restart even though live status does
 not. No external queue/DB.
 """
 
+import json
 import os
 import time
 import uuid
@@ -21,7 +22,7 @@ from typing import Dict, Optional
 from backend import config
 from backend.agents.runner import run_agent_pipeline
 from backend.pipeline.runner import run_pipeline
-from backend.email_sender import send_conflict_report
+from backend.email_sender import send_conflict_report, send_review_required
 
 _jobs: Dict[str, Dict] = {}
 _lock = threading.Lock()
@@ -46,7 +47,7 @@ def create_job(pdf_path: str, source_filename: str, email: Optional[str] = None,
     with _lock:
         _jobs[job_id] = {
             "job_id": job_id,
-            "status": "queued",          # queued -> running -> done | error
+            "status": "queued",          # queued -> running -> done | needs_review | error
             "source": source_filename,
             "email": email or None,
             "project_input": project_input or {},
@@ -72,6 +73,16 @@ def _run(job_id: str, pdf_path: str, project_input: Optional[Dict] = None,
         _set(job_id, status="running")
         # Keep a copy of the source PDF so its sheets can be viewed later.
         os.makedirs(out_dir, exist_ok=True)
+        # Persist minimal job metadata so the disk fallback in get_job can
+        # recover the recipient email / pipeline mode after a server restart
+        # (plain json.dump, matching the _dump style used elsewhere).
+        with open(os.path.join(out_dir, "job.json"), "w", encoding="utf-8") as f:
+            json.dump({
+                "job_id": job_id,
+                "email": _jobs[job_id].get("email"),
+                "pipeline_mode": pipeline_mode,
+                "source": _jobs[job_id].get("source"),
+            }, f, indent=2)
         shutil.copy2(pdf_path, os.path.join(out_dir, "source.pdf"))
         runner = run_agent_pipeline if pipeline_mode == "agent" else run_pipeline
         runner_kwargs = {
@@ -82,10 +93,21 @@ def _run(job_id: str, pdf_path: str, project_input: Optional[Dict] = None,
         }
         if pipeline_mode == "classic":
             runner_kwargs["text_local"] = text_local
+        else:
+            runner_kwargs["require_review"] = config.AGENT_REQUIRE_REVIEW
         report = runner(pdf_path, **runner_kwargs)
         report.setdefault("summary", {})["pipeline_mode"] = pipeline_mode
-        _set(job_id, status="done", report=report, finished_at=time.time(), stage=None)
-        _notify(job_id, report, out_dir)
+        if report["summary"].get("agent_status") == "needs_review":
+            # Human-review gate: hold the job, don't email the unreviewed report.
+            _set(job_id, status="needs_review", report=report,
+                 finished_at=time.time(), stage=None)
+            email = _jobs[job_id].get("email")
+            if email:
+                review_url = f"{config.APP_BASE_URL.rstrip('/')}/?job={job_id}"
+                send_review_required(email, report, review_url)
+        else:
+            _set(job_id, status="done", report=report, finished_at=time.time(), stage=None)
+            _notify(job_id, report, out_dir)
     except Exception as e:
         print(f"[Jobs] Job {job_id} failed: {e}")
         _set(job_id, status="error", error=str(e), finished_at=time.time())
@@ -158,24 +180,43 @@ def get_job(job_id: str) -> Optional[Dict]:
     if not os.path.isfile(report_path):
         return None
     try:
-        import json
         with open(report_path, encoding="utf-8") as f:
             report = json.load(f)
+        summary = report.get("summary", {})
+        # Recover the job's real state: a job that stopped at the review gate
+        # must come back as needs_review (not done) or it can never finalize.
+        status = "needs_review" if summary.get("agent_status") == "needs_review" else "done"
+        # job.json (written at job start) carries the recipient email and
+        # pipeline mode so the final notification still fires after a restart.
+        # Missing/corrupt job.json degrades to the previous derivations.
+        meta: Dict = {}
+        meta_path = os.path.join(config.OUTPUT_DIR, job_id, "job.json")
+        try:
+            with open(meta_path, encoding="utf-8") as f:
+                loaded = json.load(f)
+            if isinstance(loaded, dict):
+                meta = loaded
+        except (OSError, json.JSONDecodeError):
+            pass
         source_pdf = os.path.join(config.OUTPUT_DIR, job_id, "source.pdf")
-        return {
+        job = {
             "job_id":       job_id,
-            "status":       "done",
-            "source":       report.get("source", os.path.basename(report_path)),
-            "email":        None,
+            "status":       status,
+            "source":       meta.get("source") or report.get("source", os.path.basename(report_path)),
+            "email":        meta.get("email"),
             "project_input": report.get("project_input", {}),
-            "text_local":   report.get("summary", {}).get("text_backend") == "local",
-            "pipeline_mode": report.get("summary", {}).get("pipeline_mode", "classic"),
+            "text_local":   summary.get("text_backend") == "local",
+            "pipeline_mode": meta.get("pipeline_mode") or summary.get("pipeline_mode", "classic"),
             "stage":        None,
             "created_at":   os.path.getmtime(source_pdf) if os.path.isfile(source_pdf) else None,
             "finished_at":  os.path.getmtime(report_path),
             "report":       report,
             "error":        None,
         }
+        # Hydrate the in-memory registry so _set(...) transitions (reviewing,
+        # finalizing, done) work for restart-recovered jobs.
+        with _lock:
+            return dict(_jobs.setdefault(job_id, job))
     except Exception as e:
         print(f"[Jobs] Failed to load job {job_id} from disk: {e}")
         return None

backend/main.py

@@ -11,15 +11,21 @@ ever needs concurrency.
 
 import os
 import tempfile
+import threading
+import time
 from typing import Optional
 
 from fastapi import FastAPI, UploadFile, File, Form, HTTPException
 from fastapi.responses import HTMLResponse, JSONResponse, Response
 from fastapi.staticfiles import StaticFiles
 
+import backend.jobs
 from backend import config
-from backend.jobs import PIPELINE_MODES, create_job, get_job
+from backend.jobs import PIPELINE_MODES, create_job, get_job, _set
 from backend.pipeline.pdf_processor import render_page_jpeg
+from backend.review.feedback import decision_to_label, write_label
+from backend.review.finalizer import finalize_review
+from backend.review.store import ReviewStore
 
 app = FastAPI(title=config.APP_TITLE, version=config.APP_VERSION)
 
@@ -95,6 +101,110 @@ def job_status(job_id: str):
     return JSONResponse(job)
 
 
+@app.get("/jobs/{job_id}/review")
+def review_queue(job_id: str):
+    job = get_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    out_dir = job.get("out_dir") or os.path.join(config.OUTPUT_DIR, job_id)
+    # Read-only endpoint: don't create review/ dirs just by looking at them
+    # (readers already degrade to empty on missing files).
+    store = ReviewStore(out_dir, create=False)
+    queue = store.read_queue()
+    return {"queue": queue, "progress": store.progress(queue),
+            "decisions": store.read_decisions()}
+
+
+@app.post("/jobs/{job_id}/review-decisions")
+def save_review_decisions(job_id: str, payload: dict):
+    job = get_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    if job.get("status") not in ("needs_review", "reviewing"):
+        # Positive state guard, mirroring the finalize endpoint: only jobs
+        # sitting at (or working through) the review gate accept decisions.
+        raise HTTPException(status_code=409, detail={
+            "detail": f"cannot save review decisions for a job in status {job.get('status')}",
+        })
+    out_dir = job.get("out_dir") or os.path.join(config.OUTPUT_DIR, job_id)
+    store = ReviewStore(out_dir)
+    queue = store.read_queue()
+    items_by_id = {item.get("review_item_id"): item for item in queue}
+    saved = 0
+    try:
+        for decision in payload.get("decisions") or []:
+            store.append_decision(decision)
+            queue_item = items_by_id.get(decision.get("review_item_id"))
+            if queue_item is not None:
+                write_label(out_dir, decision_to_label(queue_item, decision, job))
+            saved += 1
+    except ValueError as e:
+        raise HTTPException(status_code=422, detail=str(e))
+    progress = store.progress(queue)
+    if job.get("status") == "needs_review" and saved > 0 and progress["remaining"] > 0:
+        try:
+            _set(job_id, status="reviewing")
+        except KeyError:
+            pass  # job not in the in-memory registry (e.g. loaded from disk)
+    return {"progress": progress}
+
+
+def _finalize_job(job_id: str, out_dir: str) -> None:
+    """Background finalization: the ONE place the final report email may fire."""
+    try:
+        report = finalize_review(job_id, out_dir)
+    except Exception as e:
+        try:
+            _set(job_id, status="finalization_error", error=str(e),
+                 finished_at=time.time(), stage=None)
+        except KeyError:
+            pass  # job not in the in-memory registry
+        return
+    try:
+        _set(job_id, status="done", report=report,
+             finished_at=time.time(), stage=None)
+    except KeyError:
+        pass
+    try:
+        backend.jobs._notify(job_id, report, out_dir)
+    except Exception as e:
+        print(f"[Jobs] Final notification for {job_id} failed: {e}")
+
+
+@app.post("/jobs/{job_id}/finalize-review")
+def finalize_review_endpoint(job_id: str):
+    job = get_job(job_id)
+    if not job:
+        raise HTTPException(status_code=404, detail="Job not found")
+    if job.get("status") in ("done", "finalizing"):
+        raise HTTPException(status_code=409, detail={
+            "detail": f"job is already {job['status']}",
+        })
+    if job.get("status") not in ("needs_review", "reviewing", "finalization_error"):
+        # Positive state-machine guard: finalization (and the final email) is
+        # only reachable after the job has passed through the review gate.
+        raise HTTPException(status_code=409, detail={
+            "detail": f"cannot finalize a job in status {job.get('status')}",
+        })
+    out_dir = job.get("out_dir") or os.path.join(config.OUTPUT_DIR, job_id)
+    store = ReviewStore(out_dir)
+    queue = store.read_queue()
+    decisions = store.read_decisions()
+    if any(item.get("blocking") and item.get("review_item_id") not in decisions
+           for item in queue):
+        # 409 detail shape: {"detail": <message>, "progress": <store.progress()>}
+        raise HTTPException(status_code=409, detail={
+            "detail": "incomplete review",
+            "progress": store.progress(queue),
+        })
+    try:
+        _set(job_id, status="finalizing")
+    except KeyError:
+        pass  # job not in the in-memory registry (e.g. loaded from disk)
+    threading.Thread(target=_finalize_job, args=(job_id, out_dir), daemon=True).start()
+    return {"status": "finalizing"}
+
+
 @app.get("/jobs/{job_id}/sheet-image/{page}")
 def sheet_image(job_id: str, page: int):
     """Render one page of a completed job's source PDF as JPEG (sheet viewer)."""

backend/review/__init__.py

@@ -0,0 +1 @@
+"""Human-review gate: decision schemas and review-trigger policy."""

backend/review/feedback.py

@@ -0,0 +1,51 @@
+"""Feedback labels: one label artifact per human-review decision, for metrics."""
+
+import json
+import os
+from datetime import datetime, timezone
+
+
+def _as_dict(value) -> dict:
+    return value if isinstance(value, dict) else {}
+
+
+def decision_to_label(queue_item: dict, decision: dict, job: dict) -> dict:
+    """Build one feedback label from a queue item, its decision, and the job.
+
+    All field access is defensive: missing fields degrade to None (or [] for
+    models_used) rather than raising.
+    """
+    queue_item = _as_dict(queue_item)
+    decision = _as_dict(decision)
+    job = _as_dict(job)
+    payload = _as_dict(queue_item.get("payload"))
+    summary = _as_dict(_as_dict(job.get("report")).get("summary"))
+    return {
+        "review_item_id": queue_item.get("review_item_id"),
+        "job_id": job.get("job_id"),
+        "pipeline_mode": job.get("pipeline_mode"),
+        "source_stage": payload.get("source_stage"),
+        "category": payload.get("category"),
+        "severity": payload.get("severity"),
+        "confidence": payload.get("confidence"),
+        "decision": decision.get("decision"),
+        "reason_code": decision.get("reason_code"),
+        "location": payload.get("location"),
+        "disciplines": payload.get("disciplines"),
+        "sheets": payload.get("sheets"),
+        "drawing_type": payload.get("drawing_type"),
+        "models_used": summary.get("models_used") or [],
+        "created_at": datetime.now(timezone.utc).isoformat(),
+    }
+
+
+def write_label(out_dir: str, label: dict) -> None:
+    """Append one label as a JSON line; never raises on I/O failure."""
+    try:
+        review_dir = os.path.join(out_dir, "review")
+        os.makedirs(review_dir, exist_ok=True)
+        path = os.path.join(review_dir, "feedback_labels.jsonl")
+        with open(path, "a", encoding="utf-8") as f:
+            f.write(json.dumps(label) + "\n")
+    except OSError as e:
+        print(f"[Review] feedback label write failed: {e}")

backend/review/finalizer.py

@@ -0,0 +1,255 @@
+"""ReviewFinalizer: apply human decisions, targeted reruns, RFIs, final artifacts.
+
+All LLM-touching helpers degrade gracefully: a failed or empty targeted rerun
+becomes a visible ``analysis_gap`` finding instead of raising, and RFI drafting
+returns whatever was produced (possibly []). Finalization never crashes the job
+on a single bad scope.
+"""
+
+import json
+import os
+from typing import Dict, List, Optional, Tuple
+
+from backend import config
+from backend.agents.base import AgentScope, AgentUsage
+from backend.agents.conflict_critic import ConflictCriticAgent
+from backend.agents.memory import ProjectMemory
+from backend.agents.orchestrator import Orchestrator
+from backend.agents.rfi_writer import RFIWriterAgent
+# Private import, acceptable here: the runner's _finding_as_conflict is the
+# canonical finding -> report["conflicts"] mapping; reusing it keeps the
+# finalized report's conflicts in exactly the shape build_report produces.
+from backend.agents.runner import _finding_as_conflict
+from backend.pipeline.report import to_markdown
+from backend.review.store import ReviewStore
+
+
+def apply_decisions(prioritized: List[dict], decisions: Dict[str, dict]) -> Tuple[List[dict], List[dict]]:
+    kept: List[dict] = []
+    suppressed: List[dict] = []
+    for issue in prioritized:
+        review_id = f"finding:{issue.get('issue_id')}"
+        decision = decisions.get(review_id) or {}
+        action = decision.get("decision")
+        if action == "reject":
+            suppressed.append({
+                **issue,
+                "review_state": "rejected",
+                "reason_code": decision.get("reason_code"),
+                "review_comment": decision.get("comment") or "",
+            })
+        elif action == "unsure":
+            kept.append({**issue, "review_state": "unsure"})
+        else:
+            kept.append({**issue, "review_state": "confirmed" if action == "confirm" else "unreviewed"})
+    return kept, suppressed
+
+
+def _gap_finding(index: int, scope_id: str, description: str) -> dict:
+    """Same shape as the runner's gap_findings: low severity, high confidence."""
+    return {
+        "issue_id": f"AGENT-GAP-CLARIFY-{index + 1:03d}",
+        "source_stage": "qaqc",
+        "category": "analysis_gap",
+        "severity": "low",
+        "confidence": "high",
+        "location": scope_id.split(":", 2)[1] if ":" in scope_id else "",
+        "disciplines": [],
+        "sheets": [],
+        "description": description,
+        "evidence": [],
+        "recommended_resolution": "Review this scope manually or rerun the job.",
+        "code_reference": None,
+        "agent": "completeness",
+        "scope_id": scope_id,
+    }
+
+
+def rerun_clarified_scopes(
+    memory_snapshot: dict,
+    decisions: Dict[str, dict],
+    prioritized: Optional[List[dict]] = None,
+) -> List[dict]:
+    """Bounded targeted reruns for ``needs_clarification`` decisions.
+
+    v1 reruns conflict scopes only: at most ONE ConflictCriticAgent scope per
+    clarified finding. The clarification answer is injected as a pseudo
+    "Reviewer" assertion prepended to the cluster's assertions so it reaches
+    the critic's evidence block (and survives front-truncation to
+    AGENT_CLUSTER_MAX_ASSERTIONS); ``page_to_b64`` is empty (cluster
+    assertions may carry their own
+    base64). Every per-scope failure degrades to an ``analysis_gap`` finding
+    and never raises. Non-conflict scopes are NOT rerun; they produce an
+    ``analysis_gap`` noting the scope is not rerunnable in v1.
+    """
+    findings_pool = list(prioritized or []) + list(memory_snapshot.get("findings") or [])
+    clusters = memory_snapshot.get("clusters") or []
+    out: List[dict] = []
+    for item_id, decision in (decisions or {}).items():
+        if (decision or {}).get("decision") != "needs_clarification":
+            continue
+        answer = str(decision.get("clarification_answer") or "").strip()
+        if not answer:
+            continue
+        issue_id = item_id.split(":", 1)[1] if item_id.startswith("finding:") else item_id
+        finding = next((f for f in findings_pool if f.get("issue_id") == issue_id), None)
+        scope_id = str((finding or {}).get("scope_id") or "")
+        if not scope_id.startswith("conflict:"):
+            out.append(_gap_finding(
+                len(out), scope_id or item_id,
+                f"Clarification rerun not supported in v1 for non-conflict scope "
+                f"{scope_id or item_id!r} (finding {issue_id}).",
+            ))
+            continue
+        cluster_key = scope_id.split(":", 1)[1]
+        cluster = next((c for c in clusters if c.get("key") == cluster_key), None)
+        if cluster is None:
+            out.append(_gap_finding(
+                len(out), scope_id,
+                f"Clarification rerun failed: cluster {cluster_key!r} not found "
+                f"for scope {scope_id} (finding {issue_id}).",
+            ))
+            continue
+        rerun_cluster = {
+            **cluster,
+            # Prepend: ConflictCriticAgent truncates assertions from the front
+            # (AGENT_CLUSTER_MAX_ASSERTIONS), so the clarification must come
+            # first or a full cluster would silently drop it.
+            "assertions": [{
+                "discipline": "Reviewer",
+                "sheet_number": "REVIEW",
+                "attribute": "clarification",
+                "value": answer,
+                "source_text": answer,
+            }] + list(cluster.get("assertions") or []),
+        }
+        scope = AgentScope(
+            scope_id=scope_id,
+            payload={"cluster": rerun_cluster, "page_to_b64": {}},
+        )
+        result = ConflictCriticAgent(AgentUsage()).run(scope)
+        if result.error or not result.artifacts:
+            out.append(_gap_finding(
+                len(out), scope_id,
+                f"Clarification rerun did not complete for scope {scope_id} "
+                f"(finding {issue_id}): {result.error or 'no findings produced'}.",
+            ))
+            continue
+        for rerun_finding in result.artifacts:
+            rerun_finding["clarification_of"] = issue_id
+            out.append(rerun_finding)
+    return out
+
+
+def _draft_rfis(kept: List[dict]) -> List[dict]:
+    """Draft RFIs for kept issues only, mirroring the runner's wave 7."""
+    orchestrator = Orchestrator(ProjectMemory())
+    scopes = [
+        AgentScope(
+            scope_id=f"rfi:{finding.get('issue_id') or index + 1}",
+            payload={"finding": finding},
+        )
+        for index, finding in enumerate(kept)
+    ]
+    try:
+        results = orchestrator.run_scopes(
+            RFIWriterAgent(AgentUsage()), scopes, config.AGENT_RFI_CONCURRENCY
+        )
+    except Exception:
+        return []
+    return [artifact for result in results for artifact in result.artifacts]
+
+
+def _read_json(path: str, default):
+    try:
+        with open(path, encoding="utf-8") as f:
+            return json.load(f)
+    except (OSError, json.JSONDecodeError):
+        return default
+
+
+def _dump(out_dir: str, name: str, value) -> None:
+    with open(os.path.join(out_dir, name), "w", encoding="utf-8") as f:
+        json.dump(value, f, indent=2)
+
+
+def finalize_review(job_id: str, out_dir: str) -> dict:
+    """Apply review decisions and write the final report artifacts.
+
+    Raises ValueError("incomplete review") if any blocking queue item lacks a
+    decision. Never raises for rerun/RFI degradation.
+    """
+    store = ReviewStore(out_dir)
+    queue = store.read_queue()
+    decisions = store.read_decisions()
+    for item in queue:
+        if item.get("blocking") and item.get("review_item_id") not in decisions:
+            raise ValueError("incomplete review")
+
+    report = _read_json(os.path.join(out_dir, "conflicts.json"), {}) or {}
+    snapshot = _read_json(os.path.join(out_dir, "agent", "memory.json"), {}) or {}
+    prioritized = list(report.get("validated_issues") or [])
+
+    rerun_findings = rerun_clarified_scopes(snapshot, decisions, prioritized)
+    replacements: Dict[str, List[dict]] = {}
+    for finding in rerun_findings:
+        origin = finding.get("clarification_of")
+        if origin:
+            replacements.setdefault(origin, []).append(finding)
+        else:
+            prioritized.append(finding)  # gap findings stay as additions
+    for origin, new_findings in replacements.items():
+        for index, issue in enumerate(prioritized):
+            if issue.get("issue_id") == origin:
+                prioritized[index:index + 1] = new_findings
+                break
+
+    kept, suppressed = apply_decisions(prioritized, decisions)
+    for issue in kept:
+        if issue.get("clarification_of"):
+            issue["review_state"] = "clarified"
+        else:
+            decision = decisions.get(f"finding:{issue.get('issue_id')}") or {}
+            if decision.get("decision") == "needs_clarification":
+                issue["review_state"] = "clarification_failed"
+
+    rfis = _draft_rfis(kept)
+
+    report["validated_issues"] = kept
+    report["suppressed_issues"] = suppressed
+    report["rfis"] = rfis
+    summary = report.setdefault("summary", {})
+    summary["agent_status"] = "complete"
+    by_stage = summary.get("by_stage")
+    if isinstance(by_stage, dict):
+        if "validated" in by_stage:
+            by_stage["validated"] = len(kept)
+        if "rfis" in by_stage:
+            by_stage["rfis"] = len(rfis)
+    # Rebuild the conflicts view and headline counts from the KEPT
+    # conflict-stage findings so rejected findings no longer appear as
+    # conflicts in report.md / the UI (mirrors pipeline.report.build_report).
+    conflicts = [
+        _finding_as_conflict(finding)
+        for finding in kept
+        if finding.get("source_stage") == "conflict"
+    ]
+    report["conflicts"] = conflicts
+    by_severity = {"high": 0, "medium": 0, "low": 0}
+    by_category: Dict[str, int] = {}
+    for conflict in conflicts:
+        by_severity[conflict["severity"]] = by_severity.get(conflict["severity"], 0) + 1
+        by_category[conflict["category"]] = by_category.get(conflict["category"], 0) + 1
+    summary["conflicts_found"] = len(conflicts)
+    summary["by_severity"] = by_severity
+    summary["by_category"] = by_category
+    summary["review"] = store.progress(queue)
+
+    os.makedirs(out_dir, exist_ok=True)
+    _dump(out_dir, "conflicts.json", report)
+    _dump(out_dir, "validated_issues.json", kept)
+    _dump(out_dir, "suppressed_issues.json", suppressed)
+    _dump(out_dir, "rfis.json", rfis)
+    with open(os.path.join(out_dir, "report.md"), "w", encoding="utf-8") as f:
+        f.write(to_markdown(report))
+    return report

backend/review/gate.py

@@ -0,0 +1,31 @@
+"""ReviewGate: build the human-review queue from prioritized findings."""
+
+from typing import Dict, List, Optional
+
+from backend.review.policy import build_audit_sample, requires_review
+
+
+def _finding_item(issue: Dict, blocking: bool, reasons: List[str], kind: str) -> Dict:
+    issue_id = issue.get("issue_id") or "unknown"
+    return {
+        "review_item_id": f"finding:{issue_id}",
+        "kind": kind,
+        "blocking": blocking,
+        "reasons": reasons,
+        "payload": issue,
+    }
+
+
+def build_review_queue(memory_snapshot: Dict, prioritized: List[Dict], decisions: List[Dict],
+                       limit: Optional[int] = None) -> List[Dict]:
+    queue: List[Dict] = []
+    for issue in prioritized:
+        reasons = requires_review(issue)
+        queue.append(_finding_item(issue, bool(reasons), reasons, "finding" if reasons else "audit_finding"))
+    if limit is None:
+        for item in build_audit_sample(memory_snapshot, prioritized):
+            queue.append(item)
+    else:
+        for item in build_audit_sample(memory_snapshot, prioritized, limit=limit):
+            queue.append(item)
+    return queue

backend/review/metrics.py

@@ -0,0 +1,25 @@
+"""Aggregate metrics over feedback labels.
+
+Default aggregates exclude source_text, images, raw sheet content, and
+reviewer free-text comments; include_text=True is the only path that embeds
+the raw labels.
+"""
+
+from collections import Counter
+from typing import Dict, List
+
+
+def aggregate_labels(labels: List[dict], include_text: bool = False) -> Dict:
+    decisions = Counter(label.get("decision") or "unknown" for label in labels)
+    reasons = Counter(label.get("reason_code") or "none" for label in labels if label.get("decision") == "reject")
+    summary = {
+        "total": len(labels),
+        "decisions": dict(decisions),
+        "reject_reasons": dict(reasons),
+    }
+    for label in labels:
+        decision = label.get("decision") or "unknown"
+        summary[decision] = summary.get(decision, 0) + 1
+    if include_text:
+        summary["labels"] = labels
+    return summary

backend/review/policy.py

@@ -0,0 +1,70 @@
+"""Review-trigger policy: which findings block on human review."""
+
+from typing import Any, Dict, List
+
+_SENSITIVE_CATEGORIES = {
+    "missing_element",
+    "ada",
+    "tas_tdlr",
+    "egress",
+    "fire_separation",
+    "occupancy",
+    "spatial_clash",
+    "clearance_conflict",
+    "penetration_conflict",
+}
+
+
+def requires_review(issue: Dict) -> List[str]:
+    """Return trigger reasons that require human review for one issue."""
+    reasons: List[str] = []
+    severity = str(issue.get("severity") or "").lower()
+    confidence = str(issue.get("confidence") or "").lower()
+    category = str(issue.get("category") or "").lower()
+    if severity in {"critical", "high"}:
+        reasons.append("severity_high")
+    if confidence == "low":
+        reasons.append("confidence_low")
+    if category in _SENSITIVE_CATEGORIES or issue.get("source_stage") == "code":
+        reasons.append("sensitive_category")
+    return reasons
+
+
+def build_audit_sample(
+    memory_snapshot: Dict,
+    prioritized: List[Dict],
+    limit: int = 5,
+) -> List[Dict[str, Any]]:
+    """Build non-blocking spot-check items for clean (finding-free) clusters."""
+    implicated = {
+        str(finding.get("scope_id") or "")
+        for finding in (memory_snapshot.get("findings") or []) + list(prioritized)
+    }
+    items: List[Dict[str, Any]] = []
+    for cluster in memory_snapshot.get("clusters") or []:
+        if len(items) >= limit:
+            break
+        assertions = cluster.get("assertions") or []
+        if len(assertions) < 2:
+            continue
+        cluster_key = cluster.get("key") or "unknown"
+        if any(cluster_key in scope_id for scope_id in implicated):
+            continue
+        items.append({
+            "review_item_id": f"clean_cluster:{cluster_key}",
+            "kind": "clean_cluster",
+            "blocking": False,
+            "reasons": ["audit_sample"],
+            "payload": _without_base64(cluster),
+        })
+    return items
+
+
+def _without_base64(cluster: Dict) -> Dict:
+    return {
+        **cluster,
+        "assertions": [
+            {key: value for key, value in assertion.items() if key != "base64"}
+            for assertion in cluster.get("assertions") or []
+        ],
+    }

backend/review/schemas.py

@@ -0,0 +1,45 @@
+"""Human-review decision schema and validation."""
+
+from typing import Optional
+
+DECISIONS = {"confirm", "reject", "unsure", "needs_clarification"}
+REASON_CODES = {
+    "wrong_cluster_link",
+    "same_value_different_representation",
+    "not_a_contradiction",
+    "missing_evidence",
+    "extraction_misread",
+    "code_path_not_applicable",
+    "duplicate",
+    "severity_too_high",
+    "severity_too_low",
+    "other",
+}
+
+
+def validate_decision(raw: dict) -> Optional[dict]:
+    """Normalize a reviewer decision payload, or return None if invalid."""
+    if not isinstance(raw, dict):
+        return None
+    decision = str(raw.get("decision") or "").strip()
+    if decision not in DECISIONS:
+        return None
+    reason_code = raw.get("reason_code")
+    if decision == "reject":
+        reason_code = str(reason_code or "").strip()
+        if reason_code not in REASON_CODES:
+            return None
+    elif reason_code is not None:
+        reason_code = str(reason_code).strip() or None
+        if reason_code and reason_code not in REASON_CODES:
+            return None
+    return {
+        "review_item_id": str(raw.get("review_item_id") or "").strip(),
+        "decision": decision,
+        "reason_code": reason_code,
+        "category_correction": raw.get("category_correction"),
+        "severity_correction": raw.get("severity_correction"),
+        "comment": str(raw.get("comment") or "").strip(),
+        "clarification_answer": raw.get("clarification_answer"),
+        "reviewed_at": raw.get("reviewed_at"),
+    }

backend/review/store.py

@@ -0,0 +1,62 @@
+"""Persistence for human-review queue and decisions within a job output dir."""
+
+import json
+import os
+from typing import Dict, List
+
+from backend.review.schemas import validate_decision
+
+
+class ReviewStore:
+    def __init__(self, job_out_dir: str, create: bool = True) -> None:
+        self.review_dir = os.path.join(job_out_dir, "review")
+        if create:
+            os.makedirs(self.review_dir, exist_ok=True)
+
+    def _path(self, name: str) -> str:
+        return os.path.join(self.review_dir, name)
+
+    def _write_json(self, name: str, value) -> None:
+        path = self._path(name)
+        tmp = f"{path}.tmp"
+        with open(tmp, "w", encoding="utf-8") as f:
+            json.dump(value, f, indent=2)
+        os.replace(tmp, path)
+
+    def write_queue(self, queue: List[dict]) -> None:
+        self._write_json("review_queue.json", queue)
+
+    def read_queue(self) -> List[dict]:
+        try:
+            with open(self._path("review_queue.json"), encoding="utf-8") as f:
+                value = json.load(f)
+            return value if isinstance(value, list) else []
+        except (OSError, json.JSONDecodeError):
+            return []
+
+    def append_decision(self, decision: dict) -> None:
+        valid = validate_decision(decision)
+        if not valid or not valid["review_item_id"]:
+            raise ValueError("invalid review decision")
+        decisions = self.read_decisions()
+        decisions[valid["review_item_id"]] = valid
+        self._write_json("review_decisions.json", decisions)
+
+    def read_decisions(self) -> Dict[str, dict]:
+        try:
+            with open(self._path("review_decisions.json"), encoding="utf-8") as f:
+                value = json.load(f)
+            return value if isinstance(value, dict) else {}
+        except (OSError, json.JSONDecodeError):
+            return {}
+
+    def progress(self, queue: List[dict]) -> dict:
+        decisions = self.read_decisions()
+        required = [item for item in queue if item.get("blocking")]
+        completed = [item for item in required if item.get("review_item_id") in decisions]
+        return {
+            "required": len(required),
+            "completed": len(completed),
+            "remaining": len(required) - len(completed),
+            "total": len(queue),
+        }