Staging

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

@@ -5,10 +5,15 @@
 and scans forward until one is confirmed; that single confirmation fixes the
 regime's candidate offset. Phase-2 owns tail verification and bulk anchoring.
 
-A probe scans from ``max(toc_range end + 1, printed)``: a printed label never
-resolves to a physical page before itself, so the offset a scan can yield is
-structurally non-negative. Scanning below the printed page would let a section
-divider that repeats the heading confirm ahead of the numbered body page.
+A probe scans from ``toc_range end + 1`` (body after the TOC). That can yield a
+negative ``physical - printed`` offset (e.g. journal reprints whose printed
+labels exceed the PDF length). Phase-2 already accepts negative offsets and
+prunes ``printed + offset`` outside ``1..page_count``.
+
+To restore the Sydney anti-false-positive floor (non-negative offset), switch
+``start_page`` back to ``max(region_scan_start, probe.printed)`` below and
+invert ``test_scan_starts_after_toc_so_negative_offset_is_allowed`` to expect
+scan start ``= printed`` and offset ``0``.
 """
 
 from __future__ import annotations
@@ -150,7 +155,10 @@ def run_calibration_phase1(
             scan = scan_title_forward(
                 ctx=ctx,
                 title=probe.title,
-                start_page=max(region_scan_start, probe.printed),
+                # Scan from after TOC. Optional floor (disabled): force
+                # non-negative offset / skip early divider hits — restore with:
+                #   start_page=max(region_scan_start, probe.printed),
+                start_page=region_scan_start,
                 page_count=resolved_page_count,
             )
             scans.append(scan)

apps/worker/app/services/document_agent/tools/extract_toc_with_boundaries.py

@@ -19,7 +19,6 @@
 )
 from app.services.document_agent.registry import register_tool
 from app.services.document_agent.tools.vlm_toc_extractor import (
-    TOC_VLM_MAX_TOKENS,
     BatchPageResult,
     vlm_entries_to_toc_hierarchies,
 )
@@ -35,6 +34,8 @@
 TOC_VLM_CONCURRENCY = 10
 MAX_BOUNDARY_ROUNDS = 6
 MAX_TOC_PAGES = BOUNDARY_STEP_PAGES * MAX_BOUNDARY_ROUNDS  # 30
+# Confirm only returns is_toc_start + brief reason per page (≤ BOUNDARY_STEP_PAGES).
+TOC_ANCHOR_CONFIRM_MAX_TOKENS = 512
 
 _CONFIRM_PROMPT = (
     "You are a document structure analysis expert. "
@@ -183,7 +184,7 @@ def _confirm_anchor_chunk(
             messages=messages,
             model=resolved,
             temperature=0.1,
-            max_tokens=TOC_VLM_MAX_TOKENS,
+            max_tokens=TOC_ANCHOR_CONFIRM_MAX_TOKENS,
             response_format={"type": "json_object"},
             usage_task="document_agent.toc_anchor_confirm",
         )

apps/worker/app/services/document_agent/tools/probe_page_features.py

@@ -29,6 +29,11 @@
 # Near-full-page sparse stroke frames are treated as borders, not figures.
 _FIGURE_FULLPAGE_AREA_RATIO = 0.92
 _FIGURE_FULLPAGE_MAX_PATHS = 25
+# Skip find_tables + drawing clustering on vector/image-dense pages.
+# Clustering ~1e5 strokes is quadratic in a crowded grid cell and exceeds
+# the 300s probe child timeout (EAS lipoprotein consensus PDF page 2).
+_DENSE_DRAWING_LIMIT = 5000
+_DENSE_IMAGE_LIMIT = 200
 # MuPDF ``page.get_texttrace()`` type matching PDF text rendering mode ``3 Tr``.
 _INVISIBLE_TEXT_TRACE_TYPE = 3
 
@@ -257,6 +262,10 @@ def _figure_bboxes_from_drawings(
     return figures
 
 
+def _is_dense_visual_page(*, image_count: int, drawings_count: int) -> bool:
+    return image_count > _DENSE_IMAGE_LIMIT or drawings_count > _DENSE_DRAWING_LIMIT
+
+
 def _probe_visual_assets(
     page: Any,
     page_area: float,
@@ -295,50 +304,54 @@ def _probe_visual_assets(
             image_area += max(box[2] - box[0], 0.0) * max(box[3] - box[1], 0.0)
             bboxes.append({"kind": "image", "bbox": box})
 
-    table_count = 0
-    try:
-        finder = page.find_tables()
-        tables = getattr(finder, "tables", []) or []
-        table_count = len(tables)
-        for table in tables:
-            raw = getattr(table, "bbox", None)
-            if raw is None:
-                continue
-            box = _clip_bbox(
-                float(raw[0]),
-                float(raw[1]),
-                float(raw[2]),
-                float(raw[3]),
-                clip=page_rect,
-            )
-            if _valid_bbox(box):
-                bboxes.append({"kind": "table", "bbox": box})
-    except Exception:
-        table_count = 0
-
     try:
         drawings = page.get_drawings() or []
     except Exception:
         drawings = []
     drawings_count = len(drawings)
-    bboxes.extend(
-        _figure_bboxes_from_drawings(
-            drawings,
-            page_rect=page_rect,
-            page_area=page_area,
-            header_y=header_y,
-            footer_y=footer_y,
-        )
+    dense_visual = _is_dense_visual_page(
+        image_count=image_count, drawings_count=drawings_count
     )
 
+    table_count = 0
+    if not dense_visual:
+        try:
+            finder = page.find_tables()
+            tables = getattr(finder, "tables", []) or []
+            table_count = len(tables)
+            for table in tables:
+                raw = getattr(table, "bbox", None)
+                if raw is None:
+                    continue
+                box = _clip_bbox(
+                    float(raw[0]),
+                    float(raw[1]),
+                    float(raw[2]),
+                    float(raw[3]),
+                    clip=page_rect,
+                )
+                if _valid_bbox(box):
+                    bboxes.append({"kind": "table", "bbox": box})
+        except Exception:
+            table_count = 0
+        bboxes.extend(
+            _figure_bboxes_from_drawings(
+                drawings,
+                page_rect=page_rect,
+                page_area=page_area,
+                header_y=header_y,
+                footer_y=footer_y,
+            )
+        )
+
     coverage = min(image_area / page_area, 1.0) if page_area > 0 else 0.0
     return {
         "image_coverage": round(coverage, 4),
         "image_count": image_count,
         "table_count": table_count,
         "drawings_count": drawings_count,
         # Geometry is only used to derive the gate; do not persist bboxes.
-        "has_asset": bool(bboxes),
+        "has_asset": bool(bboxes) or dense_visual,
     }
 
 

apps/worker/app/services/document_agent/tools/vlm_toc_extractor.py

@@ -8,8 +8,13 @@
 from dataclasses import dataclass
 from typing import Any, cast
 
-# Shared completion budget for TOC VLM calls (confirm batches + extract batches).
-TOC_VLM_MAX_TOKENS = 8192
+# Completion budget for TOC VLM calls: known pages in the call × per-page cap.
+TOC_VLM_MAX_TOKENS_PER_PAGE = 3000
+
+
+def toc_vlm_max_tokens(page_count: int) -> int:
+    """Return ``max_tokens`` for a TOC VLM call covering ``page_count`` pages."""
+    return max(1, int(page_count)) * TOC_VLM_MAX_TOKENS_PER_PAGE
 
 
 # ---------------------------------------------------------------------------
@@ -199,7 +204,7 @@ def vlm_extract_toc_batch(
         messages=cast(Any, [{"role": "user", "content": content_parts}]),
         model=model,
         temperature=0.1,
-        max_tokens=TOC_VLM_MAX_TOKENS,
+        max_tokens=toc_vlm_max_tokens(len(page_pngs)),
         response_format={"type": "json_object"},
         usage_task="document_agent.vlm_toc_batch",
     )

apps/worker/app/services/document_ingestion/processing_billing.py

@@ -4,7 +4,10 @@
 from datetime import datetime
 
 from app.services.document_ingestion.page_estimator import WorkloadEstimate
-from app.services.document_ingestion.processing_context import ParseJobContext
+from app.services.document_ingestion.processing_context import (
+    ParseJobContext,
+    persist_job_metadata_updates,
+)
 from loguru import logger
 from sqlalchemy import select
 
@@ -145,13 +148,8 @@ def record_processing_start(
         )
     if extra_metadata is not None:
         metadata_updates.update(extra_metadata)
-    with get_sync_db_context() as db:
-        job_result = db.execute(select(Job).where(Job.job_id == job_id).with_for_update())
-        job = job_result.scalar_one_or_none()
-        if job:
-            job.job_metadata = {
-                **dict(job.job_metadata or {}),
-                **metadata_updates,
-            }
-    job_context.metadata_service.update_metadata(job_id, metadata_updates)
-    job_context.job_metadata.update(metadata_updates)
+    persist_job_metadata_updates(
+        job_id=job_id,
+        job_context=job_context,
+        metadata_updates=metadata_updates,
+    )

apps/worker/app/services/document_ingestion/processing_context.py

@@ -27,6 +27,31 @@ class ParseJobContext:
     s3_key: str
 
 
+def persist_job_metadata_updates(
+    *,
+    job_id: str,
+    job_context: ParseJobContext,
+    metadata_updates: dict[str, object],
+) -> None:
+    """Merge metadata into Redis and the durable Job row.
+
+    Redis remains the live working copy during processing. The Job row is
+    the durable record after Redis TTL expires, so token usage and other
+    stage fields must be written here as well.
+    """
+    if not metadata_updates:
+        return
+    with get_sync_db_context() as db:
+        job = _select_job_row_for_update(db, job_id)
+        if job is not None:
+            job.job_metadata = {
+                **dict(job.job_metadata or {}),
+                **metadata_updates,
+            }
+    job_context.metadata_service.update_metadata(job_id, metadata_updates)
+    job_context.job_metadata.update(metadata_updates)
+
+
 def load_parse_job_context(
     job_id: str,
     requested_user_id: str | None,
@@ -111,3 +136,9 @@ def _load_job_row(job_id: str) -> Job | None:
 
 def _select_job_row(db: Session, job_id: str) -> Job | None:
     return db.execute(select(Job).where(Job.job_id == job_id)).scalar_one_or_none()
+
+
+def _select_job_row_for_update(db: Session, job_id: str) -> Job | None:
+    return db.execute(
+        select(Job).where(Job.job_id == job_id).with_for_update()
+    ).scalar_one_or_none()

apps/worker/app/services/document_ingestion/processing_run.py

@@ -17,6 +17,7 @@
 from app.services.document_ingestion.processing_context import (
     ParseJobContext,
     load_parse_job_context,
+    persist_job_metadata_updates,
 )
 from app.services.document_ingestion.source_preparation import prepare_source_file
 from app.services.document_ingestion.success_finalization import finalize_parse_success
@@ -199,10 +200,21 @@ def _run_parse_job(
             result_storage_factory=get_result_storage,
         )
     finally:
-        job_context.job_metadata["stages"] = {
+        stages = {
             "timing_ms": dict(stage_timing_dict),
             "token_usage": dict(token_usage_dict),
         }
+        job_context.job_metadata["stages"] = stages
+        try:
+            persist_job_metadata_updates(
+                job_id=job_id,
+                job_context=job_context,
+                metadata_updates={"stages": stages},
+            )
+        except Exception as exc:
+            logger.warning(
+                f"Failed to persist processing stages: job_id={job_id}, error={exc}"
+            )
         cleanup_llm_overrides()
         cleanup_token_tracker()
         cleanup_stage_tracker()

apps/worker/app/services/document_ingestion/success_finalization.py

@@ -18,7 +18,10 @@
     build_generated_result_package,
 )
 from app.services.document_ingestion.artifact_refs import collect_referenced_artifact_refs
-from app.services.document_ingestion.processing_context import ParseJobContext
+from app.services.document_ingestion.processing_context import (
+    ParseJobContext,
+    persist_job_metadata_updates,
+)
 from loguru import logger
 
 from shared.models.schemas.job_metadata import JobMetadataHelper
@@ -210,8 +213,11 @@ def _record_processing_completion(
     _refresh_processing_stages(job_context)
     if "stages" in job_context.job_metadata:
         processing_timing_updates["stages"] = job_context.job_metadata["stages"]
-    job_context.metadata_service.update_metadata(job_id, processing_timing_updates)
-    job_context.job_metadata.update(processing_timing_updates)
+    persist_job_metadata_updates(
+        job_id=job_id,
+        job_context=job_context,
+        metadata_updates=processing_timing_updates,
+    )
 
 
 def _generate_result_package(