feat: deterministic sheet-list reconciliation (cover index vs identified sheets)

backend/agents/extractors.py

@@ -6,7 +6,11 @@ from typing import Dict
 from backend import config
 from backend.agents.base import AgentResult, AgentScope, AgentUsage, failure
 from backend.llm import call_json
-from backend.pipeline.extractor import _normalize_sheet, _text_layer_block
+from backend.pipeline.extractor import (
+    _normalize_sheet,
+    _text_layer_block,
+    discipline_from_sheet_number,
+)
 from backend.pipeline.sheet_index import _index_input
 from backend.prompts import (
     EXTRACTOR_SYSTEM_PROMPT,
@@ -60,12 +64,39 @@ class SheetExtractorAgent:
             reasoning_max_tokens=config.EXTRACT_REASONING_MAX_TOKENS or None,
         )
 
+    def _text_structuring_call(self, instruction_page: Dict, sheet_hint: str):
+        """Rung 2: text-only structuring pass over the page's text layer
+        (no image). Recovers text content the vision pass missed."""
+        from backend.prompts import (TEXT_STRUCTURING_SYSTEM_PROMPT,
+                                     TEXT_STRUCTURING_USER_INSTRUCTION)
+        instruction = (TEXT_STRUCTURING_USER_INSTRUCTION
+                       .replace("{sheet_hint}", str(sheet_hint or ""))
+                       .replace("{text_layer}",
+                                (instruction_page.get("text_layer") or "")
+                                [:config.TEXT_LAYER_MAX_CHARS]))
+        return call_json(
+            system_prompt=TEXT_STRUCTURING_SYSTEM_PROMPT,
+            user_text=instruction,
+            images_b64=None,
+            max_tokens=config.EXTRACT_MAX_TOKENS,
+            model=config.AGENT_EXTRACT_MODEL,
+            usage_tracker=self.usage,
+            usage_stage="agent.extract_text",
+            reasoning_effort=config.EXTRACT_REASONING_EFFORT or None,
+            reasoning_max_tokens=config.EXTRACT_REASONING_MAX_TOKENS or None,
+        )
+
     def run(self, scope: AgentScope) -> AgentResult:
+        from backend.text_coverage import (fallback_objects, merge_objects,
+                                           recover_sheet_number, text_coverage)
         try:
             page = scope.payload["page"]
+            hint = scope.payload.get("sheet_hint") or ""
+            page_text = page.get("text_layer")
             instruction = EXTRACTOR_USER_INSTRUCTION.replace(
-                "{sheet_hint}", str(scope.payload.get("sheet_hint") or "")
-            ) + _text_layer_block(page)
+                "{sheet_hint}", str(hint)) + _text_layer_block(page)
+
+            # Rung 1: vision pass (unchanged behaviour, incl. compact retry)
             parsed = _wrap_bare_list(self._call(instruction, page),
                                      page["page_number"])
             if not isinstance(parsed, dict):
@@ -78,9 +109,74 @@ class SheetExtractorAgent:
                     page["page_number"],
                 )
             if not isinstance(parsed, dict):
-                raise ValueError("no structured extraction returned")
+                # Don't give up on the page - the ladder below can still
+                # rescue it from the text layer.
+                parsed = {"sheet": {}, "objects": []}
+
             sheet = _normalize_sheet(parsed, page["page_number"],
-                                     page_text=page.get("text_layer"))
+                                     page_text=page_text)
+            cov = text_coverage(page_text or "", sheet["assertions"])
+            sheet["coverage"] = cov
+
+            # Rung 2: text-only structuring when coverage is below floor.
+            # MERGE, never replace: vision keeps every object it found
+            # (graphical_basis content exists only in the image); the text
+            # pass fills in the text content the vision pass missed.
+            if (page_text and config.EXTRACT_TEXT_RETRY_ENABLED
+                    and cov["ratio"] < config.EXTRACT_COVERAGE_FLOOR):
+                print(f"[Extract] Page {page['page_number']}: coverage "
+                      f"{cov['ratio']:.0%} < floor - text-only structuring pass")
+                parsed2 = _wrap_bare_list(
+                    self._text_structuring_call(page, hint), page["page_number"])
+                if isinstance(parsed2, dict):
+                    sheet2 = _normalize_sheet(parsed2, page["page_number"],
+                                              page_text=page_text)
+                    before = len(sheet["assertions"])
+                    sheet["assertions"] = merge_objects(sheet["assertions"],
+                                                        sheet2["assertions"])
+                    # Fill header gaps the vision pass left null
+                    for key in ("sheet_number", "sheet_title", "discipline",
+                                "level", "scale", "drawing_type"):
+                        if not sheet.get(key) and sheet2.get(key):
+                            sheet[key] = sheet2[key]
+                    cov = text_coverage(page_text, sheet["assertions"])
+                    sheet["coverage"] = cov
+                    print(f"[Extract] Page {page['page_number']}: merged "
+                          f"{len(sheet['assertions']) - before} text-structured "
+                          f"object(s), coverage now {cov['ratio']:.0%}")
+
+            # Rung 3: deterministic fallback - dark sheets are impossible.
+            # Also merged (deduped) so stub notes never double up with
+            # objects the earlier rungs already captured.
+            if (page_text and config.EXTRACT_FALLBACK_ENABLED
+                    and cov["ratio"] < config.EXTRACT_COVERAGE_FLOOR):
+                stubs = fallback_objects(page_text, page["page_number"],
+                                         config.EXTRACT_FALLBACK_MAX_OBJECTS)
+                stubs = _normalize_sheet({"sheet": {}, "objects": stubs},
+                                         page["page_number"],
+                                         page_text=page_text)["assertions"]
+                # _normalize_sheet only stamps its own "text_layer" rescue
+                # grounding; restore the explicit fallback provenance.
+                for stub in stubs:
+                    stub["grounding"] = "text_layer_fallback"
+                before = len(sheet["assertions"])
+                sheet["assertions"] = merge_objects(sheet["assertions"], stubs)
+                print(f"[Extract] Page {page['page_number']}: fallback merged "
+                      f"{len(sheet['assertions']) - before} text-layer stub(s)")
+                sheet["coverage"] = text_coverage(page_text,
+                                                  sheet["assertions"])
+
+            # Identity recovery: never leave a text-bearing page sheet-less
+            if not sheet.get("sheet_number") and page_text:
+                recovered = recover_sheet_number(page_text)
+                if recovered:
+                    sheet["sheet_number"] = recovered
+                    sheet["discipline"] = (
+                        discipline_from_sheet_number(recovered)
+                        or sheet.get("discipline") or "Unknown")
+                    print(f"[Extract] Page {page['page_number']}: sheet number "
+                          f"recovered from text layer -> {recovered}")
+
             return AgentResult(scope_id=scope.scope_id, artifacts=[sheet])
         except Exception as exc:
             return failure(scope, exc)

backend/agents/runner.py

@@ -31,6 +31,7 @@ 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
+from backend.sheet_reconcile import declared_sheet_list, reconcile_sheets
 from backend.text_layer import (
     attach_text_layers, coverage_gaps, find_evidence_bbox, render_crop,
 )
@@ -84,6 +85,20 @@ def run_agent_pipeline(
     sheets.sort(key=lambda sheet: sheet.get("page_number") or 0)
     memory.replace("sheets", sheets)
     memory.dump("01-extract.json")
+
+    # Deterministic reconciliation: the cover sheet's own sheet index
+    # declares what the set should contain; compare against what wave 1
+    # identified (catches missed sheets AND phantom/misread sheet numbers).
+    sheet_recon = reconcile_sheets(sheets, declared_sheet_list(page_to_text))
+    if sheet_recon["declared_total"]:
+        print(f"[SheetIndex] cover declares {sheet_recon['declared_total']} "
+              f"sheets; {sheet_recon['found_total']} identified in set")
+        if sheet_recon["declared_not_in_set"]:
+            print(f"[SheetIndex] declared but not in set: "
+                  f"{', '.join(sheet_recon['declared_not_in_set'][:20])}")
+        if sheet_recon["in_set_not_declared"]:
+            print(f"[SheetIndex] in set but not declared: "
+                  f"{', '.join(sheet_recon['in_set_not_declared'][:20])}")
     # Coverage signal: text layer present but extraction failed/empty reuses
     # the failed-scopes gap-finding path (finding built below wave 6).
     for gap_page in coverage_gaps(pages, sheets):
@@ -289,6 +304,7 @@ def run_agent_pipeline(
             "project_input": merged_input,
             "jurisdiction": jurisdiction,
             "sheet_index": sheet_index,
+            "sheet_reconciliation": sheet_recon,
             "project_intelligence": object_graph,
             "validated_issues": prioritized,
             "rfis": [],
@@ -365,6 +381,7 @@ def run_agent_pipeline(
         "project_input": merged_input,
         "jurisdiction": jurisdiction,
         "sheet_index": sheet_index,
+        "sheet_reconciliation": sheet_recon,
         "project_intelligence": object_graph,
         "validated_issues": prioritized,
         "rfis": rfis,

backend/config.py

@@ -112,6 +112,15 @@ EXTRACT_REASONING_EFFORT = os.getenv("EXTRACT_REASONING_EFFORT", "low").strip()
 # the budget into visible output. 0 disables -> falls back to the effort knob.
 # Mutually exclusive with effort when set (OpenRouter rejects both together).
 EXTRACT_REASONING_MAX_TOKENS = int(os.getenv("EXTRACT_REASONING_MAX_TOKENS", "2048"))
+# Coverage-driven extraction retry ladder. After the vision pass, the fraction
+# of meaningful text-layer lines represented in extracted objects is measured;
+# below EXTRACT_COVERAGE_FLOOR the page climbs the ladder: text-only
+# structuring pass (rung 2), then deterministic text-layer fallback stubs
+# (rung 3) so no text-bearing page goes dark.
+EXTRACT_COVERAGE_FLOOR = float(os.getenv("EXTRACT_COVERAGE_FLOOR", "0.6"))
+EXTRACT_TEXT_RETRY_ENABLED = os.getenv("EXTRACT_TEXT_RETRY_ENABLED", "true").lower() == "true"
+EXTRACT_FALLBACK_ENABLED = os.getenv("EXTRACT_FALLBACK_ENABLED", "true").lower() == "true"
+EXTRACT_FALLBACK_MAX_OBJECTS = int(os.getenv("EXTRACT_FALLBACK_MAX_OBJECTS", "200"))
 REASON_MAX_TOKENS  = int(os.getenv("REASON_MAX_TOKENS",  "4096"))
 
 # -- QAQC stage knobs (Stages 0-1, 3, 6-11) -------------------------

backend/pipeline/extractor.py

@@ -21,9 +21,18 @@ from backend.llm import call_json
 from backend.prompts import (
     EXTRACTOR_SYSTEM_PROMPT,
     EXTRACTOR_USER_INSTRUCTION,
+    TEXT_STRUCTURING_SYSTEM_PROMPT,
+    TEXT_STRUCTURING_USER_INSTRUCTION,
     DISCIPLINE_PREFIXES,
     ATTRIBUTE_VOCAB,
 )
+from backend.text_coverage import (
+    _norm,
+    fallback_objects,
+    merge_objects,
+    recover_sheet_number,
+    text_coverage,
+)
 
 # prefix (upper) -> discipline, longest-prefix-first for greedy matching
 _PREFIX_TO_DISCIPLINE = sorted(
@@ -164,6 +173,8 @@ def _normalize_sheet(parsed: Dict, page_number: int,
     clean: List[Dict] = []
     dropped = 0
     rescued = 0
+    unverified = 0
+    page_norm = _norm(page_text) if page_text else ""
 
     for idx, obj in enumerate(raw_objects):
         if not isinstance(obj, dict):
@@ -179,9 +190,19 @@ def _normalize_sheet(parsed: Dict, page_number: int,
                             page_text=page_text):
             dropped += 1
             continue
-        grounding = _grounding_stamp(primary_val, source_text, page_text)
-        if grounding:
+        # Pre-set stamps (fallback/merge rungs) win; otherwise compute the
+        # text-layer rescue stamp.
+        grounding = obj.get("grounding") or _grounding_stamp(
+            primary_val, source_text, page_text)
+        if grounding == "text_layer":
             rescued += 1
+        if not grounding and page_text and source_text:
+            # Vision-unverified: survived the digit guard, but the quoted
+            # source_text is not present in the deterministic text layer.
+            # Kept and stamped - the wave-5b verifier prioritizes these.
+            if _norm(str(source_text)) not in page_norm:
+                grounding = "vision_unverified"
+                unverified += 1
 
         # --- location_key: new schema is richer; map to legacy shape + extras ---
         lk = obj.get("location_key")
@@ -236,10 +257,11 @@ def _normalize_sheet(parsed: Dict, page_number: int,
             **({"grounding": grounding} if grounding else {}),
         })
 
-    if dropped or rescued:
+    if dropped or rescued or unverified:
         print(f"[Extract] Page {page_number} ({sheet_number}): "
               f"dropped {dropped} ungrounded object(s)"
-              + (f", rescued {rescued} via text layer" if rescued else ""))
+              + (f", rescued {rescued} via text layer" if rescued else "")
+              + (f", {unverified} vision-unverified" if unverified else ""))
 
     unresolved = parsed.get("unresolved_items") or []
 
@@ -270,7 +292,25 @@ def _text_layer_block(page: Dict) -> str:
             + text[:config.TEXT_LAYER_MAX_CHARS])
 
 
+def _text_structuring_extract(page: Dict, sheet_hint: str = ""):
+    """Rung 2 of the extraction ladder: text-only structuring call (no
+    image). The text layer is authoritative for alphanumeric content - the
+    model segments it instead of transcribing pixels, so vision misreads
+    are impossible on this rung."""
+    instruction = (TEXT_STRUCTURING_USER_INSTRUCTION
+                   .replace("{sheet_hint}", sheet_hint or "")
+                   .replace("{text_layer}",
+                            (page.get("text_layer") or "")
+                            [:config.TEXT_LAYER_MAX_CHARS]))
+    return call_json(
+        system_prompt=TEXT_STRUCTURING_SYSTEM_PROMPT,
+        user_text=instruction,
+        max_tokens=config.EXTRACT_MAX_TOKENS,
+    )
+
+
 def _extract_one(page: Dict, sheet_hint: str = "") -> Dict:
+    page_text = page.get("text_layer")
     user_text = (EXTRACTOR_USER_INSTRUCTION.replace("{sheet_hint}", sheet_hint)
                  + _text_layer_block(page))
     parsed = call_json(
@@ -280,17 +320,74 @@ def _extract_one(page: Dict, sheet_hint: str = "") -> Dict:
         max_tokens=config.EXTRACT_MAX_TOKENS,
     )
     if not isinstance(parsed, dict):
-        return {
-            "page_number": page["page_number"],
-            "sheet_number": None,
-            "discipline": "Unknown",
-            "sheet_title": f"Page {page['page_number']} (extraction failed)",
-            "level": None,
-            "scale": None,
-            "assertions": [],
-        }
-    return _normalize_sheet(parsed, page["page_number"],
-                            page_text=page.get("text_layer"))
+        if not page_text:
+            # Scanned/raster page: vision-only, keep the legacy failure shape.
+            return {
+                "page_number": page["page_number"],
+                "sheet_number": None,
+                "discipline": "Unknown",
+                "sheet_title": f"Page {page['page_number']} (extraction failed)",
+                "level": None,
+                "scale": None,
+                "assertions": [],
+            }
+        # Text-bearing page: climb the ladder instead of going dark.
+        parsed = {"sheet": {}, "objects": []}
+
+    sheet = _normalize_sheet(parsed, page["page_number"], page_text=page_text)
+    cov = text_coverage(page_text or "", sheet["assertions"])
+    sheet["coverage"] = cov
+
+    # Rung 2: text-only structuring when coverage is below floor. MERGE,
+    # never replace - vision objects (graphical_basis content exists only
+    # in the image) are kept; the text pass fills what vision missed.
+    if (page_text and config.EXTRACT_TEXT_RETRY_ENABLED
+            and cov["ratio"] < config.EXTRACT_COVERAGE_FLOOR):
+        print(f"[Extract] Page {page['page_number']}: coverage "
+              f"{cov['ratio']:.0%} < floor - text-only structuring pass")
+        parsed2 = _text_structuring_extract(page, sheet_hint)
+        if isinstance(parsed2, dict):
+            sheet2 = _normalize_sheet(parsed2, page["page_number"],
+                                      page_text=page_text)
+            before = len(sheet["assertions"])
+            sheet["assertions"] = merge_objects(sheet["assertions"],
+                                                sheet2["assertions"])
+            for key in ("sheet_number", "sheet_title", "discipline",
+                        "level", "scale", "drawing_type"):
+                if not sheet.get(key) and sheet2.get(key):
+                    sheet[key] = sheet2[key]
+            cov = text_coverage(page_text, sheet["assertions"])
+            sheet["coverage"] = cov
+            print(f"[Extract] Page {page['page_number']}: merged "
+                  f"{len(sheet['assertions']) - before} text-structured "
+                  f"object(s), coverage now {cov['ratio']:.0%}")
+
+    # Rung 3: deterministic fallback - a dark text-bearing sheet is
+    # impossible. Stubs are deduped against earlier rungs.
+    if (page_text and config.EXTRACT_FALLBACK_ENABLED
+            and cov["ratio"] < config.EXTRACT_COVERAGE_FLOOR):
+        stubs = fallback_objects(page_text, page["page_number"],
+                                 config.EXTRACT_FALLBACK_MAX_OBJECTS)
+        stubs = _normalize_sheet({"sheet": {}, "objects": stubs},
+                                 page["page_number"],
+                                 page_text=page_text)["assertions"]
+        before = len(sheet["assertions"])
+        sheet["assertions"] = merge_objects(sheet["assertions"], stubs)
+        print(f"[Extract] Page {page['page_number']}: fallback merged "
+              f"{len(sheet['assertions']) - before} text-layer stub(s)")
+        sheet["coverage"] = text_coverage(page_text, sheet["assertions"])
+
+    # Identity recovery: never leave a text-bearing page sheet-less.
+    if not sheet.get("sheet_number") and page_text:
+        recovered = recover_sheet_number(page_text)
+        if recovered:
+            sheet["sheet_number"] = recovered
+            sheet["discipline"] = (discipline_from_sheet_number(recovered)
+                                   or sheet.get("discipline") or "Unknown")
+            print(f"[Extract] Page {page['page_number']}: sheet number "
+                  f"recovered from text layer -> {recovered}")
+
+    return sheet
 
 
 def extract_assertions(pages: List[Dict], on_progress=None) -> List[Dict]:

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,18 +49,22 @@ 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")})
+    summary = {
+        "sheets_analyzed": len(sheets),
+        "disciplines": disciplines,
+        "assertions_extracted": sum(len(s.get("assertions", [])) for s in sheets),
+        "clusters_checked": len(clusters),
+        "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": {
-            "sheets_analyzed": len(sheets),
-            "disciplines": disciplines,
-            "assertions_extracted": sum(len(s.get("assertions", [])) for s in sheets),
-            "clusters_checked": len(clusters),
-            "conflicts_found": len(conflicts),
-            "by_severity": by_sev,
-            "by_category": by_cat,
-        },
+        "summary": summary,
         "conflicts": conflicts,
         "sheets": [
             {

backend/pipeline/runner.py

@@ -27,6 +27,7 @@ from typing import Dict, Optional, Callable
 
 from backend.pipeline.pdf_processor import convert_pdf_to_images
 from backend.pipeline.extractor import extract_assertions
+from backend.sheet_reconcile import declared_sheet_list, reconcile_sheets
 from backend.text_layer import attach_text_layers, coverage_gaps
 from backend.pipeline.sheet_index import classify_sheets, derive_project_meta_from_cover
 from backend.pipeline.jurisdiction import run_jurisdiction
@@ -111,6 +112,19 @@ def _run_stages(
     sheets = extract_assertions(pages)
     coverage_gaps(pages, sheets)  # classic: log-only recall signal
 
+    # Deterministic reconciliation: cover-sheet index vs identified sheets.
+    page_to_text = {p["page_number"]: p.get("text_layer") for p in pages}
+    sheet_recon = reconcile_sheets(sheets, declared_sheet_list(page_to_text))
+    if sheet_recon["declared_total"]:
+        print(f"[SheetIndex] cover declares {sheet_recon['declared_total']} "
+              f"sheets; {sheet_recon['found_total']} identified in set")
+        if sheet_recon["declared_not_in_set"]:
+            print(f"[SheetIndex] declared but not in set: "
+                  f"{', '.join(sheet_recon['declared_not_in_set'][:20])}")
+        if sheet_recon["in_set_not_declared"]:
+            print(f"[SheetIndex] in set but not declared: "
+                  f"{', '.join(sheet_recon['in_set_not_declared'][:20])}")
+
     stage("Classify sheet index")
     sheet_index = classify_sheets(sheets)
 
@@ -167,6 +181,7 @@ def _run_stages(
     report["project_input"] = merged_input
     report["jurisdiction"] = jurisdiction
     report["sheet_index"] = sheet_index
+    report["sheet_reconciliation"] = sheet_recon
     report["project_intelligence"] = project_intel
     report["validated_issues"] = prioritized
     report["rfis"] = rfis

backend/prompts.py

@@ -281,6 +281,30 @@ If the sheet has no extractable objects, return an empty objects array.
 Optional sheet hint: {sheet_hint}"""
 
 
+# ---------------------------------------------------------------------------
+# Stage 2b - text-only structuring (extraction retry ladder, rung 2)
+# ---------------------------------------------------------------------------
+
+TEXT_STRUCTURING_SYSTEM_PROMPT = """You are a construction document structuring engine.
+You receive the deterministic text layer extracted from one drawing sheet. It is complete and authoritative.
+Your ONLY job is to segment it into structured objects. You are NOT reading an image. You must NOT invent, complete, or correct any text.
+Rules:
+- Every numbered note, schedule row, callout, tag, legend entry, and title-block field becomes its own object.
+- source_text must be copied VERBATIM from the input, character-for-character. Never paraphrase.
+- Cover the ENTIRE input. Omitting a note is a failure. When unsure of an object's type, use general_note with confidence low.
+- Numbers, model numbers, dimensions, and tags must appear in source_text exactly as in the input.
+Respond only with valid JSON."""
+
+TEXT_STRUCTURING_USER_INSTRUCTION = """Segment this sheet's text layer into structured construction objects.
+Every note, schedule row, callout, tag, and title-block field in the text layer must become an object - omit nothing.
+Respond ONLY with a valid JSON object - no markdown fences:
+{ "sheet": { "sheet_number": "string or null", "sheet_title": "string or null", "discipline": "string or null", "drawing_type": "string or null", "level": "string or null", "scale": "string or null" }, "objects": [ { "object_id": "string", "object_type": "room | door | window | wall | finish | ceiling | dimension | grid | callout | keynote | general_note | equipment | plumbing_fixture | mechanical_equipment | electrical_device | lighting_fixture | structural_element | schedule_reference | symbol | abbreviation", "category": "architectural | structural | mechanical | electrical | plumbing | code | general", "tag": "string or null", "name": "string or null", "description": "string or null", "attributes": { "attribute_name": "attribute_value" }, "location_key": { "room_number": "string or null", "grid": "string or null", "detail_reference": "string or null" }, "source_text": "VERBATIM text copied from the input", "graphical_basis": null, "review_uses": [ "schedule_comparison", "cross_discipline_coordination", "code_review", "constructability_review" ], "confidence": "high | medium | low" } ], "unresolved_items": [] }
+Optional sheet hint: {sheet_hint}
+
+TEXT LAYER (segment ALL of it):
+{text_layer}"""
+
+
 # ---------------------------------------------------------------------------
 # Stage 3a - assertion normalization  (WIRED: normalizer.py)
 # ---------------------------------------------------------------------------

backend/sheet_reconcile.py

@@ -0,0 +1,93 @@
+"""sheet_reconcile.py - deterministic sheet-list reconciliation (no LLM).
+
+The cover sheet's own sheet index (SHEET LIST / DRAWING INDEX) declares which
+sheets the set is SUPPOSED to contain. Comparing that declaration against the
+sheets wave-1 actually identified answers two early questions:
+
+- declared_not_in_set: sheets the index lists but we didn't identify - dark
+  pages, misidentification, or disciplines genuinely absent from this PDF.
+- in_set_not_declared: sheet numbers we extracted that the index doesn't
+  list - misread title blocks or unlisted sheets.
+
+Deterministic complement to the LLM sheet_index stage, which can only infer
+from what extraction already found.
+"""
+
+import re
+from typing import Dict, List, Optional
+
+# Markers that introduce the drawing set's own sheet index on a cover page.
+_INDEX_MARKERS = (
+    "SHEET LIST",
+    "DRAWING INDEX",
+    "SHEET INDEX",
+    "DRAWING LIST",
+    "INDEX OF DRAWINGS",
+)
+
+# Sheet ids: 1-2 letters, optional hyphen, 2-3 digits, optional decimal suffix.
+# Covers S301, A102, LS101, C-001, C-001.1; excludes dates/project numbers
+# (pure digits) and member marks (W12X26 - letter after digits).
+_SHEET_TOKEN_RE = re.compile(r"\b([A-Z]{1,2}-?\d{2,3}(?:\.\d+)?)\b")
+
+# Only cover-front pages carry the set index.
+_MAX_INDEX_PAGE = 5
+
+
+def _normalize_id(sheet_id: str) -> str:
+    return (sheet_id or "").upper().replace("-", "").strip()
+
+
+def declared_sheet_list(page_texts: Dict[int, Optional[str]]) -> List[str]:
+    """Scrape the declared sheet list off the cover page's text layer.
+
+    page_texts: {page_number: text_layer_or_None}. Returns the ordered,
+    deduped list of declared sheet ids, or [] when no index marker exists.
+    Only the FIRST page containing a marker is parsed (later 'sheet list'
+    echoes in legends/schedules are ignored).
+    """
+    for page_number in sorted(page_texts):
+        if page_number > _MAX_INDEX_PAGE:
+            break
+        text = page_texts.get(page_number) or ""
+        upper = text.upper()
+        marker_at = -1
+        for marker in _INDEX_MARKERS:
+            marker_at = upper.find(marker)
+            if marker_at >= 0:
+                break
+        if marker_at < 0:
+            continue
+        section = text[marker_at:]
+        declared: List[str] = []
+        for token in _SHEET_TOKEN_RE.findall(section):
+            if token not in declared:
+                declared.append(token)
+        return declared
+    return []
+
+
+def reconcile_sheets(sheets: List[Dict], declared: List[str]) -> Dict:
+    """Compare extracted sheet_numbers against the declared index.
+
+    Comparison is hyphen/case-normalized; output lists keep the declared /
+    extracted originals.
+    """
+    found: List[str] = [str(s["sheet_number"]) for s in sheets or []
+                        if s.get("sheet_number")]
+    found_norm = {_normalize_id(n) for n in found}
+    declared_norm = {_normalize_id(n) for n in declared}
+
+    declared_not_in_set = [n for n in declared if _normalize_id(n) not in found_norm]
+    # Preserve extraction order, dedupe, keep originals.
+    in_set_not_declared: List[str] = []
+    for n in found:
+        if _normalize_id(n) not in declared_norm and n not in in_set_not_declared:
+            in_set_not_declared.append(n)
+
+    return {
+        "declared_total": len(declared),
+        "found_total": len(found),
+        "declared_not_in_set": declared_not_in_set,
+        "in_set_not_declared": in_set_not_declared,
+    }

backend/text_coverage.py

@@ -0,0 +1,135 @@
+"""text_coverage.py - deterministic extraction-coverage measurement.
+
+The coverage guarantee: for any page with a usable text layer, measure how
+much of that layer ended up represented in extracted objects. Pages below
+the floor route into the extraction retry ladder (agents/extractors.py and
+pipeline/extractor.py). fallback_objects() is the last rung: stub objects
+segmented straight from the text layer so no text-bearing page goes dark.
+"""
+
+import re
+from typing import Dict, List, Optional
+
+MIN_LINE_CHARS = 12
+_TICK_RE = re.compile(r"^[\d\s'\"/.,-]+$")
+_WORD_RE = re.compile(r"[a-z0-9]+")
+
+
+def _meaningful_lines(text: str) -> List[str]:
+    lines = []
+    for raw in (text or "").splitlines():
+        line = " ".join(raw.split())
+        if len(line) < MIN_LINE_CHARS or _TICK_RE.match(line):
+            continue
+        lines.append(line)
+    return lines
+
+
+def _norm(text: str) -> str:
+    return " ".join(_WORD_RE.findall((text or "").lower()))
+
+
+def text_coverage(page_text: str, objects: List[Dict]) -> Dict:
+    """Fraction of meaningful text-layer lines whose normalized form appears
+    in the concatenated normalized source_text of extracted objects."""
+    lines = _meaningful_lines(page_text)
+    if not lines:
+        return {"total_lines": 0, "covered_lines": 0, "ratio": 1.0}
+    haystack = " ".join(
+        _norm(str(o.get("source_text") or o.get("object_description")
+                  or o.get("value") or ""))
+        for o in objects if isinstance(o, dict)
+    )
+    covered = sum(1 for ln in lines if _norm(ln) and _norm(ln) in haystack)
+    return {
+        "total_lines": len(lines),
+        "covered_lines": covered,
+        "ratio": covered / len(lines) if lines else 1.0,
+    }
+
+
+def segment_text_layer(text: str) -> List[str]:
+    """Segment a page text layer into note-sized blocks."""
+    segments: List[str] = []
+    buf: List[str] = []
+    number_re = re.compile(r"^(\d{1,2}[.)]?|[A-Z]\d{0,2}[.)]?)\s*$")
+
+    def flush():
+        joined = " ".join(buf).strip()
+        if len(joined) >= MIN_LINE_CHARS:
+            segments.append(joined)
+        buf.clear()
+
+    for raw in (text or "").splitlines():
+        line = raw.strip()
+        if not line:
+            flush()
+            continue
+        if number_re.match(line):
+            flush()
+            buf.append(line.rstrip(".)"))
+            continue
+        buf.append(line)
+        if line.endswith(".") and len(" ".join(buf)) > 120:
+            flush()
+    flush()
+    return segments
+
+
+def fallback_objects(page_text: str, page_number: int,
+                     max_objects: int = 200) -> List[Dict]:
+    """Last-rung deterministic extraction: one stub object per text segment,
+    source_text verbatim from the text layer."""
+    objs = []
+    for idx, seg in enumerate(segment_text_layer(page_text)[:max_objects]):
+        objs.append({
+            "object_id": f"p{page_number}-tl{idx}",
+            "object_type": "general_note",
+            "category": "general",
+            "tag": None,
+            "name": seg[:80],
+            "description": seg,
+            "attributes": {},
+            "location_key": {},
+            "source_text": seg,
+            "graphical_basis": None,
+            "review_uses": ["code_review", "constructability_review"],
+            "confidence": "low",
+            "grounding": "text_layer_fallback",
+        })
+    return objs
+
+
+def merge_objects(vision_objs: List[Dict], text_objs: List[Dict]) -> List[Dict]:
+    """Union of vision and text-structured objects. Vision results come first
+    and are never dropped. Text objects are appended unless their normalized
+    source_text is already represented."""
+    merged = list(vision_objs or [])
+    seen = {_norm(str(o.get("source_text") or ""))
+            for o in merged if isinstance(o, dict)}
+    seen.discard("")
+    for obj in text_objs or []:
+        if not isinstance(obj, dict):
+            continue
+        key = _norm(str(obj.get("source_text") or ""))
+        if key and key in seen:
+            continue
+        seen.add(key)
+        merged.append(obj)
+    return merged
+
+
+_SHEET_ID_RE = re.compile(r"\b([A-Z]{1,2}\d{2,3}(?:\.\d+)?)\b")
+
+
+def recover_sheet_number(page_text: str) -> Optional[str]:
+    """Deterministic sheet id from the text layer: prefer candidates in the
+    last ~15% of the page (title block lives at the drawing edge)."""
+    text = page_text or ""
+    cands = _SHEET_ID_RE.findall(text)
+    if not cands:
+        return None
+    tail = text[int(len(text) * 0.85):]
+    for cand in reversed(_SHEET_ID_RE.findall(tail)):
+        return cand
+    return cands[0]

tests/agents/test_extraction_ladder.py

@@ -0,0 +1,86 @@
+from backend import config
+from backend.agents.base import AgentScope, AgentUsage
+from backend.agents.extractors import SheetExtractorAgent
+from backend.prompts import TEXT_STRUCTURING_SYSTEM_PROMPT, TEXT_STRUCTURING_USER_INSTRUCTION
+
+def test_text_structuring_prompt_demands_verbatim_and_completeness():
+    assert "verbatim" in TEXT_STRUCTURING_USER_INSTRUCTION.lower()
+    assert "every" in TEXT_STRUCTURING_USER_INSTRUCTION.lower()
+    assert "{text_layer}" in TEXT_STRUCTURING_USER_INSTRUCTION
+
+
+def _page(n=8, text="1. \nALL SAWN LUMBER IN CONTACT WITH SOIL TO BE SOUTHERN PINE, PRESSURE TREATED.\n2. \nROOF SHEATHING: 5/8\" PLYWOOD, C-D GRADE, STRUCTURAL I."):
+    return {"page_number": n, "base64": "AAAA", "text_layer": text}
+
+
+def _run(agent, page, hint=""):
+    scope = AgentScope(scope_id=f"sheet:{page['page_number']}",
+                       payload={"page": page, "sheet_hint": hint})
+    result = agent.run(scope)
+    assert not result.error, result.error
+    return result.artifacts[0]
+
+
+def test_ladder_falls_back_when_vision_returns_nothing(monkeypatch):
+    # vision pass returns 1 summary object that the guard drops;
+    # text-structuring disabled to exercise the deterministic rung
+    monkeypatch.setattr("backend.agents.extractors.call_json",
+                        lambda **kw: [{"name": "general notes", "value": "notes"}])
+    monkeypatch.setattr("backend.config.EXTRACT_TEXT_RETRY_ENABLED", False)
+    agent = SheetExtractorAgent(AgentUsage())
+    sheet = _run(agent, _page())
+    assert sheet["assertions"], "dark sheet must be impossible with fallback enabled"
+    assert all(a.get("grounding") == "text_layer_fallback" for a in sheet["assertions"])
+    assert sheet["coverage"]["ratio"] >= 0.6
+
+
+def test_ladder_merge_preserves_graphical_objects(monkeypatch):
+    # vision finds a graphical symbol; text rung adds notes.
+    # The graphical object MUST survive the merge.
+    calls = {"n": 0}
+    def fake_call_json(**kw):
+        calls["n"] += 1
+        if kw.get("images_b64"):  # vision pass
+            return {"sheet": {}, "objects": [
+                {"object_id": "g1", "object_type": "lighting_fixture",
+                 "name": "pendant at grid C-4", "source_text": None,
+                 "graphical_basis": "16in pendant symbol at grid C-4"}]}
+        return {"sheet": {}, "objects": [  # text-structuring pass
+            {"object_id": "t1", "object_type": "general_note",
+             "source_text": "ALL SAWN LUMBER IN CONTACT WITH SOIL TO BE SOUTHERN PINE, PRESSURE TREATED.",
+             "name": "lumber note"}]}
+    monkeypatch.setattr("backend.agents.extractors.call_json", fake_call_json)
+    agent = SheetExtractorAgent(AgentUsage())
+    sheet = _run(agent, _page())
+    assert calls["n"] >= 2, "text-structuring rung should have fired"
+    assert any(a.get("graphical_basis") for a in sheet["assertions"])
+    assert any("SAWN LUMBER" in (a.get("source_text") or "") for a in sheet["assertions"])
+
+
+def test_ladder_recovers_sheet_number_from_text_layer(monkeypatch):
+    monkeypatch.setattr(
+        "backend.agents.extractors.call_json",
+        lambda **kw: {"sheet": {}, "objects": [
+            {"object_id": "o1", "name": "RCP note",
+             "source_text": "GYP. BD. CEILING 8'-11 3/8\" A.F.F. TYP. FOR ALL STOREFRONT",
+             "attributes": {"height": "8'-11 3/8\""}}]})
+    agent = SheetExtractorAgent(AgentUsage())
+    sheet = _run(agent, _page(18, "REFLECTED CEILING PLAN\nGYP. BD. CEILING 8'-11 3/8\" A.F.F. TYP. FOR ALL STOREFRONT\nA102"))
+    assert sheet["sheet_number"] == "A102"
+
+
+def test_ladder_skips_retry_when_coverage_healthy(monkeypatch):
+    # vision covers every meaningful text-layer line -> no rung 2/3 calls
+    calls = {"n": 0}
+    def fake_call_json(**kw):
+        calls["n"] += 1
+        return {"sheet": {"sheet_number": "A101"}, "objects": [
+            {"object_id": "o1", "object_type": "general_note", "name": "lumber note",
+             "source_text": "ALL SAWN LUMBER IN CONTACT WITH SOIL TO BE SOUTHERN PINE, PRESSURE TREATED."},
+            {"object_id": "o2", "object_type": "general_note", "name": "sheathing note",
+             "source_text": "ROOF SHEATHING: 5/8\" PLYWOOD, C-D GRADE, STRUCTURAL I."}]}
+    monkeypatch.setattr("backend.agents.extractors.call_json", fake_call_json)
+    agent = SheetExtractorAgent(AgentUsage())
+    sheet = _run(agent, _page())
+    assert sheet["coverage"]["ratio"] >= config.EXTRACT_COVERAGE_FLOOR
+    assert calls["n"] == 1

tests/agents/test_sheet_extractor_fallback.py

@@ -68,9 +68,12 @@ def test_run_compact_retry_after_hard_failure():
     assert "COMPACT RETRY" in mock_call.call_args_list[1].kwargs["user_text"]
 
 
-def test_run_fails_only_after_both_attempts_miss():
+def test_run_returns_empty_sheet_after_both_attempts_miss():
     agent = SheetExtractorAgent(usage=AgentUsage())
     with patch("backend.agents.extractors.call_json", return_value=None) as mock_call:
         result = agent.run(_scope())
-    assert result.error == "no structured extraction returned"
+    # Coverage ladder: no text layer to rescue the page -> empty sheet,
+    # but no hard failure (the ladder replaced the old raise).
+    assert not result.error
+    assert result.artifacts[0]["assertions"] == []
     assert mock_call.call_count == 2

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

tests/test_extraction_ladder_classic.py

@@ -0,0 +1,98 @@
+"""Coverage-driven extraction retry ladder — classic path (pipeline/extractor.py)."""
+
+from backend import config
+from backend.pipeline import extractor
+
+PAGE_TEXT = ("1. \nALL SAWN LUMBER IN CONTACT WITH SOIL TO BE SOUTHERN PINE, "
+             "PRESSURE TREATED.\n2. \nROOF SHEATHING: 5/8\" PLYWOOD, C-D GRADE, "
+             "STRUCTURAL I.")
+
+
+def _page(n=8, text=PAGE_TEXT):
+    return {"page_number": n, "base64": "AAAA", "text_layer": text}
+
+
+def test_classic_fallback_when_vision_returns_nothing(monkeypatch):
+    """Vision pass returns an unusable bare list; text retry disabled ->
+    deterministic fallback stubs make a dark sheet impossible."""
+    monkeypatch.setattr(extractor, "call_json",
+                        lambda **kw: [{"name": "general notes", "value": "notes"}])
+    monkeypatch.setattr(config, "EXTRACT_TEXT_RETRY_ENABLED", False)
+    sheet = extractor._extract_one(_page())
+    assert sheet["assertions"], "dark sheet must be impossible with fallback enabled"
+    assert all(a.get("grounding") == "text_layer_fallback"
+               for a in sheet["assertions"])
+    assert sheet["coverage"]["ratio"] >= 0.6
+
+
+def test_classic_merge_preserves_graphical_objects(monkeypatch):
+    """Rung-2 merge must never drop vision-only graphical objects."""
+    def fake(**kw):
+        if kw.get("images_b64"):
+            return {"sheet": {}, "objects": [
+                {"object_id": "g1", "object_type": "lighting_fixture",
+                 "name": "pendant at grid C-4", "source_text": None,
+                 "graphical_basis": "16in pendant symbol at grid C-4"}]}
+        return {"sheet": {}, "objects": [
+            {"object_id": "t1", "object_type": "general_note",
+             "source_text": "ALL SAWN LUMBER IN CONTACT WITH SOIL TO BE "
+                            "SOUTHERN PINE, PRESSURE TREATED.",
+             "name": "lumber note"}]}
+
+    monkeypatch.setattr(extractor, "call_json", fake)
+    sheet = extractor._extract_one(_page())
+    assert any(a.get("graphical_basis") for a in sheet["assertions"])
+    assert any("SAWN LUMBER" in (a.get("source_text") or "")
+               for a in sheet["assertions"])
+
+
+def test_classic_recovers_sheet_number(monkeypatch):
+    text = ("REFLECTED CEILING PLAN\n"
+            "GYP. BD. CEILING 8'-11 3/8\" A.F.F. TYP. FOR ALL STOREFRONT\n"
+            "LED TAPE LIGHT. SEE ELEC. SCONCE 8'-0\" A.F.F., SEE ELEC.\n"
+            "A102")
+
+    def fake(**kw):
+        if kw.get("images_b64"):
+            return {"sheet": {}, "objects": [
+                {"object_id": "o1", "name": "RCP ceiling note",
+                 "source_text": "GYP. BD. CEILING 8'-11 3/8\" A.F.F. TYP. "
+                                "FOR ALL STOREFRONT",
+                 "attributes": {"height": "8'-11 3/8\""}}]}
+        return {"sheet": {}, "objects": []}
+
+    monkeypatch.setattr(extractor, "call_json", fake)
+    sheet = extractor._extract_one(_page(18, text))
+    assert sheet["sheet_number"] == "A102"
+
+
+def test_classic_skips_retry_when_coverage_healthy(monkeypatch):
+    calls = []
+
+    def fake(**kw):
+        calls.append(kw)
+        return {"sheet": {"sheet_number": "S202"}, "objects": [
+            {"object_id": "o1", "name": "lumber note",
+             "source_text": "ALL SAWN LUMBER IN CONTACT WITH SOIL TO BE "
+                            "SOUTHERN PINE, PRESSURE TREATED.",
+             "attributes": {"species": "southern pine"}},
+            {"object_id": "o2", "name": "sheathing note",
+             "source_text": "ROOF SHEATHING: 5/8\" PLYWOOD, C-D GRADE, "
+                            "STRUCTURAL I.",
+             "attributes": {"sheathing": "5/8 plywood"}}]}
+
+    monkeypatch.setattr(extractor, "call_json", fake)
+    sheet = extractor._extract_one(_page())
+    assert len(calls) == 1, "healthy coverage must not trigger the text-only rung"
+    assert sheet["coverage"]["ratio"] == 1.0
+    assert sheet["sheet_number"] == "S202"
+
+
+def test_classic_scanned_page_keeps_failed_sheet_shape(monkeypatch):
+    """No text layer (scanned page): total parse failure keeps the existing
+    'extraction failed' empty-sheet return — ladder is text-layer-only."""
+    monkeypatch.setattr(extractor, "call_json", lambda **kw: None)
+    sheet = extractor._extract_one({"page_number": 4, "base64": "AAAA",
+                                    "text_layer": None})
+    assert sheet["assertions"] == []
+    assert "extraction failed" in (sheet.get("sheet_title") or "")

tests/test_extractor_text_grounding.py

@@ -84,3 +84,39 @@ def test_extractor_instruction_fully_substituted():
     out = (EXTRACTOR_USER_INSTRUCTION.replace("{sheet_hint}", "")
            + _text_layer_block(page))
     assert "{sheet_hint}" not in out
+
+
+def test_vision_unverified_stamp_when_source_text_not_in_text_layer():
+    """Digits ground the object against the page text, but its quoted
+    source_text is not actually present in the text layer: kept, stamped
+    vision_unverified (the wave-5b verifier consumes grounding stamps)."""
+    page_text = "WALL: 2X6 WD STUD @ 16\" O.C. WITH R-13 BATT INSULATION"
+    parsed = {"sheet": {}, "objects": [
+        {"object_id": "x1", "name": "stud pack",
+         "source_text": "(5) 2X6 STUD PACK AT JAMB",  # NOT in page text
+         "attributes": {"count": "5"}}]}
+    sheet = _normalize_sheet(parsed, 1, page_text=page_text)
+    assert len(sheet["assertions"]) == 1
+    assert sheet["assertions"][0]["grounding"] == "vision_unverified"
+
+
+def test_no_unverified_stamp_when_source_text_in_text_layer():
+    page_text = "WALL: 2X6 WD STUD @ 16\" O.C. WITH R-13 BATT INSULATION"
+    parsed = {"sheet": {}, "objects": [
+        {"object_id": "x1", "name": "stud note",
+         "source_text": "2X6 WD STUD @ 16\" O.C.",
+         "attributes": {"size": "2x6"}}]}
+    sheet = _normalize_sheet(parsed, 1, page_text=page_text)
+    assert len(sheet["assertions"]) == 1
+    assert "grounding" not in sheet["assertions"][0]
+
+
+def test_preset_grounding_stamp_survives_normalization():
+    """Fallback/merge rungs stamp grounding upstream; normalization must
+    preserve a pre-set stamp instead of recomputing it away."""
+    parsed = {"sheet": {}, "objects": [
+        {"object_id": "f1", "name": "lumber note",
+         "source_text": "ALL LUMBER SOUTHERN PINE",
+         "grounding": "text_layer_fallback"}]}
+    sheet = _normalize_sheet(parsed, 1, page_text="ALL LUMBER SOUTHERN PINE")
+    assert sheet["assertions"][0]["grounding"] == "text_layer_fallback"

tests/test_sheet_reconcile.py

@@ -0,0 +1,86 @@
+"""Deterministic sheet-list reconciliation: cover index vs extracted sheets."""
+
+from backend.sheet_reconcile import declared_sheet_list, reconcile_sheets
+
+COVER_TEXT = """VERIZON CYPRESS
+SHEET LIST
+SHEET NUMBER
+SHEET NAME
+G000
+COVER
+G001
+GENERAL INFO
+C-001
+CIVIL COVER
+C-001.1
+ALTA SURVEY
+L-101
+LANDSCAPE PLAN
+S101
+FOUNDATION PLAN
+S301
+WALL SECTIONS
+S401
+PERSPECTIVE VIEW
+A101
+FLOOR PLAN
+A102
+REFLECTED CEILING PLAN
+E400
+ELECTRICAL SITE PLAN
+"""
+
+
+def test_declared_sheet_list_from_cover():
+    declared = declared_sheet_list({1: COVER_TEXT, 2: "symbols legend"})
+    assert declared[0] == "G000"
+    assert "C-001" in declared and "C-001.1" in declared  # hyphenated ids kept
+    assert "L-101" in declared
+    assert "A102" in declared
+    assert declared.count("G000") == 1
+    assert len(declared) == 11
+
+
+def test_declared_sheet_list_uses_first_index_page_only():
+    texts = {1: "no index here", 2: COVER_TEXT, 3: "SHEET LIST\nXX999\nBOGUS"}
+    declared = declared_sheet_list(texts)
+    assert "XX999" not in declared  # only the first marker page is parsed
+
+
+def test_declared_sheet_list_none_when_no_marker():
+    assert declared_sheet_list({1: "just notes", 2: "floor plan stuff"}) == []
+
+
+def _sheets(*nums):
+    return [{"page_number": i + 1, "sheet_number": n}
+            for i, n in enumerate(nums)]
+
+
+def test_reconcile_both_directions():
+    declared = declared_sheet_list({1: COVER_TEXT})
+    rec = reconcile_sheets(_sheets("G000", "G001", "S101", "S301", "S302", "A101"),
+                           declared)
+    # declared but not extracted (civil/landscape not in this PDF + missing)
+    assert "C-001" in rec["declared_not_in_set"]
+    assert "A102" in rec["declared_not_in_set"]
+    assert "E400" in rec["declared_not_in_set"]
+    # extracted but not on the cover index (misread or unlisted sheet)
+    assert rec["in_set_not_declared"] == ["S302"]
+    assert rec["declared_total"] == 11
+    assert rec["found_total"] == 6
+
+
+def test_reconcile_normalizes_hyphens():
+    declared = ["C-001", "S301"]
+    rec = reconcile_sheets(_sheets("C001", "S301"), declared)
+    assert rec["declared_not_in_set"] == []
+    assert rec["in_set_not_declared"] == []
+
+
+def test_reconcile_ignores_unidentified_sheets():
+    rec = reconcile_sheets(
+        [{"page_number": 8, "sheet_number": None},
+         {"page_number": 9, "sheet_number": "S301"}],
+        ["S301", "A102"])
+    assert rec["found_total"] == 1
+    assert rec["declared_not_in_set"] == ["A102"]

tests/test_text_coverage.py

@@ -0,0 +1,76 @@
+from backend.text_coverage import (text_coverage, segment_text_layer,
+                                   fallback_objects, merge_objects,
+                                   recover_sheet_number)
+
+
+def test_coverage_full():
+    text = "NOTE 1\nALL LUMBER NO. 2 SOUTHERN PINE\nNOTE 2\nUSE 5/8\" PLYWOOD"
+    objects = [{"source_text": "ALL LUMBER NO. 2 SOUTHERN PINE"},
+               {"source_text": "USE 5/8\" PLYWOOD"}]
+    cov = text_coverage(text, objects)
+    assert cov["covered_lines"] == 2
+    assert cov["total_lines"] == 2
+    assert cov["ratio"] == 1.0
+
+
+def test_coverage_zero_on_empty_objects():
+    cov = text_coverage("LINE ALPHA CONTENT\nLINE BETA CONTENT\nLINE GAMMA CONTENT", [])
+    assert cov["ratio"] == 0.0 and cov["total_lines"] == 3
+
+
+def test_coverage_ignores_short_and_numeric_noise_lines():
+    text = "15\"\n19\"\nA\nB\nREAL NOTE ABOUT FRAMING HERE"
+    cov = text_coverage(text, [{"source_text": "REAL NOTE ABOUT FRAMING HERE"}])
+    assert cov["total_lines"] == 1 and cov["ratio"] == 1.0
+
+
+def test_segment_notes_and_rows():
+    text = "WOOD CONSTRUCTION\n1. \nALL SAWN LUMBER TO BE SOUTHERN PINE.\n2. \nROOF SHEATHING 5/8\" PLYWOOD."
+    segs = segment_text_layer(text)
+    assert any("ALL SAWN LUMBER" in s for s in segs)
+    assert any("ROOF SHEATHING" in s for s in segs)
+
+
+def test_fallback_objects_verbatim_and_stamped():
+    objs = fallback_objects("1. \nALL SAWN LUMBER TO BE SOUTHERN PINE.", page_number=8)
+    assert len(objs) == 1
+    assert objs[0]["source_text"] == "1 ALL SAWN LUMBER TO BE SOUTHERN PINE."
+    assert objs[0]["grounding"] == "text_layer_fallback"
+    assert objs[0]["confidence"] == "low"
+
+
+def test_merge_objects_keeps_vision_and_unions_text():
+    vision = [
+        {"source_text": "2X6 WD STUD @ 16\" O.C.", "object_type": "wall"},
+        {"source_text": None, "graphical_basis": "light fixture symbol, grid C-4",
+         "object_type": "lighting_fixture"},
+    ]
+    text = [
+        {"source_text": "2X6 WD STUD @ 16\" O.C.", "object_type": "wall"},
+        {"source_text": "ALL LUMBER NO. 2 SOUTHERN PINE", "object_type": "general_note"},
+    ]
+    merged = merge_objects(vision, text)
+    assert len(merged) == 3
+    assert any(o.get("graphical_basis") for o in merged)
+    assert merged[0]["object_type"] == "wall"
+
+
+def test_merge_objects_dedupes_by_normalized_text():
+    a = [{"source_text": "RTU-1: 5 TON, 1600 CFM"}]
+    b = [{"source_text": "rtu 1 5 ton 1600 cfm"}]
+    assert len(merge_objects(a, b)) == 1
+
+
+def test_recover_sheet_number_from_title_block():
+    text = ("WALL SECTIONS\n...\nSheet Information\nS301\n"
+            "Issue Date 05.29.26\nProject Number 25177")
+    assert recover_sheet_number(text) == "S301"
+
+
+def test_recover_sheet_number_none_when_absent():
+    assert recover_sheet_number("just some notes about lumber") is None
+
+
+def test_recover_prefers_discipline_pattern_over_dates():
+    text = "Issue Date 05.29.26\nProject Number 25177\nA102 REFLECTED CEILING PLAN"
+    assert recover_sheet_number(text) == "A102"