From da9e4e9a30180861690c8a8238ec834fc103cb57 Mon Sep 17 00:00:00 2001
From: chengke <404835780@qq.com>
Date: Tue, 1 Sep 2026 19:00:02 +0800
Subject: [PATCH 1/2] refactor(retrieval): remove n_chunks references and
 optimize frequency query logic

This commit removes the `n_chunks` attribute from various classes and functions related to navigation and retrieval, streamlining the codebase. Additionally, it updates the frequency query logic to drive lookups from `document_map_unit_tokens`, enhancing performance and clarity in the retrieval process. Tests have been adjusted to reflect these changes.
---
 .../test_retrieval_map_unit_index_contract.py |  9 ++--
 .../services/retrieval/nav/nav_actions.py     |  4 +-
 .../services/retrieval/nav/nav_hierarchy.py   |  5 +--
 .../services/retrieval/nav/nav_knowhere.py    | 43 +++++++------------
 .../services/retrieval/nav/nav_map_scores.py  | 16 +++++--
 .../services/retrieval/nav/nav_projection.py  | 11 +----
 .../services/retrieval/nav/nav_types.py       |  1 -
 7 files changed, 37 insertions(+), 52 deletions(-)

diff --git a/apps/api/tests/contract/test_retrieval_map_unit_index_contract.py b/apps/api/tests/contract/test_retrieval_map_unit_index_contract.py
index c44f04fac..b4b047d86 100644
--- a/apps/api/tests/contract/test_retrieval_map_unit_index_contract.py
+++ b/apps/api/tests/contract/test_retrieval_map_unit_index_contract.py
@@ -59,7 +59,7 @@ def execute(self, statement: str, parameters: object = None) -> None:
                 self.rows = [(document_id, job_result_id, 1, 1, 0.0, 0.0)]
             elif "FROM document_map_units AS units" in statement:
                 self.rows = [("unit-frequency", document_id, "chunk-frequency", "section-frequency", 1, 1)]
-            elif "matching_tokens AS MATERIALIZED" in statement:
+            elif "FROM document_map_unit_tokens" in statement:
                 self.rows = [("unit-frequency", "path", "retrieval", 1)]
             else:
                 self.rows = []
@@ -101,14 +101,15 @@ def close(self) -> None:
     frequency_executions = [
         (statement, parameters)
         for statement, parameters in executions
-        if "matching_tokens AS MATERIALIZED" in statement
+        if "FROM document_map_unit_tokens" in statement
     ]
     assert len(frequency_executions) == 1
     statement, parameters = frequency_executions[0]
-    assert "FROM matching_tokens" in statement
+    assert "map_unit_id = ANY" in statement
     assert "token_hash = ANY" in statement
     assert isinstance(parameters, list)
-    assert parameters[0] == [
+    assert parameters[0] == ["unit-frequency"]
+    assert parameters[1] == [
         "6e51d6a3d90b6a3243d38e6da6b3f31f49867c1360beba83da8ca9630f9672c7"
     ]
 
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_actions.py b/packages/shared-python/shared/services/retrieval/nav/nav_actions.py
index 1c50e2452..dc7566228 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_actions.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_actions.py
@@ -103,7 +103,6 @@ def view_score(view: SectionView) -> float:
                     score=score,
                     metadata={
                         "map_id": view.map_id,
-                        "n_chunks": view.n_chunks,
                         "highlight": is_hit,
                         "multi": True,
                     },
@@ -186,9 +185,8 @@ def node_actions(sid: str) -> str:
         hit_tag = format_hit_tag(is_highlight=bool(view.is_highlight))
         harvested_tag = format_harvested_tag(getattr(view, "harvested_by", "") or "")
         map_id = view.map_id or "?"
-        meta = f"({view.n_chunks} chunks)"
         lines.append(
-            f"{indent}[{map_id}] {view.title or view.section_id} {meta}"
+            f"{indent}[{map_id}] {view.title or view.section_id}"
             f"{leaf_tag}{hit_tag}{harvested_tag} actions: {node_actions(view.section_id)}"
         )
         if inline_summary and view.summary:
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_hierarchy.py b/packages/shared-python/shared/services/retrieval/nav/nav_hierarchy.py
index 8c2a3f4fc..80b9cde01 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_hierarchy.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_hierarchy.py
@@ -45,7 +45,6 @@ class NodeMeta:
     title: str = ""
     summary: str = ""
     has_children: bool = False
-    n_chunks: int = 0
 
 
 @runtime_checkable
@@ -61,7 +60,7 @@ def children(self, section_id: str) -> Sequence[str]:
         ...
 
     def node_meta(self, section_id: str) -> NodeMeta:
-        """Title/summary/chunk-count/has_children for one node."""
+        """Title/summary/has_children for one node."""
         ...
 
     def relations(self, section_id: str) -> Tuple[Set[str], Set[str]]:
@@ -134,7 +133,6 @@ def get_structure(self, section_id: str) -> dict:
             "preview": meta.title,
             "summary": str(meta.summary or ""),
             "n_lines": 1,
-            "n_chunks": int(meta.n_chunks),
             "children": [
                 {"section_id": cid, "preview": self._provider.node_meta(cid).title}
                 for cid in child_ids
@@ -332,7 +330,6 @@ def node_meta(self, section_id: str) -> NodeMeta:
             title=node.title,
             summary=self._summaries.get(section_id, ""),
             has_children=bool(node.children),
-            n_chunks=1,
         )
 
     def parent_id(self, section_id: str) -> Optional[str]:
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py b/packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py
index 68d76d3e1..bd4e24f76 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py
@@ -435,26 +435,21 @@ def load_persisted_score_corpus(
             frequencies: Dict[Tuple[str, str], Dict[str, int]] = {}
             if unit_rows and query_tokens:
                 stage_started = time.perf_counter()
+                # Restrict the token scan to this episode's map units instead of
+                # matching token_hash across the whole table then filtering.
+                allowed_map_unit_ids = [str(row["map_unit_id"]) for row in unit_rows]
                 cur.execute(
-                    "WITH matching_tokens AS MATERIALIZED ("
                     "SELECT map_unit_id, channel, token, frequency "
                     "FROM document_map_unit_tokens "
-                    "WHERE token_hash = ANY(%s) AND channel = ANY(%s)"
-                    ") "
-                    "SELECT matching_tokens.map_unit_id, matching_tokens.channel, "
-                    "matching_tokens.token, matching_tokens.frequency "
-                    "FROM matching_tokens "
-                    "JOIN document_map_units AS units "
-                    "ON units.id = matching_tokens.map_unit_id "
-                    f"JOIN (VALUES {values_sql}) AS revisions(document_id, job_result_id) "
-                    "ON units.document_id = revisions.document_id "
-                    "AND units.job_result_id = revisions.job_result_id",
-                    [list(query_token_hashes), list(_MAP_SCORE_CHANNELS), *revision_params],
+                    "WHERE map_unit_id = ANY(%s) "
+                    "AND token_hash = ANY(%s) AND channel = ANY(%s)",
+                    [
+                        allowed_map_unit_ids,
+                        list(query_token_hashes),
+                        list(_MAP_SCORE_CHANNELS),
+                    ],
                 )
-                allowed_map_unit_ids = {str(row["map_unit_id"]) for row in unit_rows}
                 for map_unit_id, channel, token, frequency in cur.fetchall():
-                    if str(map_unit_id) not in allowed_map_unit_ids:
-                        continue
                     frequencies.setdefault((str(map_unit_id), str(channel)), {})[
                         str(token)
                     ] = int(frequency)
@@ -502,6 +497,11 @@ def load_persisted_score_corpus(
                         ),
                     )
                     for row in unit_rows
+                    # Only units with at least one query-token frequency can
+                    # score > 0; zero-frequency units are implicit 0 and are
+                    # never materialized for BM25.
+                    if frequencies.get((str(row["map_unit_id"]), "path"))
+                    or frequencies.get((str(row["map_unit_id"]), "content"))
                 ],
                 path_stats=path_stats,
                 content_stats=content_stats,
@@ -708,7 +708,6 @@ def node_meta(self, section_id: str) -> NodeMeta:
             title=row.section_title,
             summary=row.summary,
             has_children=bool(self._children.get(section_id)),
-            n_chunks=len(self.subtree_units(section_id)),
         )
 
     def relations(self, section_id: str) -> Tuple[Set[str], Set[str]]:
@@ -799,9 +798,6 @@ def summaries(self) -> Dict[str, str]:
     def all_section_ids(self) -> List[str]:
         return list(self._sections)
 
-    def chunk_count(self) -> int:
-        return len(self._chunk_ids)
-
 
 class LazyKnowhereProvider(KnowhereProvider):
     """Hierarchy provider that loads full chunk rows only on first access."""
@@ -1113,19 +1109,10 @@ def node_meta(self, section_id: str) -> NodeMeta:
         sid = str(section_id or "").strip()
         if sid in self._docs:
             provider = self._docs[sid]
-            count_fn = getattr(provider, "chunk_count", None)
-            n_chunks = (
-                int(count_fn())
-                if callable(count_fn)
-                else sum(
-                    len(provider.self_units(sec)) for sec in provider.all_section_ids()
-                )
-            )
             return NodeMeta(
                 title=self._titles.get(sid, sid),
                 summary="",
                 has_children=bool(provider.roots(sid)),
-                n_chunks=n_chunks,
             )
         owner = self._section_owner.get(sid)
         if not owner:
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py b/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py
index b48d70249..fcde20e8e 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py
@@ -395,15 +395,25 @@ def compute_corpus_map_and_unit_scores_many(
 
     del namespace  # Dense scoring is intentionally disabled for the corpus path.
 
+    # Tree shape is query-independent; reuse it across the episode's two
+    # scoring passes (user query + per-subgoal retrieval_query) on the same
+    # ToolSpace instead of re-walking every document each time.
+    tree_cache = getattr(ts, "_mapnav_tree_cache", None)
+    if not isinstance(tree_cache, dict):
+        tree_cache = {}
+        setattr(ts, "_mapnav_tree_cache", tree_cache)
     tree_by_doc: Dict[
         str,
         Tuple[Dict[str, List[str]], Set[str], Dict[str, str]],
     ] = {}
     tree_started = time.perf_counter()
     for doc_id in valid_doc_ids:
-        root_ids = list(ts.sections_for_doc(doc_id))
-        children_map, leaves, titles = _walk_tree(ts, doc_id, root_ids)
-        tree_by_doc[doc_id] = (children_map, leaves, titles)
+        cached = tree_cache.get(doc_id)
+        if cached is None:
+            root_ids = list(ts.sections_for_doc(doc_id))
+            cached = _walk_tree(ts, doc_id, root_ids)
+            tree_cache[doc_id] = cached
+        tree_by_doc[doc_id] = cached
     _logger.info(
         "retrieval mapnav phase=tree_build seconds=%.3f documents=%d sections=%d",
         time.perf_counter() - tree_started,
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_projection.py b/packages/shared-python/shared/services/retrieval/nav/nav_projection.py
index 9bf049651..b4ffd5794 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_projection.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_projection.py
@@ -76,7 +76,6 @@ def _section_view_from_structure(
         preview=preview,
         score=_lexical_score(query, f"{section_id} {preview}"),
         n_lines=int(st.get("n_lines") or 0),
-        n_chunks=int(st.get("n_chunks") or 0),
         has_children=bool(children),
         depth_from_scope=depth_from_scope,
         title=preview[:80] if preview else section_id,
@@ -127,7 +126,6 @@ class _MapNode:
     title: str
     score: float
     n_lines: int
-    n_chunks: int
     has_children: bool
     children: List["_MapNode"] = field(default_factory=list)
     n_descendants: int = 0
@@ -353,7 +351,6 @@ def make_node(section_id: str, depth: int, parent_id: Optional[str]) -> Optional
             title=title,
             score=score,
             n_lines=int(st.get("n_lines") or 0),
-            n_chunks=int(st.get("n_chunks") or 0),
             has_children=False,
             parent_id=parent_id,
         )
@@ -461,10 +458,7 @@ def render(node: _MapNode) -> None:
         leaf_tag = " [Leaf]" if not node.has_children else ""
         hit_tag = format_hit_tag(is_highlight=is_hit)
         harvested_tag = format_harvested_tag(node.harvested_by)
-        line = (
-            f"{indent}[{map_id}] {node.title} ({node.n_chunks} chunks)"
-            f"{leaf_tag}{hit_tag}{harvested_tag}"
-        )
+        line = f"{indent}[{map_id}] {node.title}{leaf_tag}{hit_tag}{harvested_tag}"
         lines.append(line)
         summary = ""
         if inline_summary:
@@ -484,7 +478,6 @@ def render(node: _MapNode) -> None:
                 preview="",
                 score=node.score,
                 n_lines=node.n_lines,
-                n_chunks=node.n_chunks,
                 has_children=node.has_children,
                 depth_from_scope=node.depth,
                 map_id=map_id,
@@ -703,7 +696,7 @@ def add_line(text: str) -> None:
         leaf_tag = " [Leaf]" if not view.has_children else ""
         title = view.preview[:80] if view.preview else view.section_id
         add_line(
-            f"{indent}[{view.section_id}] {title} ({view.n_chunks} chunks){leaf_tag}"
+            f"{indent}[{view.section_id}] {title}{leaf_tag}"
         )
         if view.preview:
             add_line(f"{indent}     Preview: \"{view.preview[:80]}\"")
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_types.py b/packages/shared-python/shared/services/retrieval/nav/nav_types.py
index d164d1aa6..aebee1428 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_types.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_types.py
@@ -175,7 +175,6 @@ class SectionView:
     preview: str
     score: float = 0.0
     n_lines: int = 0
-    n_chunks: int = 0
     has_children: bool = False
     depth_from_scope: int = 0
     map_id: str = ""

From a5a1858083cc61905f1a5e25fbbe544f1263486a Mon Sep 17 00:00:00 2001
From: chengke <404835780@qq.com>
Date: Tue, 1 Sep 2026 19:23:49 +0800
Subject: [PATCH 2/2] refactor(retrieval): plan query-only then light map from
 retrieval_query

Defer corpus map scoring until after the planner emits subgoals, cache those
scores for harvest, and keep refine map-aware after lighting.

Co-authored-by: Cursor <cursoragent@cursor.com>
---
 .../services/retrieval/nav/nav_agent.py       |  31 +--
 .../services/retrieval/nav/nav_map_scores.py  |   3 +-
 .../services/retrieval/nav/nav_orchestrate.py | 151 +++++++++--
 .../shared/services/retrieval/nav/nav_plan.py |  74 ++---
 .../shared/services/retrieval/trace/mapnav.py |   3 -
 .../shared/tests/test_nav_plan_node_filter.py |   2 +
 .../shared/tests/test_nav_plan_query_only.py  | 253 ++++++++++++++++++
 7 files changed, 420 insertions(+), 97 deletions(-)
 create mode 100644 packages/shared-python/shared/tests/test_nav_plan_query_only.py

diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_agent.py b/packages/shared-python/shared/services/retrieval/nav/nav_agent.py
index 3b41214b4..7df6eca34 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_agent.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_agent.py
@@ -23,11 +23,6 @@
     pack_nav_evidence,
     unit_score_for_evidence_chunk,
 )
-from .nav_map_scores import (
-    compute_corpus_map_and_unit_scores,
-    compute_map_and_unit_scores,
-    select_map_highlights,
-)
 from .nav_types import (
     LegalAction,
     NavConfig,
@@ -395,31 +390,16 @@ def _run_nav_episode_body(
 
     state = NavState(doc_id=episode_doc, query=query, task_type=task_type)
     steps: List[AgentStep] = []
-    map_started = time.perf_counter()
     if namespace_mode:
         section_ids = list(ts.sections_for_doc(""))
-        state.map_scores, state.unit_scores = compute_corpus_map_and_unit_scores(
-            ts, doc_ids=corpus_ids, query=query
-        )
     else:
         section_ids = ts.sections_for_doc(episode_doc)
-        state.map_scores, state.unit_scores = compute_map_and_unit_scores(
-            ts, doc_id=episode_doc, query=query, root_ids=section_ids
-        )
-    _logger.info(
-        "retrieval mapnav phase=map_scoring seconds=%.3f documents=%d sections=%d",
-        time.perf_counter() - map_started,
-        len(corpus_ids),
-        len(section_ids),
-    )
-    state.highlight_ids = select_map_highlights(
-        state.unit_scores, k=int(cfg.collect_top_k)
-    )
 
     from .nav_plan import plan_query
 
+    # Planner is query-only: no pre-lit map. Score/light after the plan exists.
     plan_t0 = time.perf_counter()
-    retrieval_plan = plan_query(ts, state, cfg)
+    retrieval_plan = plan_query(state, cfg)
     state.retrieval_plan = retrieval_plan
     steps.append(
         AgentStep(
@@ -430,9 +410,6 @@ def _run_nav_episode_body(
                 "n_subgoals": len(retrieval_plan.subgoals),
                 "reason": retrieval_plan.reason,
                 "plan": retrieval_plan.to_dict(),
-                "planning_map_char_limit": int(
-                    getattr(cfg, "planning_map_char_limit", 0) or cfg.map_char_limit
-                ),
                 "seconds": time.perf_counter() - plan_t0,
             }, t0=plan_t0),
         )
@@ -448,7 +425,9 @@ def _run_nav_episode_body(
         from .nav_plan import fallback_plan
 
         state.retrieval_plan = fallback_plan(query, reason="missing_plan")
-    from .nav_orchestrate import execute_plan
+    from .nav_orchestrate import execute_plan, seed_episode_map_scores_from_plan
+
+    seed_episode_map_scores_from_plan(ts, state, cfg, state.retrieval_plan)
 
     orch_t0 = time.perf_counter()
     orch_detail = execute_plan(
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py b/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py
index fcde20e8e..28c046001 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py
@@ -518,8 +518,7 @@ def relight_map_for_query(
 ) -> Tuple[Dict[str, float], Dict[str, float], List[str]]:
     """Re-score the whole shared map against ``query``.
 
-    Same namespace / single-doc split as the episode-level pass in ``nav_agent``:
-    an empty ``doc_id`` means the corpus root, where document ids are map nodes
+    An empty ``doc_id`` means the corpus root, where document ids are map nodes
     and ``ts.document_ids()`` is already restricted to the episode's corpus.
     """
     doc = str(doc_id or "").strip()
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_orchestrate.py b/packages/shared-python/shared/services/retrieval/nav/nav_orchestrate.py
index bbb3bce1b..c48038179 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_orchestrate.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_orchestrate.py
@@ -136,6 +136,94 @@ def _resolve_subgoal_query(state: NavState, subgoal: Subgoal) -> str:
     return refined or query
 
 
+def seed_episode_map_scores_from_plan(
+    ts: Any,
+    state: NavState,
+    config: NavConfig,
+    plan: RetrievalPlan,
+) -> int:
+    """Light the map from executable plan retrieval_query strings; set baseline.
+
+    Scores bound (or refined) retrieval_query values only — queries still holding
+    ``{{slot}}`` placeholders are skipped and scored later when harvest resolves
+    them. Empty plan falls back to ``state.query``. Harvest may temporarily
+    relight per subgoal; ``_relit_map`` restores this baseline for evidence pack.
+    Returns the number of unique queries scored.
+    """
+    queries: List[str] = []
+    seen: Set[str] = set()
+    for subgoal in plan.subgoals:
+        refined = str(
+            (state.subgoal_refined_queries or {}).get(subgoal.id) or ""
+        ).strip()
+        if refined:
+            q = refined
+        else:
+            bound = bind_slots(subgoal.retrieval_query, state.slot_bindings)
+            # Skip not-yet-bound slot queries — scoring a stripped skeleton
+            # cannot cache-hit the later bound harvest string.
+            if unbound_slots(bound):
+                continue
+            q = str(bound or "").strip()
+        if q and q not in seen:
+            seen.add(q)
+            queries.append(q)
+    if not queries:
+        q = str(state.query or "").strip()
+        if q:
+            queries = [q]
+    if not queries:
+        state.map_scores = {}
+        state.unit_scores = {}
+        state.highlight_ids = []
+        state.relit_map_cache = {}
+        return 0
+
+    # Replace prior episode lighting entirely (including after replan).
+    state.relit_map_cache = {}
+
+    map_started = time.perf_counter()
+    from .nav_map_scores import relight_maps_for_queries
+
+    prepared = relight_maps_for_queries(
+        ts,
+        doc_id=state.doc_id,
+        queries=queries,
+        top_k=int(config.collect_top_k),
+    )
+    for query, triple in prepared.items():
+        scores, units, highlights = triple
+        if scores:
+            state.relit_map_cache[query] = (
+                dict(scores),
+                dict(units),
+                list(highlights),
+            )
+
+    baseline = None
+    for query in queries:
+        triple = prepared.get(query)
+        if triple and triple[0]:
+            baseline = triple
+            break
+    if baseline is not None:
+        scores, units, highlights = baseline
+        state.map_scores = dict(scores)
+        state.unit_scores = dict(units)
+        state.highlight_ids = list(highlights)
+    else:
+        state.map_scores = {}
+        state.unit_scores = {}
+        state.highlight_ids = []
+
+    _logger.info(
+        "retrieval mapnav phase=map_scoring seconds=%.3f queries=%d",
+        time.perf_counter() - map_started,
+        len(queries),
+    )
+    return len(queries)
+
+
 def _wave_subgoal_result(
     plan: RetrievalPlan,
     state: NavState,
@@ -174,11 +262,10 @@ def _relit_map(
 ) -> Iterator[None]:
     """Score the shared map against the harvest ``query`` for one call.
 
-    Episode-level ``state.map_scores`` is computed from the original user query.
-    Checklist harvests run under a per-subgoal ``retrieval_query``, so the map
-    must be re-scored against that string — otherwise the ranking disagrees with
-    the query the policy is told to pursue. Scoring failures degrade to the
-    episode lighting.
+    Episode-level ``state.map_scores`` is the post-plan baseline (lit from
+    plan ``retrieval_query`` strings). Checklist harvests may run under a
+    different per-subgoal string, so the map is re-scored for that call and
+    restored to the episode baseline afterward for evidence packing.
     """
     relit = prepared
     q = (query or "").strip()
@@ -195,15 +282,19 @@ def _relit_map(
                 top_k=int(config.collect_top_k),
             )
             if scores:
-                relit = (scores, units, highlights)
+                relit = (dict(scores), dict(units), list(highlights))
                 state.relit_map_cache[q] = relit
         except Exception:
             relit = None
     if relit is None:
         yield
         return
+    scores, units, highlights = relit
     saved = (state.map_scores, state.unit_scores, state.highlight_ids)
-    state.map_scores, state.unit_scores, state.highlight_ids = relit
+    # Copy so harvest readers cannot mutate the cached baseline triple.
+    state.map_scores = dict(scores)
+    state.unit_scores = dict(units)
+    state.highlight_ids = list(highlights)
     try:
         yield
     finally:
@@ -548,23 +639,40 @@ def execute_plan(
         by_id = {s.id: s for s in plan.subgoals}
         outputs: List[Dict[str, Any]] = []
         query_by_subgoal = {
-            sid: _resolve_subgoal_query(state, by_id[sid]) for sid in ready
+            sid: str(_resolve_subgoal_query(state, by_id[sid]) or "").strip()
+            for sid in ready
         }
         prepared_relights: Dict[
             str,
             Tuple[Dict[str, float], Dict[str, float], List[str]],
         ] = {}
-        try:
-            from .nav_map_scores import relight_maps_for_queries
+        missing_queries: List[str] = []
+        for query in query_by_subgoal.values():
+            if not query:
+                continue
+            cached = state.relit_map_cache.get(query)
+            if cached is not None:
+                prepared_relights[query] = cached
+            elif query not in missing_queries:
+                missing_queries.append(query)
+        if missing_queries:
+            try:
+                from .nav_map_scores import relight_maps_for_queries
 
-            prepared_relights = relight_maps_for_queries(
-                ts,
-                doc_id=state.doc_id,
-                queries=list(query_by_subgoal.values()),
-                top_k=int(config.collect_top_k),
-            )
-        except Exception:
-            prepared_relights = {}
+                scored = relight_maps_for_queries(
+                    ts,
+                    doc_id=state.doc_id,
+                    queries=missing_queries,
+                    top_k=int(config.collect_top_k),
+                )
+                for query, triple in scored.items():
+                    scores, units, highlights = triple
+                    prepared = (dict(scores), dict(units), list(highlights))
+                    prepared_relights[query] = prepared
+                    if scores:
+                        state.relit_map_cache[query] = prepared
+            except Exception:
+                pass
 
         def _run_one(sid: str, working_state: NavState, out_steps: Optional[List[Any]]) -> Dict[str, Any]:
             query = query_by_subgoal[sid]
@@ -655,7 +763,7 @@ def _run_one(sid: str, working_state: NavState, out_steps: Optional[List[Any]])
             if cap > 0 and int(state.replan_count) < cap:
                 state.replan_count += 1
                 t0 = time.perf_counter()
-                new_plan = plan_query(ts, state, config)
+                new_plan = plan_query(state, config)
                 state.retrieval_plan = new_plan
                 plan = new_plan
                 # A regenerated plan gets fresh subgoal ids (s1, s2, ... again),
@@ -663,7 +771,9 @@ def _run_one(sid: str, working_state: NavState, out_steps: Optional[List[Any]])
                 # qualified "sX.slot" bindings) cannot be safely carried over —
                 # those ids now mean something else. What IS safe and worth
                 # keeping is unqualified slot bindings (plain fact values) and
-                # every chunk already in state.collected.
+                # every chunk already in state.collected. Clear before re-seeding
+                # so stale refined queries / qualified bindings cannot leak into
+                # the new plan's retrieval_query scoring.
                 state.satisfied_subgoal_ids = set()
                 state.attempted_subgoal_ids = set()
                 state.dropped_subgoal_ids = set()
@@ -675,6 +785,7 @@ def _run_one(sid: str, working_state: NavState, out_steps: Optional[List[Any]])
                 state.slot_bindings = {
                     k: v for k, v in state.slot_bindings.items() if "." not in k
                 }
+                seed_episode_map_scores_from_plan(ts, state, config, new_plan)
                 if steps_out is not None:
                     steps_out.append(
                         AgentStep(
diff --git a/packages/shared-python/shared/services/retrieval/nav/nav_plan.py b/packages/shared-python/shared/services/retrieval/nav/nav_plan.py
index 8e643a6e5..cf7aef876 100644
--- a/packages/shared-python/shared/services/retrieval/nav/nav_plan.py
+++ b/packages/shared-python/shared/services/retrieval/nav/nav_plan.py
@@ -1,7 +1,8 @@
 """Structure-conditioned query planning (M2).
 
-Looks at a planning map observation and emits a coverage checklist plus an
-auditable RetrievalPlan over one shared search space.
+``plan_query`` is query-only: it emits a coverage checklist plus an auditable
+RetrievalPlan from the user question (no pre-lit map). ``refine_subgoal_query``
+may still read a folded planning map after harvest has lit scores.
 """
 
 from __future__ import annotations
@@ -284,9 +285,9 @@ def language_reference_text(
 ) -> str:
     """Script reference for retrieval_query checks.
 
-    Uses the user query plus visible map *titles* only. The actionable map
-    observation is intentionally excluded — its English chrome (collect=/dispatch=
-    /[Hit]/ would falsely dominate script detection.
+    Planner is query-only (projection=None). Refine still sees the folded map;
+    its English chrome (collect=/dispatch=/[Hit]) is excluded — only visible
+    map *titles* are used so they do not dominate script detection.
     """
     parts: List[str] = []
     q = (query or "").strip()
@@ -420,7 +421,6 @@ def parse_retrieval_plan(
     obj: dict,
     *,
     query: str,
-    projection: Optional[Projection] = None,
 ) -> RetrievalPlan:
     """Parse LLM JSON into a RetrievalPlan; invalid refs are dropped."""
     rows = obj.get("subgoals") or obj.get("goals") or obj.get("steps") or []
@@ -620,38 +620,34 @@ def _planner_system_prompt(*, max_subgoals: int) -> str:
     if max_subgoals > 0:
         cap = f" Prefer at most {max_subgoals} subgoals."
     return (
-        "You are a retrieval planner for a hierarchical document map.\n"
-        "You see a folded title map of the corpus/document. Nodes may carry "
-        "collect=C* and dispatch=D* action ids; [Hit] marks hybrid retrieval beacons.\n"
-        "Your job is to emit a coverage checklist plus a retrieval plan over ONE "
-        "shared search space — do not partition the corpus with per-subgoal "
-        "scopes or map anchors.\n\n"
+        "You are a retrieval planner. You receive only the user query (no "
+        "document map). Emit a coverage checklist plus a retrieval plan over "
+        "ONE shared search space — do not invent per-subgoal corpus partitions.\n\n"
         "Rules:\n"
         "1. coverage_checklist lists the facts that episode evidence must cover "
         "(short, concrete facts in the query's language)."
         f"{cap}\n"
-        "2. Default to a SINGLE subgoal over the whole map. Only add more "
-        "subgoals for a hard data dependency ({{s1.slot}} in a later query) or "
-        "for clearly independent cross-entity comparisons. Do not split merely "
-        "to list checklist items.\n"
+        "2. Default to a SINGLE subgoal. Only add more subgoals for a hard data "
+        "dependency ({{s1.slot}} in a later query) or for clearly independent "
+        "cross-entity comparisons. Do not split merely to list checklist items.\n"
         "3. Each subgoal produces at most ONE slot name in produces "
         "(enumeration = one list-valued slot).\n"
         "4. retrieval_query is a SHORT KEYWORD QUERY for THIS subgoal only "
         "(space-separated entity/role/topic tokens, e.g. \"王仁坤 总工程师 设计成果\"). "
         "Split or adapt the user question into compact lexical terms — do NOT "
-        "emit a full natural-language question or long prose. It is scored by "
-        "lexical/hybrid retrieval and lights the map, so keep entity names and "
-        "role terms; drop filler words. Same language/script as the map titles "
-        "and user query — do not translate section terms into another script.\n"
+        "emit a full natural-language question or long prose. Downstream lexical "
+        "retrieval scores and lights the map from these tokens, so keep entity "
+        "names and role terms; drop filler words. Same language/script as the "
+        "user query — do not translate terms into another script.\n"
         "5. If a later retrieval_query needs a value from an earlier subgoal, "
         "write it as {{s1.slot}} (not prose). That implies depends_on.\n"
         "6. depends_on = hard data dependency. prefer_after = soft ordering only.\n"
         "7. All subgoals share one search space — do not invent per-subgoal scopes.\n"
         "8. relations only for parent-child or sibling (omit unrelated pairs).\n"
-        "9. map_coverage: sufficient | partial | insufficient — whether the planning "
-        "map shows enough structure to ground this plan.\n"
-        "10. reason must be English, under 40 words. Document titles stay original "
-        "language.\n"
+        "9. map_coverage: sufficient | partial | insufficient — whether the user "
+        "query alone is enough to write executable retrieval_query terms "
+        "(use sufficient unless the query is empty or unusable).\n"
+        "10. reason must be English, under 40 words.\n"
         "11. Set use_node_filter=true when the subgoal enumerates or compares "
         "named facets you can write as path predicates (filenames, section "
         "titles). Keep it false for vague semantic needs; "
@@ -694,7 +690,6 @@ def _planner_system_prompt(*, max_subgoals: int) -> str:
 
 
 def _language_repair_user(
-    observation: str,
     state: NavState,
     bad_plan: RetrievalPlan,
     *,
@@ -708,26 +703,21 @@ def _language_repair_user(
     return (
         f"User query: {state.query}\n"
         f"Task type: {state.task_type}\n\n"
-        f"=== Planning Map ===\n{observation}\n=== End Planning Map ===\n\n"
         "Your previous plan had retrieval_query values that do not match the "
-        "language/script of the map titles and user query. Those queries will "
-        "fail lexical retrieval. Rewrite the FULL plan JSON. Keep structure, but "
-        "rewrite every mismatched retrieval_query as a short keyword query "
-        "in the map's own language and terms (space-separated tokens, not a "
+        "language/script of the user query. Those queries will fail lexical "
+        "retrieval. Rewrite the FULL plan JSON. Keep structure, but rewrite "
+        "every mismatched retrieval_query as a short keyword query in the "
+        "query's own language and terms (space-separated tokens, not a "
         "full-sentence question).\n"
         f"Mismatched retrieval_query lines:\n{bad_block}\n"
     )
 
 
 def plan_query(
-    ts: Any,
     state: NavState,
     config: NavConfig,
-    *,
-    observation: Optional[str] = None,
-    projection: Optional[Projection] = None,
 ) -> RetrievalPlan:
-    """LLM plan over the planning map. Falls back to a single subgoal on failure."""
+    """LLM plan from the user query only (no pre-lit map). Falls back to one subgoal."""
     from .nav_llm import (  # type: ignore
         nav_chat,
         planner_output_max_tokens,
@@ -739,9 +729,6 @@ def plan_query(
     if nav_token_budget_exhausted():
         return fallback_plan(state.query, reason="token_limit")
 
-    if projection is None or observation is None:
-        projection, observation = build_planning_observation(ts, state, config)
-
     model = resolve_nav_model(
         model=config.planner_model,
         model_env="NAV_PLANNER_MODEL",
@@ -754,11 +741,10 @@ def plan_query(
         timeout_s = 300.0 if thinking_mode == "enabled" else 90.0
     max_subgoals = int(getattr(config, "planner_max_subgoals", 0) or 0)
     system = _planner_system_prompt(max_subgoals=max_subgoals)
-    reference = language_reference_text(query=state.query, projection=projection)
+    reference = language_reference_text(query=state.query)
     user = (
         f"User query: {state.query}\n"
         f"Task type: {state.task_type}\n\n"
-        f"=== Planning Map ===\n{observation}\n=== End Planning Map ===\n\n"
         "Return the retrieval plan JSON."
     )
     max_tokens = planner_output_max_tokens(
@@ -804,9 +790,7 @@ def plan_query(
                 last_err = "empty_content"
                 continue
             obj = extract_plan_json(last_raw) or {}
-            plan = parse_retrieval_plan(
-                obj, query=state.query, projection=projection
-            )
+            plan = parse_retrieval_plan(obj, query=state.query)
             plan.raw = last_raw[:2000]
             ok, why = validate_retrieval_plan(plan)
             if not ok:
@@ -815,9 +799,7 @@ def plan_query(
             if plan_has_language_mismatch(plan, reference) and not language_repair_used:
                 language_repair_used = True
                 last_err = "language_mismatch"
-                user = _language_repair_user(
-                    observation, state, plan, reference=reference
-                )
+                user = _language_repair_user(state, plan, reference=reference)
                 continue
             if plan_has_language_mismatch(plan, reference):
                 last_err = "language_mismatch"
diff --git a/packages/shared-python/shared/services/retrieval/trace/mapnav.py b/packages/shared-python/shared/services/retrieval/trace/mapnav.py
index 528223475..6a2e83244 100644
--- a/packages/shared-python/shared/services/retrieval/trace/mapnav.py
+++ b/packages/shared-python/shared/services/retrieval/trace/mapnav.py
@@ -94,9 +94,6 @@ def _map_one(
             parent_step_index=parent_step_index,
             scope=scope,
             observation={
-                "projection_chars": detail.get("projection_chars"),
-                "hit_section_ids": detail.get("hit_section_ids") or [],
-                "planning_map_char_limit": detail.get("planning_map_char_limit"),
                 "llm_raw": _clip_raw(
                     plan_payload.get("raw") or detail.get("llm_raw") or detail.get("raw")
                 ),
diff --git a/packages/shared-python/shared/tests/test_nav_plan_node_filter.py b/packages/shared-python/shared/tests/test_nav_plan_node_filter.py
index 4c8172aba..9099034e5 100644
--- a/packages/shared-python/shared/tests/test_nav_plan_node_filter.py
+++ b/packages/shared-python/shared/tests/test_nav_plan_node_filter.py
@@ -62,3 +62,5 @@ def test_planner_prompt_mentions_where_vs_fuzzy() -> None:
     text = _planner_system_prompt(max_subgoals=0)
     assert "use_node_filter" in text
     assert "fallback" in text
+    assert "only the user query" in text.lower()
+    assert "folded" not in text.lower()
diff --git a/packages/shared-python/shared/tests/test_nav_plan_query_only.py b/packages/shared-python/shared/tests/test_nav_plan_query_only.py
new file mode 100644
index 000000000..d94ce9f24
--- /dev/null
+++ b/packages/shared-python/shared/tests/test_nav_plan_query_only.py
@@ -0,0 +1,253 @@
+"""Planner is query-only; map lighting happens after the plan exists."""
+
+from __future__ import annotations
+
+from typing import Any
+
+from shared.services.retrieval.nav.nav_orchestrate import seed_episode_map_scores_from_plan
+from shared.services.retrieval.nav.nav_plan import (
+    RetrievalPlan,
+    Subgoal,
+    _planner_system_prompt,
+    fallback_plan,
+)
+from shared.services.retrieval.nav.nav_types import NavConfig, NavState
+
+
+def test_planner_system_prompt_is_query_only() -> None:
+    text = _planner_system_prompt(max_subgoals=3)
+    assert "only the user query" in text.lower()
+    assert "no document map" in text.lower()
+    assert "folded" not in text.lower()
+    assert "Hit" not in text
+    assert "collect=C" not in text
+    assert "whether the user query alone is enough" in text.lower()
+
+
+def test_seed_episode_scores_plan_retrieval_queries(monkeypatch: Any) -> None:
+    seen: dict[str, Any] = {}
+
+    def fake_relight(ts: Any, *, doc_id: str, queries: list[str], top_k: int):
+        seen["queries"] = list(queries)
+        seen["top_k"] = top_k
+        out = {}
+        for i, q in enumerate(queries):
+            out[q] = ({f"n{i}": 1.0}, {f"u{i}": 2.0}, [f"u{i}"])
+        return out
+
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_map_scores.relight_maps_for_queries",
+        fake_relight,
+    )
+    state = NavState(doc_id="", query="user original", task_type="unknown")
+    state.relit_map_cache = {"stale": ({}, {}, [])}
+    plan = RetrievalPlan(
+        subgoals=[
+            Subgoal(id="s1", need="a", retrieval_query="alpha terms"),
+            Subgoal(id="s2", need="b", retrieval_query="beta terms"),
+            Subgoal(id="s3", need="c", retrieval_query="alpha terms"),
+        ]
+    )
+    n = seed_episode_map_scores_from_plan(
+        object(), state, NavConfig(collect_top_k=7), plan
+    )
+    assert n == 2
+    assert seen["queries"] == ["alpha terms", "beta terms"]
+    assert seen["top_k"] == 7
+    assert "stale" not in state.relit_map_cache
+    assert state.map_scores == {"n0": 1.0}
+    assert state.unit_scores == {"u0": 2.0}
+    assert state.highlight_ids == ["u0"]
+    assert "alpha terms" in state.relit_map_cache
+    assert "beta terms" in state.relit_map_cache
+
+
+def test_seed_episode_fallback_uses_user_query(monkeypatch: Any) -> None:
+    seen: list[str] = []
+
+    def fake_relight(ts: Any, *, doc_id: str, queries: list[str], top_k: int):
+        seen.extend(queries)
+        return {queries[0]: ({"n": 1.0}, {"u": 1.0}, ["u"])}
+
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_map_scores.relight_maps_for_queries",
+        fake_relight,
+    )
+    state = NavState(doc_id="doc1", query="心血管", task_type="unknown")
+    plan = RetrievalPlan(
+        subgoals=[Subgoal(id="s1", need="should not score", retrieval_query="")]
+    )
+    n = seed_episode_map_scores_from_plan(
+        object(), state, NavConfig(collect_top_k=5), plan
+    )
+    assert n == 1
+    assert seen == ["心血管"]
+    assert state.map_scores == {"n": 1.0}
+
+
+def test_seed_skips_unbound_slot_queries(monkeypatch: Any) -> None:
+    seen: list[str] = []
+
+    def fake_relight(ts: Any, *, doc_id: str, queries: list[str], top_k: int):
+        seen.extend(queries)
+        return {q: ({f"n:{q}": 1.0}, {f"u:{q}": 1.0}, [f"u:{q}"]) for q in queries}
+
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_map_scores.relight_maps_for_queries",
+        fake_relight,
+    )
+    state = NavState(doc_id="", query="user original", task_type="unknown")
+    plan = RetrievalPlan(
+        subgoals=[
+            Subgoal(id="s1", need="a", retrieval_query="alpha"),
+            Subgoal(
+                id="s2",
+                need="b",
+                retrieval_query="beta {{s1.entity}}",
+                depends_on=["s1"],
+            ),
+        ]
+    )
+    n = seed_episode_map_scores_from_plan(object(), state, NavConfig(), plan)
+    assert n == 1
+    assert seen == ["alpha"]
+    assert "alpha" in state.relit_map_cache
+    assert not any("beta" in q for q in state.relit_map_cache)
+
+
+def test_wave_uses_seeded_cache_and_scores_missing_once(monkeypatch: Any) -> None:
+    score_calls: list[list[str]] = []
+
+    def fake_relight(ts: Any, *, doc_id: str, queries: list[str], top_k: int):
+        score_calls.append(list(queries))
+        return {
+            q: ({f"n:{q}": 1.0}, {f"u:{q}": 1.0}, [f"u:{q}"])
+            for q in queries
+        }
+
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_map_scores.relight_maps_for_queries",
+        fake_relight,
+    )
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_orchestrate._apply_plan_control",
+        lambda *args, **kwargs: {"done": True},
+    )
+
+    harvest_queries: list[str] = []
+
+    def fake_harvest(ts: Any, state: NavState, config: NavConfig, **kwargs: Any):
+        harvest_queries.append(str(kwargs.get("query") or ""))
+        return type(
+            "HR",
+            (),
+            {
+                "n_policy_calls": 0,
+                "visited_section_ids": [],
+                "max_depth_hit": False,
+                "reason": "test",
+            },
+        )()
+
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_orchestrate._harvest_after_node_filter",
+        fake_harvest,
+    )
+
+    state = NavState(doc_id="", query="user original", task_type="unknown")
+    plan = RetrievalPlan(
+        subgoals=[
+            Subgoal(id="s1", need="a", retrieval_query="alpha"),
+            Subgoal(id="s2", need="b", retrieval_query="beta"),
+        ]
+    )
+    state.retrieval_plan = plan
+    seed_episode_map_scores_from_plan(object(), state, NavConfig(), plan)
+    assert score_calls == [["alpha", "beta"]]
+
+    from shared.services.retrieval.nav.nav_orchestrate import execute_plan
+
+    execute_plan(object(), state, NavConfig(), episode_query="user original")
+    # Wave relight must hit the seeded cache; no second full-corpus scoring.
+    assert score_calls == [["alpha", "beta"]]
+    assert harvest_queries == ["alpha", "beta"]
+
+
+def test_run_nav_episode_scores_after_plan(monkeypatch: Any) -> None:
+    order: list[str] = []
+
+    class _TS:
+        def sections_for_doc(self, _doc_id: str) -> list[str]:
+            return ["sec_root"]
+
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav._compat.load_llm_env",
+        lambda: None,
+    )
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav._compat.require_llm_env",
+        lambda *_args, **_kwargs: None,
+    )
+
+    def fake_plan_query(state: NavState, config: NavConfig, **_kwargs: Any):
+        order.append("plan")
+        assert state.map_scores == {}
+        assert state.unit_scores == {}
+        return fallback_plan(state.query, reason="test_plan")
+
+    def fake_seed(ts: Any, state: NavState, config: NavConfig, plan: RetrievalPlan):
+        order.append("score")
+        assert plan.subgoals
+        state.map_scores = {"sec_root": 1.0}
+        state.unit_scores = {"u1": 1.0}
+        state.highlight_ids = ["u1"]
+        return 1
+
+    def fake_execute(ts: Any, state: NavState, config: NavConfig, **_kwargs: Any):
+        order.append("orch")
+        assert state.map_scores == {"sec_root": 1.0}
+        return {"waves": []}
+
+    class _Fill:
+        scored_chunks = []
+        kept_chunks = []
+        evidence_text = ""
+        evidence_chars_actual = 0
+        n_chunks_kept = 0
+        truncated_last = False
+
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_plan.plan_query",
+        fake_plan_query,
+    )
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_orchestrate.seed_episode_map_scores_from_plan",
+        fake_seed,
+    )
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_orchestrate.execute_plan",
+        fake_execute,
+    )
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_agent.pack_nav_evidence",
+        lambda *_args, **_kwargs: _Fill(),
+    )
+    monkeypatch.setattr(
+        "shared.services.retrieval.nav.nav_agent.uses_document_nodes",
+        lambda _ts: False,
+    )
+
+    from shared.services.retrieval.nav.nav_agent import _run_nav_episode_body
+
+    result = _run_nav_episode_body(
+        None,
+        "心血管",
+        doc_id="doc1",
+        budget_chars=2000,
+        compose_answer=False,
+        policy="llm",
+        config=NavConfig(policy="llm"),
+        toolspace=_TS(),
+    )
+    assert order == ["plan", "score", "orch"]
+    assert result.section_ids == ["sec_root"]