chore: sync main into staging

apps/api/app/services/document_ingestion/service.py

@@ -48,7 +48,7 @@
 _PUBLIC_MODE_SELECTOR_FIELDS = {"mode", "processing"}
 _PARSE_TRACK_FIELD = "parse_track"
 _PAGE_MEMORY_FIELD_PREFIX = "page_memory"
-_PUBLIC_COMPATIBILITY_EXTRA_FIELDS = frozenset({_PARSE_TRACK_FIELD})
+_PUBLIC_COMPATIBILITY_EXTRA_FIELDS = frozenset({_PARSE_TRACK_FIELD, "result_mode"})
 IngestionCommandFactory = Callable[[str], DocumentIngestionCommand]
 
 

apps/worker/app/services/common/file_utils.py

@@ -4,6 +4,10 @@
 
 import pandas as pd
 
+# Shared cap for cosmetic asset filename stems (images/tables). Keeps OS
+# basename limits safe while preserving a short debug-friendly context.
+MAX_ASSET_FILE_NAME_CHARS = 80
+
 
 def clean_file(path_, mode="remove", cols=None):
     """

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

@@ -1,15 +1,15 @@
 """Prompts for executor reflexion."""
 
 REFLEXION_INSTRUCTIONS = (
-    "You are the executor of a document profiling agent. Decide the next action "
-    "from the blackboard facts and available tools. Return strict JSON with keys: "
-    "action (tool_call or verdict_now), rationale, optional tool_name/tool_args, "
-    "optional verdict {status, rationale}. Use inspect.pages when more visual "
-    "evidence is needed, grep.text when native-PDF text evidence is needed, "
-    "propose.shard_plan when evidence is sufficient to shard, validate.anatomy_map "
-    "after a shard plan exists, and verdict only after validation succeeds. If a "
-    "tool failed or validation is invalid, either gather targeted evidence and "
-    "retry the relevant tool or abort with a clear rationale."
+    "You are the executor of a document profiling agent. Decide the next tool "
+    "call from the blackboard facts and available tools. Return strict JSON with "
+    "keys: action (must be tool_call), rationale, tool_name, tool_args. "
+    "Use inspect.pages when more visual evidence is needed, grep.text when "
+    "native-PDF text evidence is needed, propose.shard_plan when evidence is "
+    "sufficient to shard, validate.anatomy_map after a shard plan exists, and "
+    "the verdict tool to finish: verdict(status=success) only after validation "
+    "succeeds, or verdict(status=abort, rationale=...) only when the document "
+    "cannot be profiled. Do not invent other finish actions."
 )
 
 __all__ = ["REFLEXION_INSTRUCTIONS"]

apps/worker/app/services/document_agent/executor/react_loop.py

@@ -56,27 +56,49 @@ def _compact_blackboard(ctx: ToolContext) -> dict[str, Any]:
     }
 
 
+def _coerce_legacy_finish(data: dict[str, Any]) -> ReflexionDecision:
+    """Map obsolete ``action=verdict_now`` into a real tool call.
+
+    Finish belongs exclusively to the ``verdict`` tool. A bare ``verdict_now``
+    without an explicit status is treated as ready-to-shard, never as abort.
+    """
+    rationale = str(data.get("rationale") or "")
+    raw_verdict = data.get("verdict")
+    if isinstance(raw_verdict, dict):
+        status = str(raw_verdict.get("status") or "").strip().lower()
+        if status in {"success", "abort"}:
+            return ReflexionDecision(
+                action="tool_call",
+                rationale=rationale,
+                tool_name="verdict",
+                tool_args={
+                    "status": status,
+                    "rationale": str(
+                        raw_verdict.get("rationale") or rationale or status
+                    ),
+                },
+            )
+    return ReflexionDecision(
+        action="tool_call",
+        rationale=rationale or "Legacy verdict_now without status; propose shard plan.",
+        tool_name="propose.shard_plan",
+        tool_args={},
+    )
+
+
 def _parse_decision(raw: str) -> ReflexionDecision:
     data = json.loads(raw)
-    action = str(data.get("action") or "tool_call")
-    if action not in {"tool_call", "verdict_now"}:
+    action = str(data.get("action") or "tool_call").strip().lower()
+    if action == "verdict_now":
+        return _coerce_legacy_finish(data if isinstance(data, dict) else {})
+    if action != "tool_call":
         action = "tool_call"
-    verdict = None
-    if isinstance(data.get("verdict"), dict):
-        verdict_data = data["verdict"]
-        status = str(verdict_data.get("status") or "abort")
-        if status not in {"success", "abort"}:
-            status = "abort"
-        verdict = AgentVerdict(
-            status=status,  # type: ignore[arg-type]
-            rationale=str(verdict_data.get("rationale") or data.get("rationale") or ""),
-        )
     return ReflexionDecision(
-        action=action,  # type: ignore[arg-type]
+        action="tool_call",
         rationale=str(data.get("rationale") or ""),
         tool_name=data.get("tool_name"),
         tool_args=dict(data.get("tool_args") or {}),
-        verdict=verdict,
+        verdict=None,
     )
 
 
@@ -98,6 +120,15 @@ def run(self) -> ExecutorResult:
         for round_index in range(self.max_rounds):
             pending_recovery_verdict: AgentVerdict | None = None
             decision, result = self._next_decision(round_index)
+            if decision.action != "tool_call":
+                # Defensive: only tool_call is a legal executor step.
+                decision = ReflexionDecision(
+                    action="tool_call",
+                    rationale=decision.rationale
+                    or "Non-tool executor action coerced to propose.shard_plan.",
+                    tool_name="propose.shard_plan",
+                    tool_args={},
+                )
             self.ctx.blackboard.global_signals.setdefault("reflexion_decisions", []).append(
                 decision.to_dict()
             )
@@ -111,14 +142,9 @@ def run(self) -> ExecutorResult:
                     tool_args=decision.tool_args,
                 )
 
-            tool_name: str | None = None
-            tool_args: dict[str, Any] = {}
-            if decision.action == "verdict_now":
-                verdict = decision.verdict or AgentVerdict(
-                    status="abort",
-                    rationale=decision.rationale or "Executor stopped without verdict.",
-                )
-                if verdict.status == "success" and not (
+            tool_name, tool_args = self._resolve_tool_call(decision)
+            if tool_name == "verdict" and str(tool_args.get("status") or "") == "success":
+                if not (
                     self.ctx.blackboard.validation_report
                     and self.ctx.blackboard.validation_report.get("valid") is True
                 ):
@@ -131,11 +157,6 @@ def run(self) -> ExecutorResult:
                         tool_args={},
                     )
                     tool_name, tool_args = self._resolve_tool_call(decision)
-                else:
-                    self.ctx.blackboard.verdict = verdict
-                    return ExecutorResult(verdict=verdict, rounds=round_index + 1)
-            else:
-                tool_name, tool_args = self._resolve_tool_call(decision)
 
             if not tool_name:
                 verdict = AgentVerdict(
@@ -205,6 +226,22 @@ def _is_deterministic_mode(self) -> bool:
     def _next_decision(self, round_index: int) -> tuple[ReflexionDecision, ToolResult]:
         if round_index == 0 and self._initial_decision is not None:
             decision = self._initial_decision
+            if decision.action != "tool_call":
+                decision = ReflexionDecision(
+                    action="tool_call",
+                    rationale=decision.rationale
+                    or "Initial non-tool decision coerced to propose.shard_plan.",
+                    tool_name="propose.shard_plan",
+                    tool_args={},
+                )
+            elif not decision.tool_name:
+                decision = ReflexionDecision(
+                    action="tool_call",
+                    rationale=decision.rationale
+                    or "Initial decision missing tool; propose shard plan.",
+                    tool_name="propose.shard_plan",
+                    tool_args={},
+                )
             return decision, ToolResult(status="ok", payload=decision.to_dict())
         model = self.ctx.settings.get("executor_model") or self.ctx.settings.get("model")
         if not model:
@@ -224,9 +261,13 @@ def _next_decision(self, round_index: int) -> tuple[ReflexionDecision, ToolResul
         est = estimate_tokens(prompt)
         if not self.ctx.budget.try_reserve("plan", est):
             decision = ReflexionDecision(
-                action="verdict_now",
+                action="tool_call",
                 rationale="Planner budget exhausted.",
-                verdict=AgentVerdict(status="abort", rationale="Planner budget exhausted."),
+                tool_name="verdict",
+                tool_args={
+                    "status": "abort",
+                    "rationale": "Planner budget exhausted.",
+                },
             )
             return decision, ToolResult(
                 status="ok",
@@ -303,4 +344,3 @@ def _deterministic_decision(self) -> ReflexionDecision:
             tool_name="validate.anatomy_map",
             tool_args={},
         )
-

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

@@ -10,7 +10,10 @@
 PageKind = Literal["normal", "landscape"]
 TocFailureKind = Literal["none", "confirm_failed", "rejected_all", "degraded"]
 
-ReflexionAction = Literal["tool_call", "verdict_now"]
+# Executor loop steps are always tool calls. Profile success/abort is owned
+# exclusively by the ``verdict`` tool (AgentVerdict.status), not by a separate
+# ReflexionAction shortcut.
+ReflexionAction = Literal["tool_call"]
 VerdictStatus = Literal["success", "abort"]
 
 
@@ -28,8 +31,6 @@ class PageFeature:
     height: float
     has_asset: bool
     is_blank_like: bool
-    # 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)

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

@@ -1,8 +1,13 @@
 """Persist anatomy map artifacts."""
 
 from app.services.document_agent.tools.persist_anatomy_map import (
+    DOC_PROFILE_FILENAME,
     build_anatomy_map,
     persist_anatomy_map,
 )
 
-__all__ = ["build_anatomy_map", "persist_anatomy_map"]
+__all__ = [
+    "DOC_PROFILE_FILENAME",
+    "build_anatomy_map",
+    "persist_anatomy_map",
+]

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

@@ -156,7 +156,11 @@ def _parse_profile_and_decision(raw: str) -> tuple[DocumentProfile, ReflexionDec
         header_y=header_y,
         footer_y=footer_y,
     )
-    next_action = str(data.get("next_action") or "ready_to_shard")
+    next_action = str(data.get("next_action") or "ready_to_shard").strip().lower()
+    # Legacy models may still emit verdict_now; that is not a planner finish
+    # signal — fall through to ready_to_shard so the executor owns success/abort.
+    if next_action == "verdict_now":
+        next_action = "ready_to_shard"
     tool_name: str | None = None
     tool_args: dict[str, Any] = {}
     if next_action == "inspect_more":
@@ -171,12 +175,6 @@ def _parse_profile_and_decision(raw: str) -> tuple[DocumentProfile, ReflexionDec
         if query:
             tool_name = "grep.text"
             tool_args = {"query": query, "max_results": 20}
-    elif next_action == "verdict_now":
-        return profile, ReflexionDecision(
-            action="verdict_now",
-            rationale=profile.rationale,
-            verdict=None,
-        )
     if tool_name:
         return profile, ReflexionDecision(
             action="tool_call",

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

@@ -17,10 +17,12 @@
     "footer_y is the highest footer line you observe (smallest y) when any "
     "footer is present, otherwise null. When both are set, require "
     "header_y < footer_y. "
-    "next_action must be one of inspect_more, grep_text, ready_to_shard, "
-    "verdict_now. Use inspect_more only when extra page screenshots are needed. "
+    "next_action must be one of inspect_more, grep_text, ready_to_shard. "
+    "Use inspect_more only when extra page screenshots are needed. "
     "Use grep_text only for native PDFs when a global text search would clarify "
-    "structure. Do not output a fixed step plan."
+    "structure. Use ready_to_shard when evidence is sufficient to propose shards. "
+    "Do not finish or abort the profile run from next_action; the executor owns "
+    "success/abort via the verdict tool. Do not output a fixed step plan."
 )
 
 __all__ = ["PLANNER_INSTRUCTIONS"]

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

@@ -651,8 +651,9 @@ def _entries_to_tree(entries: list[dict[str, Any]]) -> list[TitleNode]:
     stack: list[tuple[int, TitleNode]] = []
 
     for entry in entries:
-        raw_title = str(entry.get("heading") or "").strip()
-        title = clean_toc_title(raw_title) or normalize_heading_text(raw_title)
+        # Keep original TOC heading (incl. numbering). Prefix stripping belongs
+        # only in text/compact match helpers used for null-page parents.
+        title = normalize_heading_text(str(entry.get("heading") or ""))
         level = _safe_int(entry.get("level")) or 1
         if not title or len(title) < 2:
             continue