fix: null-page probe bounds and printed-page offset backfill

apps/worker/app/services/document_agent/calibration/phase1.py

@@ -64,7 +64,7 @@ def _regime_probes(
 ) -> dict[str, list[_Probe]]:
     """Keep leaf probes from the first distinct printed pages of each kind."""
     from app.services.document_parser.structure.body_boundary import (
-        normalize_heading_text,
+        normalize_heading_label,
     )
 
     probes: dict[str, list[_Probe]] = {}
@@ -82,7 +82,7 @@ def _regime_probes(
             continue
         if printed in seen_printed.get(kind, set()):
             continue
-        title = normalize_heading_text(str(entry.get("heading") or ""))
+        title = normalize_heading_label(str(entry.get("heading") or ""))
         if not title:
             continue
         probes.setdefault(kind, []).append(_Probe(title=title, printed=printed))

apps/worker/app/services/document_agent/calibration/procedure.py

@@ -4,7 +4,7 @@
 1. Builds TitleNodes the same way production does
 2. Runs Phase-2 **per regime** (prune → bulk/bisect → recalibrate)
 3. Merges physical-page ``match_overrides`` across regimes
-4. Runs null-page parent locate once on the combined tree
+4. Backfills printed-page nodes, runs unified null-page ReAct, then final prune
 
 Returns production ``SkeletonAnchor`` plus regime diagnostics for debug payloads.
 """
@@ -33,8 +33,7 @@
 )
 from app.services.document_agent.structure.anchoring_primitives import (
     SkeletonAnchor,
-    backfill_parent_offset_matches,
-    locate_null_page_parent_overrides,
+    apply_null_page_locates_and_prune,
     prune_unanchored_toc_leaves,
     serialize_skeleton_anchor,
 )
@@ -142,7 +141,7 @@ def _entry_titles_for_regime(
 ) -> set[str] | None:
     """Titles belonging to this regime; None means fall back to page_kind match."""
     from app.services.document_parser.structure.body_boundary import (
-        normalize_heading_text,
+        normalize_heading_label,
     )
 
     kind = normalize_kind(regime.kind)
@@ -160,7 +159,7 @@ def _entry_titles_for_regime(
         if idx < 0 or idx >= len(entries):
             continue
         heading = entries[idx].get("heading")
-        title = normalize_heading_text(str(heading or ""))
+        title = normalize_heading_label(str(heading or ""))
         if title:
             titles.add(title)
     return titles or None
@@ -350,9 +349,17 @@ def anchor_hierarchy_from_regimes(
                 len(regime_seed),
             )
 
-    # Failed suffix / never-confirmed printed leaves → drop from TOC tree.
+    # Capture parent identity before prune so empty shells retain ``kind=parent``.
+    structural_parent_paths = {
+        path
+        for path, node in _iter_all_title_nodes(working)
+        if node.children
+    }
+    # Failed printed-page leaves → drop; keep null-page nodes for ReAct.
     working, unanchored_removed = prune_unanchored_toc_leaves(
-        working, match_overrides=merged
+        working,
+        match_overrides=merged,
+        keep_null_page_nodes=True,
     )
     total_pruned += unanchored_removed
     if working:
@@ -366,26 +373,24 @@ def anchor_hierarchy_from_regimes(
             if path in surviving_paths
         }
 
-    parent_matches = backfill_parent_offset_matches(
-        nodes=working,
-        matches=merged,
-        page_count=page_count,
-    )
-    if parent_matches:
-        merged.update(parent_matches)
-        logger.info(
-            "[calibration.phase2] parent backfill: {} printed-page TOC parents "
-            "anchored from descendant offset",
-            len(parent_matches),
-        )
-
-    match_overrides, null_page_report = locate_null_page_parent_overrides(
+    (
+        working,
+        match_overrides,
+        null_page_report,
+        failed_null_removed,
+        pre_react_override_count,
+    ) = apply_null_page_locates_and_prune(
         nodes=working,
         match_overrides=merged,
-        page_texts=page_texts,
         body_pages=body_pages,
+        page_count=page_count,
         ctx=ctx,
+        structural_parent_paths=structural_parent_paths,
     )
+    total_pruned += failed_null_removed
+
+    # Freeze bulk after printed backfill and before ReAct (ReAct hits stay out).
+    bulk_count = pre_react_override_count if regime_bulk > 0 else 0
 
     primary = pick_primary_offset(result)
     if primary is None and usable_regimes:
@@ -401,7 +406,6 @@ def anchor_hierarchy_from_regimes(
         if match_overrides and (regime_bulk > 0 or seed)
         else "offset_only"
     )
-    bulk_count = len(match_overrides)
 
     return working, SkeletonAnchor(
         offset=primary,
@@ -445,10 +449,10 @@ def _annotate_regimes_from_anchor(
                 continue
             heading = str(entries[idx].get("heading") or "")
             from app.services.document_parser.structure.body_boundary import (
-                normalize_heading_text,
+                normalize_heading_label,
             )
 
-            title = normalize_heading_text(heading)
+            title = normalize_heading_label(heading)
             path = path_by_title.get(title)
             if path is not None and path in (anchor.match_overrides or {}):
                 ok_indices.append(idx)

apps/worker/app/services/document_agent/calibration/scan.py

@@ -25,6 +25,27 @@
 DEFAULT_WINDOW_SCHEDULE: tuple[int, ...] = (2, 4, 6, 10)
 
 
+def progressive_page_windows(
+    *,
+    start_page: int,
+    end_page: int,
+    window_schedule: tuple[int, ...] = DEFAULT_WINDOW_SCHEDULE,
+) -> list[list[int]]:
+    """Return non-overlapping physical-page windows clipped to ``end_page``."""
+    cursor = max(int(start_page), 1)
+    last_page = int(end_page)
+    windows: list[list[int]] = []
+    for size in window_schedule:
+        if cursor > last_page:
+            break
+        pages = list(range(cursor, min(cursor + int(size), last_page + 1)))
+        if not pages:
+            break
+        windows.append(pages)
+        cursor = pages[-1] + 1
+    return windows
+
+
 @dataclass
 class ScanRound:
     pages: list[int]
@@ -76,17 +97,15 @@ def scan_title_forward(
     Each round opens ``window_schedule[i]`` consecutive pages starting at the
     cursor left by the previous round, so no page is inspected twice.
     """
-    cursor = max(int(start_page), 1)
     scanned: list[int] = []
     rounds: list[ScanRound] = []
+    next_start = max(int(start_page), 1)
 
-    for size in window_schedule:
-        if cursor > page_count:
-            break
-        pages = [page for page in range(cursor, cursor + size) if page <= page_count]
-        if not pages:
-            break
-
+    for pages in progressive_page_windows(
+        start_page=start_page,
+        end_page=page_count,
+        window_schedule=window_schedule,
+    ):
         result = inspect_pages(
             ctx,
             {
@@ -99,7 +118,7 @@ def scan_title_forward(
                 "usage_task": "calibration.scan_title_forward",
             },
         )
-        cursor = pages[-1] + 1
+        next_start = pages[-1] + 1
         scanned.extend(pages)
 
         if result.status != "ok":
@@ -128,7 +147,7 @@ def scan_title_forward(
                 found=True,
                 found_page=found_page,
                 scanned_pages=scanned,
-                next_start=cursor if cursor <= page_count else None,
+                next_start=next_start if next_start <= page_count else None,
                 rounds=rounds,
             )
 
@@ -142,6 +161,6 @@ def scan_title_forward(
         found=False,
         found_page=None,
         scanned_pages=scanned,
-        next_start=cursor if cursor <= page_count else None,
+        next_start=next_start if next_start <= page_count else None,
         rounds=rounds,
     )

apps/worker/app/services/document_agent/calibration/service.py

@@ -16,6 +16,7 @@
     CalibrationResult,
 )
 from app.services.document_agent.manifest import ToolContext
+from app.services.document_agent.pdf_text import page_bands_map
 
 
 def calibrate_offset(
@@ -44,7 +45,7 @@ def calibrate_offset(
     if page_count and not ctx.blackboard.page_count:
         ctx.blackboard.page_count = int(page_count)
     if page_texts and not ctx.blackboard.page_full_text_cache:
-        ctx.blackboard.page_full_text_cache = dict(page_texts)
+        ctx.blackboard.page_full_text_cache = page_bands_map(page_texts)
 
     try:
         phase1 = run_calibration_phase1(

apps/worker/app/services/document_agent/coordinator.py

@@ -20,7 +20,7 @@
     ToolContext,
     ToolResult,
 )
-from app.services.document_agent.pdf_text import read_page_texts
+from app.services.document_agent.pdf_text import read_page_text_bands
 from app.services.document_agent.persist import build_anatomy_map, persist_anatomy_map
 from app.services.document_agent.coarse_profile import CoarseProfiler
 from app.services.document_agent.registry import REGISTRY
@@ -297,6 +297,7 @@ def _run_text_scan(self) -> None:
         pages = list(range(1, page_count + 1))
         if not pages:
             self.blackboard.page_full_text_cache = {}
+            self.blackboard.page_text_search_view = None
             return
         if profile.is_scanned:
             result = REGISTRY.dispatch("ocr.pages", self.ctx, {"pages": pages})
@@ -312,19 +313,30 @@ def _run_text_scan(self) -> None:
                 raise RuntimeError(result.error or "ocr.pages failed")
             self.round_index += 1
             return
-        texts = read_page_texts(self.ctx.pdf_path, pages, timeout=300)
-        self.blackboard.page_full_text_cache = texts
+        bands = read_page_text_bands(
+            self.ctx.pdf_path,
+            pages,
+            header_y=profile.header_y,
+            footer_y=profile.footer_y,
+            timeout=300,
+        )
+        self.blackboard.page_full_text_cache = bands
+        self.blackboard.page_text_search_view = None
         self.trace.record_step(
             round_index=self.round_index,
-            actor="scan:read_page_texts",
+            actor="scan:read_page_text_bands",
             action_type="scan",
             result=ToolResult(
                 status="ok",
-                payload={"page_count": len(texts)},
-                output_summary={"page_count": len(texts)},
+                payload={"page_count": len(bands)},
+                output_summary={"page_count": len(bands)},
             ),
-            tool_name="read_page_texts",
-            tool_args={"pages": pages},
+            tool_name="read_page_text_bands",
+            tool_args={
+                "pages": pages,
+                "header_y": profile.header_y,
+                "footer_y": profile.footer_y,
+            },
         )
         self.round_index += 1