feat: extraction coverage summary in report

backend/pipeline/report.py

@@ -1,5 +1,4 @@
-"""
+"""report.py - Stage 4: assemble the final report.
-report.py - Stage 4: assemble the final report.
 
 Produces a single JSON object (also the web API payload) and a human-readable
 Markdown summary grouped by severity.
@@ -8,6 +7,37 @@ Markdown summary grouped by severity.
 from typing import List, Dict
 from datetime import datetime, timezone
 
+from backend import config
+
+
+def _extraction_coverage(sheets: List[Dict]) -> Dict | None:
+    """Summarize per-sheet extraction coverage for the report summary.
+
+    Sheets may carry a ``coverage`` dict (``total_lines``/``covered_lines``/
+    ``ratio``) attached during wave-1 extraction. Older paths and scanned
+    pages have none; when no sheet is measured, return None so callers can
+    omit the key entirely.
+    """
+    measured = [s for s in sheets if isinstance(s.get("coverage"), dict)]
+    if not measured:
+        return None
+    floor = getattr(config, "EXTRACT_COVERAGE_FLOOR", 0.6)
+    return {
+        "pages_measured": len(measured),
+        "pages_below_floor": [
+            s.get("page_number") for s in measured
+            if s["coverage"].get("ratio", 0.0) < floor
+        ],
+        "fallback_pages": [
+            s.get("page_number") for s in measured
+            if any(a.get("grounding") == "text_layer_fallback"
+                   for a in s.get("assertions", []))
+        ],
+        "mean_ratio": round(
+            sum(s["coverage"].get("ratio", 0.0) for s in measured)
+            / len(measured), 3),
+    }
+
 
 def build_report(conflicts: List[Dict], sheets: List[Dict], clusters: List[Dict],
                  source: str = "") -> Dict:
@@ -19,10 +49,7 @@ def build_report(conflicts: List[Dict], sheets: List[Dict], clusters: List[Dict]
         by_cat[c["category"]] = by_cat.get(c["category"], 0) + 1
 
     disciplines = sorted({s["discipline"] for s in sheets if s.get("discipline")})
-    return {
+    summary = {
-        "source": source,
-        "generated_at": datetime.now(timezone.utc).isoformat(),
-        "summary": {
         "sheets_analyzed": len(sheets),
         "disciplines": disciplines,
         "assertions_extracted": sum(len(s.get("assertions", [])) for s in sheets),
@@ -30,7 +57,14 @@ def build_report(conflicts: List[Dict], sheets: List[Dict], clusters: List[Dict]
         "conflicts_found": len(conflicts),
         "by_severity": by_sev,
         "by_category": by_cat,
-        },
+    }
+    coverage = _extraction_coverage(sheets)
+    if coverage is not None:
+        summary["extraction_coverage"] = coverage
+    return {
+        "source": source,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "summary": summary,
         "conflicts": conflicts,
         "sheets": [
             {

tests/test_coverage_summary.py

@@ -0,0 +1,84 @@
+"""Unit tests for the extraction_coverage block in the report summary.
+
+Both the classic pipeline and the agent runner build their summary via
+backend.pipeline.report.build_report, so unit tests on that function cover
+every summary-producing path.
+"""
+
+from backend.pipeline.report import build_report
+
+
+def _sheet(page, coverage=None, assertions=None):
+    sheet = {
+        "page_number": page,
+        "sheet_number": f"S{page:03d}",
+        "discipline": "S",
+        "assertions": assertions if assertions is not None else [
+            {"text": "NOTE ALPHA", "object_type": "note"},
+            {"text": "NOTE BETA", "object_type": "note"},
+        ],
+    }
+    if coverage is not None:
+        sheet["coverage"] = coverage
+    return sheet
+
+
+def _cov(total, covered):
+    return {
+        "total_lines": total,
+        "covered_lines": covered,
+        "ratio": covered / total if total else 0.0,
+    }
+
+
+def test_extraction_coverage_omitted_without_coverage_data():
+    report = build_report(conflicts=[], sheets=[_sheet(1), _sheet(2)], clusters=[])
+    assert "extraction_coverage" not in report["summary"]
+
+
+def test_extraction_coverage_healthy():
+    sheets = [_sheet(1, _cov(100, 95)), _sheet(2, _cov(80, 76))]
+    report = build_report(conflicts=[], sheets=sheets, clusters=[])
+    cov = report["summary"]["extraction_coverage"]
+    assert cov["pages_measured"] == 2
+    assert cov["pages_below_floor"] == []
+    assert cov["fallback_pages"] == []
+    assert cov["mean_ratio"] == round((0.95 + 0.95) / 2, 3)
+
+
+def test_extraction_coverage_flags_below_floor():
+    sheets = [
+        _sheet(1, _cov(100, 95)),
+        _sheet(2, _cov(100, 40)),   # ratio 0.4 < 0.6 floor
+        _sheet(3, _cov(100, 59)),   # ratio 0.59 < 0.6 floor
+    ]
+    report = build_report(conflicts=[], sheets=sheets, clusters=[])
+    cov = report["summary"]["extraction_coverage"]
+    assert cov["pages_measured"] == 3
+    assert cov["pages_below_floor"] == [2, 3]
+    assert cov["mean_ratio"] == round((0.95 + 0.4 + 0.59) / 3, 3)
+
+
+def test_extraction_coverage_flags_fallback_pages():
+    fallback_assertions = [
+        {"text": "NOTE ALPHA", "object_type": "note"},
+        {"text": "NOTE BETA", "object_type": "note",
+         "grounding": "text_layer_fallback"},
+    ]
+    sheets = [
+        _sheet(1, _cov(100, 90), assertions=fallback_assertions),
+        _sheet(2, _cov(100, 90)),
+    ]
+    report = build_report(conflicts=[], sheets=sheets, clusters=[])
+    cov = report["summary"]["extraction_coverage"]
+    assert cov["fallback_pages"] == [1]
+
+
+def test_extraction_coverage_mixed_sheets_only_counts_measured():
+    # Sheet 2 has no coverage dict (e.g. scanned page / older path).
+    sheets = [_sheet(1, _cov(100, 50)), _sheet(2)]
+    report = build_report(conflicts=[], sheets=sheets, clusters=[])
+    cov = report["summary"]["extraction_coverage"]
+    assert cov["pages_measured"] == 1
+    assert cov["pages_below_floor"] == [1]
+    assert cov["mean_ratio"] == 0.5