fix: concurrent single-page calibration scan and null-page planner JSON

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

@@ -167,7 +167,7 @@ def run_calibration_phase1(
             )
             break
 
-    inspect_calls = sum(len(scan.rounds) for scan in scans)
+    inspect_calls = sum(len(scan.scanned_pages) for scan in scans)
     logger.info(
         "[calibration.phase1] region={} regimes={} offsets={} inspect_calls={}",
         region_index,

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

@@ -4,19 +4,57 @@
 
 from typing import Any
 
+# Multi-page batch verify (null-page / Phase-2 helpers).
 SECTION_START_ANSWER_KEYS = {
-    "found": "boolean, true only when the section heading starts on one of these pages",
+    "found": (
+        "boolean, true when that section starts on one of these pages "
+        "(exact heading or accepted variant)"
+    ),
     "found_page": "number|null, the physical page number where it starts",
 }
 
+# Single-page concurrent scan (calibration Phase-1 forward scan).
+SECTION_START_PAGE_ANSWER_KEYS = {
+    "found": (
+        "boolean, true when THIS page's main heading starts the given section "
+        "(exact or accepted variant)"
+    ),
+    "reason": "string, at most 20 Chinese characters explaining the decision",
+}
+
+
+def _variant_rules() -> str:
+    return (
+        "Count as a match when either:\n"
+        "(1) Exact: the page's main heading matches the given title as written; or\n"
+        "(2) Variant: the page's main heading is clearly the same section, "
+        "even if a number/letter prefix differs or a document-id/code suffix "
+        "appears on only one side "
+        "(e.g. given title '3.2 Foo' ↔ page heading 'Foo'; "
+        "given title 'Foo (DOC-12)' ↔ page heading 'Foo'; "
+        "given title 'A.1 Foo Bar' ↔ page heading 'Foo Bar').\n"
+        "Do NOT count: a contents-list line, a running header/footer, or "
+        "a passing mention inside ordinary body paragraphs."
+    )
+
 
 def build_section_start_question(title: str) -> str:
     """Ask whether ``title`` starts as a body heading on the provided pages."""
     return (
-        f"Does the section titled {title!r} START on one of these pages, as a "
-        "body heading? A table-of-contents line, a running header or footer, or "
-        "a passing mention in body text does not count. Report the physical page "
-        "number printed in the page label above each image."
+        f"Does the section titled {title!r} START on one of these pages as a "
+        "body heading (where that section begins)?\n"
+        f"{_variant_rules()}\n"
+        "Report the physical page number from the page label above each image."
+    )
+
+
+def build_section_start_page_question(title: str) -> str:
+    """Ask whether ``title`` starts as a body heading on this single page."""
+    return (
+        f"Does the section titled {title!r} START on THIS page as a "
+        "body heading (where that section begins)?\n"
+        f"{_variant_rules()}\n"
+        "Return found true or false, and a reason of at most 20 Chinese characters."
     )
 
 
@@ -36,3 +74,7 @@ def coerce_found_page(value: Any, *, pages: list[int]) -> int | None:
     except (TypeError, ValueError):
         return None
     return page if page in pages else None
+
+
+def coerce_reason(value: Any) -> str:
+    return str(value or "").strip().replace("\n", " ")

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

@@ -4,20 +4,24 @@
 some page after it. The scan walks forward from the candidate with a widening
 window, feeding each round's cursor into the next one, so a miss never re-opens
 pages that were already inspected.
+
+Each round still covers ``window_schedule[i]`` pages, but pages are inspected
+one-at-a-time concurrently (never batched into a single VLM call).
 """
 
 from __future__ import annotations
 
+from concurrent.futures import ThreadPoolExecutor, as_completed
 from dataclasses import dataclass, field
 from typing import Any
 
 from loguru import logger
 
 from app.services.document_agent.calibration.prompts import (
-    SECTION_START_ANSWER_KEYS,
-    build_section_start_question,
+    SECTION_START_PAGE_ANSWER_KEYS,
+    build_section_start_page_question,
     coerce_found,
-    coerce_found_page,
+    coerce_reason,
 )
 from app.services.document_agent.manifest import ToolContext
 from app.services.document_agent.tools.inspect_pages import inspect_pages
@@ -46,12 +50,21 @@ def progressive_page_windows(
     return windows
 
 
+@dataclass
+class PageInspectResult:
+    page: int
+    found: bool
+    reason: str = ""
+    error: str = ""
+
+
 @dataclass
 class ScanRound:
     pages: list[int]
     found: bool
     found_page: int | None = None
     error: str = ""
+    page_results: list[PageInspectResult] = field(default_factory=list)
 
 
 @dataclass
@@ -78,12 +91,78 @@ def to_dict(self) -> dict[str, Any]:
                     "found": item.found,
                     "found_page": item.found_page,
                     "error": item.error,
+                    "page_results": [
+                        {
+                            "page": pr.page,
+                            "found": pr.found,
+                            "reason": pr.reason,
+                            "error": pr.error,
+                        }
+                        for pr in item.page_results
+                    ],
                 }
                 for item in self.rounds
             ],
         }
 
 
+def _inspect_one_page(
+    *,
+    ctx: ToolContext,
+    title: str,
+    page: int,
+) -> PageInspectResult:
+    result = inspect_pages(
+        ctx,
+        {
+            "pages": [page],
+            "page_cap": 1,
+            "question": build_section_start_page_question(title),
+            "answer_keys": SECTION_START_PAGE_ANSWER_KEYS,
+            "folder_name": "calibration_scan",
+            "prefix": "scan",
+            "usage_task": "calibration.scan_title_forward",
+        },
+    )
+    if result.status != "ok":
+        return PageInspectResult(
+            page=page, found=False, error=result.error or "inspect.pages failed"
+        )
+    fields = (result.payload or {}).get("fields") or {}
+    return PageInspectResult(
+        page=page,
+        found=coerce_found(fields.get("found")),
+        reason=coerce_reason(fields.get("reason")),
+    )
+
+
+def _inspect_pages_concurrent(
+    *,
+    ctx: ToolContext,
+    title: str,
+    pages: list[int],
+) -> list[PageInspectResult]:
+    """Inspect each page alone; keep page order in the returned list."""
+    if len(pages) == 1:
+        return [_inspect_one_page(ctx=ctx, title=title, page=pages[0])]
+
+    by_page: dict[int, PageInspectResult] = {}
+    with ThreadPoolExecutor(max_workers=len(pages)) as pool:
+        futures = {
+            pool.submit(_inspect_one_page, ctx=ctx, title=title, page=page): page
+            for page in pages
+        }
+        for future in as_completed(futures):
+            page = futures[future]
+            try:
+                by_page[page] = future.result()
+            except Exception as exc:  # noqa: BLE001 — surface as page error
+                by_page[page] = PageInspectResult(
+                    page=page, found=False, error=str(exc)
+                )
+    return [by_page[page] for page in pages]
+
+
 def scan_title_forward(
     *,
     ctx: ToolContext,
@@ -94,8 +173,10 @@ def scan_title_forward(
 ) -> TitleScanResult:
     """Scan forward from ``start_page`` until the title is found or rounds run out.
 
-    Each round opens ``window_schedule[i]`` consecutive pages starting at the
-    cursor left by the previous round, so no page is inspected twice.
+    Each round covers ``window_schedule[i]`` consecutive pages starting at the
+    cursor left by the previous round. Pages inside a round are inspected
+    concurrently, one page per VLM call; the earliest true page wins. A page
+    error without a hit is logged and the scan continues to the next window.
     """
     scanned: list[int] = []
     rounds: list[ScanRound] = []
@@ -106,36 +187,21 @@ def scan_title_forward(
         end_page=page_count,
         window_schedule=window_schedule,
     ):
-        result = inspect_pages(
-            ctx,
-            {
-                "pages": pages,
-                "page_cap": len(pages),
-                "question": build_section_start_question(title),
-                "answer_keys": SECTION_START_ANSWER_KEYS,
-                "folder_name": "calibration_scan",
-                "prefix": "scan",
-                "usage_task": "calibration.scan_title_forward",
-            },
-        )
+        page_results = _inspect_pages_concurrent(ctx=ctx, title=title, pages=pages)
         next_start = pages[-1] + 1
         scanned.extend(pages)
 
-        if result.status != "ok":
-            rounds.append(ScanRound(pages=pages, found=False, error=result.error or ""))
-            logger.warning(
-                "[calibration.scan] title={!r} pages={} inspect failed: {}",
-                title,
-                pages,
-                result.error,
+        hits = [pr.page for pr in page_results if pr.found]
+        if hits:
+            found_page = min(hits)
+            rounds.append(
+                ScanRound(
+                    pages=pages,
+                    found=True,
+                    found_page=found_page,
+                    page_results=page_results,
+                )
             )
-            break
-
-        fields = (result.payload or {}).get("fields") or {}
-        found_page = coerce_found_page(fields.get("found_page"), pages=pages)
-        found = coerce_found(fields.get("found")) and found_page is not None
-        rounds.append(ScanRound(pages=pages, found=found, found_page=found_page))
-        if found:
             logger.info(
                 "[calibration.scan] title={!r} found on page={} after {} round(s)",
                 title,
@@ -151,6 +217,25 @@ def scan_title_forward(
                 rounds=rounds,
             )
 
+        errors = [pr.error for pr in page_results if pr.error]
+        if errors:
+            logger.warning(
+                "[calibration.scan] title={!r} pages={} inspect failed: {}; "
+                "continuing to next window",
+                title,
+                pages,
+                errors[0],
+            )
+        rounds.append(
+            ScanRound(
+                pages=pages,
+                found=False,
+                found_page=None,
+                error=errors[0] if errors else "",
+                page_results=page_results,
+            )
+        )
+
     logger.info(
         "[calibration.scan] title={!r} not found in pages={}",
         title,

apps/worker/app/services/document_agent/structure/null_page_react.py

@@ -30,56 +30,34 @@ def react_budget() -> int:
 
 
 _REACT_INSTRUCTIONS = """\
-You are the search planner in a small ReAct loop. Propose the next whole-line
-text query to find the physical START page of a section. Grep normalizes each
-PDF text line independently, collapses whitespace with CJK-aware spacing, and
-matches the complete normalized line case-insensitively. It never accepts a
-substring inside a longer line. Every candidate page, including a unique one,
-must pass visual section-start confirmation.
-
-The system already grepped the full TOC title once before this loop (see
-previous_attempts). Do not repeat that exact full-title query.
-
-Return one strict json object with action one of (no other keys):
-{"action":"grep","query":"..."}
-{"action":"strip_header","query":""}
-{"action":"strip_footer","query":""}
-{"action":"give_up","query":""}
-
-Do not include a reason field. Use give_up only when no useful untried query
-or strip remains.
-
-Ordered query strategy after the automatic full-title grep (follow this order;
-skip a step only if already tried or not applicable to the TOC title / parent
-path). Pattern-level only — do not invent document-specific titles:
-1. Remove the leading number / letter / punctuation prefix from the TOC title
-   and grep the remaining title body.
-2. When the parent path indicates appendices/annexes (or the TOC label is a
+Return exactly one strict json object and no other text.
+Fields: action is one of grep, strip_header, strip_footer, give_up; query is a string.
+
+Ordered query strategy (follow this order; skip a step only if already tried
+or not applicable to the given title / parent path). Pattern-level only —
+do not invent document-specific titles:
+1. Derive the search line from the given title by removing leading number /
+   letter / punctuation prefixes and trailing metadata qualifiers (document
+   identifiers/codes, revision labels, and similar). Keep the semantic title
+   body. Prefer that body over a metadata-only query when both are present.
+2. When the parent path indicates appendices/annexes (or the title is a
    lettered appendix-style entry): grep the structural form
-   "Appendix <letter>" using the letter taken from the TOC label. Prefer this
+   "Appendix <letter>" using the letter taken from the title. Prefer this
    before inventing other phrases.
 3. Only after the above: try other variants such as "Appendix <letter>" plus
    the title body, a shorter distinctive complete-line title variant, or
    another structural prefix (chapter / part / section / annex) when supported
    by the title or parent path.
-4. Prefer queries specific enough to avoid running headers and passing mentions.
-   Do not guess page numbers.
 
-Reflection rules (mandatory):
-- Read previous_attempts. Reflect on query, hit_page_count, and observation
-  before answering.
+Rules:
+- Prefer queries specific enough to avoid running headers and passing mentions.
+  Do not guess page numbers.
 - If the last observation is no_line_hits, visual_rejected,
   empty_normalized_query, grep_tool_error, or duplicate_normalized_query, you
-  MUST change the query when choosing grep. Emitting the same grep query again
-  after whitespace/case normalization is invalid for planner-chosen greps.
-- Prefer strip_header or strip_footer when candidate pages look like running
-  headers/footers. Each strip automatically re-greps the last query once for
-  free. Otherwise advance to the next ordered query strategy.
-- strip_header / strip_footer only update a temporary search view; they do not
-  change stored page text. They do NOT consume react_budget. Call each at most
-  once per locate.
-- Planner greps consume react_budget. The automatic full-title seed grep and
-  strip auto re-greps are free.
+  must choose a different normalized query when using grep.
+- Use strip_header or strip_footer at most once each when margin text causes
+  false hits; either action re-runs the last query.
+- Use give_up only when no useful untried query or strip action remains.
 """
 
 
@@ -195,7 +173,7 @@ def _propose_react_query(
     grep_loops_used: int,
 ) -> tuple[dict[str, Any] | None, dict[str, Any]]:
     state = {
-        "toc_title": title,
+        "title": title,
         "parent_path": list(parent_titles),
         "physical_search_scope": [left, right],
         "grep_loops_remaining": max(0, budget - grep_loops_used),