feat: doc profile coarse margins and asset probing

.gitignore

@@ -70,6 +70,7 @@ test_*.csv
 
 # Local debugging scripts
 apps/worker/scripts/
+apps/worker/experiments/
 apps/worker/start_celery_worker.py
 apps/worker/start_celery_debug.sh
 apps/worker/clear_celery_queues.sh

apps/api/tests/contract/test_job_creation_contract.py

@@ -417,7 +417,7 @@ async def test_should_create_a_v2_page_memory_job_for_pdf_uploads(
     assert job_metadata["processing_generation"] == "page_memory"
     assert job_metadata["source_file_name"] == payload["file_name"]
     assert page_memory_config["max_pages"] == 1500
-    assert page_memory_config["asset_extraction_enabled"] is False
+    assert page_memory_config["asset_extraction_enabled"] is True
     assert original_request["file_name"] == payload["file_name"]
     assert "parse_track" not in original_request
 

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

@@ -1,29 +1,15 @@
 """Page anatomy agent for hierarchy-first PDF profiling."""
 
-from typing import TYPE_CHECKING
-
 from app.services.document_agent.manifest import (
     PageAnatomyMap,
     PageFeature,
     PageLabel,
     ShardPlan,
 )
 
-if TYPE_CHECKING:
-    from app.services.document_agent.profile_agent import ProfileAgent
-
-
-def __getattr__(name: str):
-    if name == "ProfileAgent":
-        from app.services.document_agent.profile_agent import ProfileAgent
-
-        return ProfileAgent
-    raise AttributeError(name)
-
 __all__ = [
     "PageAnatomyMap",
     "PageFeature",
     "PageLabel",
-    "ProfileAgent",
     "ShardPlan",
 ]

apps/worker/app/services/document_agent/bootstrap/__init__.py

@@ -2,6 +2,14 @@
 
 from app.services.document_agent.bootstrap.aggregate_stats import aggregate_doc_stats
 from app.services.document_agent.bootstrap.classify import classify_page_kinds
-from app.services.document_agent.bootstrap.probe import probe_page_features
+from app.services.document_agent.bootstrap.probe import (
+    probe_page_assets,
+    probe_page_features,
+)
 
-__all__ = ["aggregate_doc_stats", "classify_page_kinds", "probe_page_features"]
+__all__ = [
+    "aggregate_doc_stats",
+    "classify_page_kinds",
+    "probe_page_assets",
+    "probe_page_features",
+]

apps/worker/app/services/document_agent/bootstrap/aggregate_stats.py

@@ -8,28 +8,22 @@
 
 from app.services.document_agent.manifest import PageFeature, ToolContext, ToolResult
 
+# Coarse sampling extrema are text-only. Low text / density already proxy
+# chart-heavy or asset-heavy pages; table/drawing counts are not used to
+# nominate extrema pages for VLM coarse classification.
 PROFILE_METRICS = (
     "raw_text_length",
     "text_density",
-    "image_coverage",
-    "table_count",
-    "drawings_count",
 )
 
 EXTREMA_ROLES = {
     "raw_text_length": ("min", "max"),
     "text_density": ("min", "max"),
-    "image_coverage": ("max",),
-    "table_count": ("max",),
-    "drawings_count": ("max",),
 }
 
 EXTREMA_LABELS = {
     "raw_text_length": "text_length",
     "text_density": "text_density",
-    "image_coverage": "image_heavy",
-    "table_count": "table_heavy",
-    "drawings_count": "drawing_heavy",
 }
 
 
@@ -97,38 +91,30 @@ def aggregate_doc_stats(ctx: ToolContext, _args: dict[str, Any]) -> ToolResult:
 
     deduped_extrema = sorted(set(extrema_pages))
     landscape_pages = sum(1 for feature in features if feature.orientation == "landscape")
-    scan_like_pages = sum(
-        1
-        for feature in features
-        if feature.raw_text_length < 50 and feature.image_coverage >= 0.5
-    )
-    image_heavy_pages = sum(
-        1 for feature in features if feature.image_coverage >= 0.35
-    )
-    table_signal_pages = sum(
-        1
-        for feature in features
-        if feature.table_count > 0 or feature.drawings_count >= 25
-    )
-    doc_shape = {
+    doc_shape: dict[str, Any] = {
         "page_count": page_count,
         "landscape_pages": landscape_pages,
         "landscape_ratio": round(landscape_pages / page_count, 4)
         if page_count
         else 0.0,
-        "scan_like_pages": scan_like_pages,
-        "scan_like_ratio": round(scan_like_pages / page_count, 4)
-        if page_count
-        else 0.0,
-        "image_heavy_pages": image_heavy_pages,
-        "image_heavy_ratio": round(image_heavy_pages / page_count, 4)
-        if page_count
-        else 0.0,
-        "table_signal_pages": table_signal_pages,
-        "table_signal_ratio": round(table_signal_pages / page_count, 4)
-        if page_count
-        else 0.0,
     }
+    if ctx.blackboard.global_signals.get("assets_probed"):
+        scan_like_pages = sum(
+            1
+            for feature in features
+            if feature.raw_text_length < 50 and feature.image_coverage >= 0.5
+        )
+        asset_pages = sum(1 for feature in features if feature.has_asset)
+        doc_shape.update(
+            {
+                "scan_like_pages": scan_like_pages,
+                "scan_like_ratio": round(scan_like_pages / page_count, 4)
+                if page_count
+                else 0.0,
+                "asset_pages": asset_pages,
+                "asset_ratio": round(asset_pages / page_count, 4) if page_count else 0.0,
+            }
+        )
     ctx.blackboard.doc_stats = stats
     ctx.blackboard.extrema_pages = deduped_extrema
     ctx.blackboard.global_signals["doc_stats"] = stats

apps/worker/app/services/document_agent/bootstrap/probe.py

@@ -1,5 +1,8 @@
-"""Bootstrap wrapper for deterministic page probing."""
+"""Bootstrap wrappers for deterministic page probing."""
 
-from app.services.document_agent.tools.probe_page_features import probe_page_features
+from app.services.document_agent.tools.probe_page_features import (
+    probe_page_assets,
+    probe_page_features,
+)
 
-__all__ = ["probe_page_features"]
+__all__ = ["probe_page_assets", "probe_page_features"]

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

@@ -10,6 +10,7 @@
 from app.services.document_agent.bootstrap import (
     aggregate_doc_stats,
     classify_page_kinds,
+    probe_page_assets,
     probe_page_features,
 )
 from app.services.document_agent.budget import BudgetTracker, StageEnvelope
@@ -96,23 +97,16 @@ def __init__(
         self.round_index = 0
         self._planner_cache: tuple[DocumentProfile, Any, ToolResult] | None = None
 
-    def run(self) -> PageAnatomyMap:
-        try:
-            return self._run_structural()
-        except Exception as exc:
-            self._record_failure(exc)
-            raise
-
     def run_coarse(self) -> DocumentProfile:
         try:
             return self._run_coarse()
         except Exception as exc:
             self._record_failure(exc)
             raise
 
-    def run_structural(self) -> PageAnatomyMap:
+    def run_structural(self, *, skip_shard_plan: bool = False) -> PageAnatomyMap:
         try:
-            return self._run_structural()
+            return self._run_structural(skip_shard_plan=skip_shard_plan)
         except Exception as exc:
             self._record_failure(exc)
             raise
@@ -151,9 +145,10 @@ def _run_coarse(self) -> DocumentProfile:
         profile, _initial_decision, _planner_result = self._propose_profile(
             actor="planner:coarse"
         )
+        self._ensure_asset_probe()
         return profile
 
-    def _run_structural(self) -> PageAnatomyMap:
+    def _run_structural(self, *, skip_shard_plan: bool = False) -> PageAnatomyMap:
         self.state = DocumentAgentState.RUNNING
         if not self.blackboard.page_features:
             self._run_bootstrap()
@@ -164,14 +159,22 @@ def _run_structural(self) -> PageAnatomyMap:
         profile, initial_decision, _planner_result = self._propose_profile(
             actor="planner"
         )
-        executor_result = ReActExecutor(
-            self.ctx,
-            registry=REGISTRY,
-            max_rounds=int(self.ctx.settings.get("max_rounds", 30)),
-            initial_decision=initial_decision,
-        ).run()
-        if executor_result.verdict.status != "success":
-            raise RuntimeError(f"profile aborted: {executor_result.verdict.rationale}")
+        self._ensure_asset_probe()
+        if skip_shard_plan:
+            # Page-memory oversized path never consumes shard_plan; only
+            # build_anatomy_map's invariant needs a non-empty plan.
+            self._apply_single_shard_placeholder()
+        else:
+            executor_result = ReActExecutor(
+                self.ctx,
+                registry=REGISTRY,
+                max_rounds=int(self.ctx.settings.get("max_rounds", 30)),
+                initial_decision=initial_decision,
+            ).run()
+            if executor_result.verdict.status != "success":
+                raise RuntimeError(
+                    f"profile aborted: {executor_result.verdict.rationale}"
+                )
         anatomy = build_anatomy_map(self.ctx)
         self._persist_ready_anatomy(anatomy)
         return anatomy
@@ -200,6 +203,7 @@ def _run_lightweight_anatomy(
         self.state = DocumentAgentState.RUNNING
         if not self.blackboard.page_features:
             self._run_bootstrap()
+        self._ensure_asset_probe()
         if self.blackboard.toc_result is None:
             if self._toc_profile_enabled():
                 self.blackboard.toc_result = TocResult(
@@ -214,9 +218,7 @@ def _run_lightweight_anatomy(
             # it. Populate a single-shard placeholder to skip the LLM shard
             # decision + H2 refinement (kept global for chunk-track oversized
             # MinerU sharding).
-            self.blackboard.shard_plan = single_shard_plan(
-                self.blackboard.page_count
-            )
+            self._apply_single_shard_placeholder()
         else:
             result = REGISTRY.dispatch("propose.shard_plan", self.ctx, {})
             self.trace.record_step(
@@ -234,6 +236,9 @@ def _run_lightweight_anatomy(
         self._persist_ready_anatomy(anatomy)
         return anatomy
 
+    def _apply_single_shard_placeholder(self) -> None:
+        self.blackboard.shard_plan = single_shard_plan(self.blackboard.page_count)
+
     def _persist_ready_anatomy(self, anatomy: PageAnatomyMap) -> None:
         persist_result = persist_anatomy_map(self.ctx, {})
         self.trace.record_step(
@@ -284,6 +289,28 @@ def _run_bootstrap(self) -> None:
                 raise RuntimeError(result.error or f"{tool_name} failed")
             self.round_index += 1
 
+    def _ensure_asset_probe(self) -> None:
+        if self.blackboard.global_signals.get("assets_probed"):
+            return
+        if not self.blackboard.page_features:
+            raise RuntimeError("page_features missing; run text bootstrap first")
+        for tool_name, handler in (
+            ("probe.page_assets", probe_page_assets),
+            ("aggregate.doc_stats", aggregate_doc_stats),
+        ):
+            result = handler(self.ctx, {})
+            self.trace.record_step(
+                round_index=self.round_index,
+                actor=f"bootstrap:{tool_name}",
+                action_type="bootstrap",
+                result=result,
+                tool_name=tool_name,
+                tool_args={},
+            )
+            if result.status != "ok":
+                raise RuntimeError(result.error or f"{tool_name} failed")
+            self.round_index += 1
+
     def _toc_result_requires_strict_retry(self) -> bool:
         toc_result = self.blackboard.toc_result
         return bool(

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

@@ -7,7 +7,7 @@
 from typing import Any, Literal
 
 
-PageKind = Literal["normal", "table_heavy", "image_heavy", "low_content", "landscape"]
+PageKind = Literal["normal", "landscape"]
 TocFailureKind = Literal["none", "confirm_failed", "rejected_all", "degraded"]
 
 ReflexionAction = Literal["tool_call", "verdict_now"]
@@ -26,8 +26,10 @@ class PageFeature:
     orientation: Literal["portrait", "landscape"]
     width: float
     height: float
+    has_asset: bool
     is_blank_like: bool
-    text_lines_preview: list[str] = field(default_factory=list)
+    # PDF-space boxes for detected assets; None when none were extracted.
+    asset_bboxes: list[dict[str, Any]] | None = None
 
     def to_dict(self) -> dict[str, Any]:
         return asdict(self)
@@ -52,6 +54,10 @@ class DocumentProfile:
     category_rationale: str = ""
     language: str = "unknown"
     rationale: str = ""
+    # Content-band margins as fractions of page height (top origin, y down).
+    # header_y: lowest header line among sample pages; footer_y: highest footer.
+    header_y: float | None = None
+    footer_y: float | None = None
 
     def to_dict(self) -> dict[str, Any]:
         return asdict(self)
@@ -229,6 +235,7 @@ class PageAnatomyMap:
     def to_dict(self) -> dict[str, Any]:
         return {
             "version": self.version,
+            "toc_hierarchies": self.toc_hierarchies,
             "job_id": self.job_id,
             "file_path": self.file_path,
             "page_count": self.page_count,
@@ -240,7 +247,6 @@ def to_dict(self) -> dict[str, Any]:
             "document_profile": self.document_profile.to_dict()
             if self.document_profile
             else None,
-            "toc_hierarchies": self.toc_hierarchies,
             "toc_page_offset": self.toc_page_offset,
             "global_signals": dict(self.global_signals),
             "trace_summary": dict(self.trace_summary),