Resolve Issue #115: Fix agentic navigation for deep nested document sections

apps/api/app/api/v1/routes/retrieval.py

@@ -99,8 +99,8 @@ class RetrievalQueryResponse(BaseModel):
         default=None,
         description=(
             "Per-step navigation decisions from agentic retrieval. "
-            "Each entry has phase, document, action, reason, stop_type, "
-            "selected_paths, and hydrated_count. Use this to understand "
+            "Each entry has phase, document, action, reason, collected_paths, "
+            "and drill_into. Use this to understand "
             "why KNOWHERE stopped or made specific navigation choices."
         ),
     )

apps/api/tests/contract/test_retrieval_contract.py

@@ -346,15 +346,14 @@ async def test_should_return_empty_results_for_an_empty_query(
         )
 
     assert response.status_code == 200
-    assert response.json() == {
-        "namespace": "default",
-        "query": "",
-        "router_used": "empty_query_filtered",
-        "evidence_text": "",
-        "answer_text": "",
-        "results": [],
-        "referenced_chunks": [],
-    }
+    response_json = response.json()
+    assert response_json["namespace"] == "default"
+    assert response_json["query"] == ""
+    assert response_json["router_used"] == "empty_query_filtered"
+    assert response_json["evidence_text"] == ""
+    assert response_json["answer_text"] == ""
+    assert response_json["results"] == []
+    assert response_json["referenced_chunks"] == []
 
 
 @pytest.mark.asyncio

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

@@ -110,7 +110,7 @@ def to_dict(self) -> dict[str, Any]:
 class TocResult:
     toc_pages: list[int] = field(default_factory=list)
     candidates: list[TocCandidate] = field(default_factory=list)
-    method: Literal["toc_marker", "vlm_progressive", "visual_scan", "none"] = "none"
+    method: Literal["toc_marker", "vlm_progressive", "vlm_batch", "visual_scan", "none"] = "none"
     notes: str = ""
 
     def to_dict(self) -> dict[str, Any]:

apps/worker/app/services/document_parser/formats/pdf/shard_merger.py

@@ -61,6 +61,7 @@ def merge_shard_lines(shard_lines_list: list[list[str]]) -> list[str]:
 
             if (
                 last_heading_pos is not None
+                and last_heading_key is not None
                 and last_heading_key == next_first_heading
             ):
                 # Truncate from the last (duplicate) heading onward

apps/worker/app/services/document_parser/structure/heading_candidates.py

@@ -22,6 +22,8 @@ def get_max_lvl(code_str: str) -> int:
     integers, so the ``[…]`` bracket match is guaranteed.
     """
     match = re.search(r"\[([^]]+)]", code_str)
+    if match is None:
+        return -2
     nums = [int(item.strip()) for item in match.group(1).split(",")]
     max_value = int(max(nums))
     return max_value if max_value > 1 else -2

packages/shared-python/shared/services/ai/llm_mock.py

@@ -1,5 +1,7 @@
 """Helpers for deterministic mock responses from OpenAI-compatible LLM calls."""
 
+import json
+import re
 from typing import Any, Dict, List
 
 from loguru import logger
@@ -18,6 +20,13 @@ def build_mock_chat_completion_response(
         model_name,
         task_name,
     )
+    # For agentic tasks that need dynamic content extraction, build the response here.
+    if task_name == "agentic-planner":
+        return _build_planner_mock_response(prompt_text)
+    if task_name == "agentic-navigate":
+        return _build_navigate_mock_response(prompt_text)
+    if task_name == "agentic-discovery-select":
+        return _build_discovery_select_mock_response(prompt_text)
     return _build_mock_response(task_name)
 
 
@@ -68,6 +77,24 @@ def _detect_mock_task(prompt_text: str) -> str:
     """Infer the prompt task so the mock can return a compatible response shape."""
     normalized_prompt = prompt_text.lower()
 
+    # ── Agentic retrieval prompts (check first — they are structurally distinct) ──
+    if (
+        "you are a retrieval workflow planner" in normalized_prompt
+        and "concat_final_parts" in normalized_prompt
+    ):
+        return "agentic-planner"
+    if (
+        "you are a document navigation agent" in normalized_prompt
+        and "=== section tree ==" in normalized_prompt
+    ):
+        return "agentic-navigate"
+    if (
+        "=== discovery candidates ==" in normalized_prompt
+        and "\"selections\"" in normalized_prompt
+    ):
+        return "agentic-discovery-select"
+
+    # ── Document parsing / ingestion prompts ──
     if (
         "generate a concise title" in normalized_prompt
         and "return only the title" in normalized_prompt
@@ -134,9 +161,136 @@ def _detect_mock_task(prompt_text: str) -> str:
     return "default"
 
 
+def _extract_first_section_path(prompt_text: str) -> str | None:
+    """Pull the first path value from a COLLECTOR_PROMPT section tree block.
+
+    The section tree is rendered by section_prompt_projection.format_items_for_llm()
+    and each item line looks like::
+
+        ▸ [L1] path="Root" [text=1] ~100 tokens [Leaf]
+        └ [L2] path="Root / Sub" [text=2] ~200 tokens
+
+    We extract the value inside path="..." from the first matching line
+    within the === Section Tree === block.
+    """
+    tree_match = re.search(
+        r"=== Section Tree ===(.*?)=== End Section Tree ===",
+        prompt_text,
+        re.DOTALL | re.IGNORECASE,
+    )
+    if not tree_match:
+        return None
+    tree_block = tree_match.group(1)
+    # Match path="..." in the section tree — this is the canonical format
+    path_match = re.search(r'path="([^"]+)"', tree_block)
+    if path_match:
+        return path_match.group(1)
+    return None
+
+
+
+def _extract_user_query(prompt_text: str) -> str:
+    """Extract the user query line from a planner/navigation prompt.
+
+    The PLANNER_PROMPT and COLLECTOR_PROMPT both contain::
+        User query: {query}
+    """
+    match = re.search(r"User query:\s*(.+)", prompt_text)
+    if match:
+        return match.group(1).strip()
+    return "mock query"
+
+
+def _build_planner_mock_response(prompt_text: str) -> str:
+    """Return a valid single-step QueryPlan JSON using the real query from the prompt."""
+    query = _extract_user_query(prompt_text)
+    response = {
+        "reasoning_summary": "mock single-step plan",
+        "steps": [
+            {
+                "id": "s1",
+                "sub_query": query,
+                "step_kind": "retrieve",
+                "depends_on": [],
+                "output_role": "final_part",
+                "top_k": 10,
+            }
+        ],
+        "final_strategy": "concat_final_parts",
+    }
+    return json.dumps(response)
+
+
+def _build_navigate_mock_response(prompt_text: str) -> str:
+    """Return a mock COLLECTOR_PROMPT response that COLLECTs the first visible path."""
+    path = _extract_first_section_path(prompt_text)
+    if path:
+        response = {
+            "collect": [{"path": path, "confidence": 0.9, "outline": False}],
+            "action": "STOP",
+            "drill_into": None,
+            "tools": [],
+            "reason": "Mock: collected first available section",
+        }
+    else:
+        # No path found — STOP without collecting (safe fallback)
+        response = {
+            "collect": [],
+            "action": "STOP",
+            "drill_into": None,
+            "tools": [],
+            "reason": "Mock: no section path found in tree",
+        }
+    return json.dumps(response)
+
+
+def _extract_first_discovery_path(prompt_text: str) -> str | None:
+    """Pull the first path value from a DISCOVERY_SELECT_PROMPT candidates block.
+
+    Discovery hints are rendered by selection._project_discovery_hints() as::
+
+        ▸ path="Findings"
+            <summary text>
+
+    We extract the value inside path="..." from the candidates block.
+    """
+    candidates_match = re.search(
+        r"=== Discovery Candidates ===(.*?)=== End Discovery Candidates ===",
+        prompt_text,
+        re.DOTALL | re.IGNORECASE,
+    )
+    if not candidates_match:
+        return None
+    block = candidates_match.group(1)
+    # Match path="..." — same canonical format as section tree
+    path_match = re.search(r'path="([^"]+)"', block)
+    if path_match:
+        return path_match.group(1)
+    return None
+
+
+def _build_discovery_select_mock_response(prompt_text: str) -> str:
+    """Return a mock DISCOVERY_SELECT_PROMPT response selecting the first candidate."""
+    path = _extract_first_discovery_path(prompt_text)
+    if path:
+        response = {"selections": [{"path": path, "confidence": 0.85}]}
+    else:
+        response = {"selections": []}
+    return json.dumps(response)
+
+
 def _build_mock_response(task_name: str) -> str:
     """Return a canned response compatible with the inferred task contract."""
     response_by_task: Dict[str, str] = {
+        # Agentic retrieval — static fallbacks (dynamic responses built elsewhere)
+        "agentic-planner": (
+            '{"reasoning_summary": "mock single-step plan", '
+            '"steps": [{"id": "s1", "sub_query": "mock query", '
+            '"step_kind": "retrieve", "depends_on": [], '
+            '"output_role": "final_part", "top_k": 10}], '
+            '"final_strategy": "concat_final_parts"}'
+        ),
+        # Document parsing / ingestion tasks
         "fragment-title": "Mock Fragment Title",
         "detect-toc-range": '{"toc_start": null, "toc_end": null, "confidence": "low"}',
         "detect-table-headers": '{"answer": [0]}',

packages/shared-python/shared/services/retrieval/agentic/core/runtime.py

@@ -17,8 +17,8 @@
 
 def build_config_from_env() -> AgentRunConfig:
     return AgentRunConfig(
-        max_nav_depth=int(os.environ.get("RETRIEVAL_AGENTIC_MAX_NAV_DEPTH", "3")),
-        latency_budget_ms=int(os.environ.get("RETRIEVAL_AGENTIC_LATENCY_BUDGET_MS", "12000")),
+        max_nav_steps=int(os.environ.get("RETRIEVAL_AGENTIC_MAX_NAV_STEPS", "6")),
+        latency_budget_ms=int(os.environ.get("RETRIEVAL_AGENTIC_LATENCY_BUDGET_MS", "30000")),
         token_budget_total=int(os.environ.get("RETRIEVAL_AGENTIC_TOKEN_BUDGET_TOTAL", "40000")),
         planning_ratio=float(os.environ.get("RETRIEVAL_AGENTIC_PLANNING_RATIO", "0.5")),
         bootstrap_budget=int(os.environ.get("RETRIEVAL_AGENTIC_BOOTSTRAP_BUDGET", "2000")),
@@ -104,15 +104,15 @@ def for_document(
         llm_fn: LLMFn,
         *,
         doc_id: str,
-        depth: int,
+        step: int = 0,
     ) -> LLMFn:
         async def _call(prompt: Any) -> str:
             return await self.call(
                 llm_fn,
                 prompt,
                 pool="planning",
                 doc_id=doc_id,
-                priority="low" if depth >= 2 else "normal",
+                priority="low" if step >= 4 else "normal",
             )
 
         return _call

packages/shared-python/shared/services/retrieval/agentic/core/types.py

@@ -16,7 +16,7 @@
 @dataclass
 class AgentRunConfig:
     """Budget and limit configuration for a single agent run."""
-    max_nav_depth: int = 3  # max scope_navigate recursion depth
+    max_nav_steps: int = 6  # max navigation steps per document (no depth limit)
     latency_budget_ms: int = 12000
     token_budget_total: int = 40000
     planning_ratio: float = 0.5
@@ -79,6 +79,17 @@ def has_content(self) -> bool:
             return any(c.has_content() for c in self.children.values())
         return False
 
+    def has_leaf_content(self) -> bool:
+        """Check if this tree has any actual hydrated chunk content (not just outline).
+
+        Unlike ``has_content()`` which returns True for outline-only trees,
+        this method only returns True when real text/table/image chunks have
+        been hydrated into leaf_content.
+        """
+        if self.leaf_content:
+            return True
+        return any(c.has_leaf_content() for c in self.children.values())
+
     def flatten_chunk_rows(self) -> list[dict[str, Any]]:
         """Recursively collect all hydrated chunk rows (document order)."""
         rows: list[dict[str, Any]] = []
@@ -108,10 +119,16 @@ def add_leaf_chunks(self, path: str, chunks: list[dict[str, Any]]) -> None:
             existing.append(chunk)
 
     def reparent_leaf_content(self) -> None:
-        """Move descendant leaf paths into matching child nodes."""
+        """Move descendant leaf paths into matching child nodes.
+
+        Only moves true descendants (prefix match).  Content whose path
+        exactly equals a child key stays here — the renderer handles the
+        case where a path is both a child and a leaf (section with own
+        content *and* sub-sections).
+        """
         for child_path, child in list(self.children.items()):
             for leaf_path in list(self.leaf_content.keys()):
-                if leaf_path == child_path or leaf_path.startswith(child_path + ' / '):
+                if leaf_path.startswith(child_path + ' / '):
                     child.add_leaf_chunks(leaf_path, self.leaf_content.pop(leaf_path))
             child.reparent_leaf_content()
 
@@ -165,13 +182,32 @@ def merge(self, other: 'DocTreeNode') -> None:
 
 @dataclass
 class NavigateStepResult:
-    """Return type for navigate_step — typed replacement for raw tuple."""
-    action: str  # "NAVIGATE" or "STOP"
+    """Return type for navigate_step — Collector Agent model.
+
+    Each step returns:
+    - ``collect``: paths to add to the evidence collection (full hydration)
+    - ``drill``: paths to explore deeper in subsequent steps
+    - ``action``: navigation direction — DRILL/BACK/STOP
+    - ``tools``: optional asset tools (FIND_IMAGES/FIND_TABLES)
+    - ``node``: outline tree node for rendering context
+    - ``reason``: LLM reasoning for trace
+    """
+    action: str = "STOP"  # DRILL | BACK | STOP
+    collect: list[dict[str, Any]] = field(default_factory=list)
+    drill: list[dict[str, Any]] = field(default_factory=list)
     tools: list[str] = field(default_factory=list)
     node: DocTreeNode = field(default_factory=DocTreeNode)
-    pending: list[dict[str, Any]] = field(default_factory=list)
     reason: str = ""
-    stop_type: str = ""  # only for STOP: sufficient_outline | no_relevant_child | ...
+
+    @property
+    def drill_into(self) -> str | None:
+        """Single drill target path, or None."""
+        return self.drill[0]["path"] if self.drill else None
+
+    @property
+    def is_terminal(self) -> bool:
+        """True when navigation should stop (STOP or empty collect+drill)."""
+        return self.action == "STOP" or (not self.collect and not self.drill)
 
     @staticmethod
     def stop(scope_path: str | None = None) -> 'NavigateStepResult':

packages/shared-python/shared/services/retrieval/agentic/discovery/selection.py

@@ -125,7 +125,7 @@ def _project_discovery_hints(
         section_path = normalize_section_path(hint.get("section_path", ""))
         if not section_path:
             continue
-        if section_path in exclude_set:
+        if _is_covered_by_exclude(section_path, exclude_set):
             continue
         if section_path in hint_by_path:
             continue
@@ -139,6 +139,21 @@ def _project_discovery_hints(
     return hint_lines, hint_by_path
 
 
+def _is_covered_by_exclude(path: str, exclude_set: set[str]) -> bool:
+    """Check if *path* is covered by any entry in *exclude_set*.
+
+    A path is covered if it exactly matches an exclude entry, OR if any
+    exclude entry is a prefix of this path (i.e. the parent path was
+    already collected by navigation).
+    """
+    if path in exclude_set:
+        return True
+    for excluded in exclude_set:
+        if path.startswith(excluded + " / "):
+            return True
+    return False
+
+
 def _build_discovery_selection_prompt(
     *,
     document_id: str,
@@ -187,6 +202,10 @@ def _build_discovery_path_selections(
                 "hydrate_mode": "self_only",
             })
             continue
-        path_selections.append({"path": path, "confidence": confidence})
+        path_selections.append({
+            "path": path,
+            "confidence": confidence,
+            "hydrate_mode": "self_only",
+        })
 
     return path_selections, chunk_refs