perf(retrieval): drive frequency lookup from token hash

apps/api/tests/contract/test_retrieval_classic_map_unit_contract.py

@@ -106,8 +106,9 @@ def capture_frequency_query(
         _executemany: bool,
     ) -> None:
         if (
-            "FROM document_map_unit_tokens AS tokens" in statement
-            and "tokens.frequency" in statement
+            "FROM document_map_unit_tokens" in statement
+            and "frequency" in statement
+            and "token_hash" in statement
         ):
             statements.append(statement)
 
@@ -168,6 +169,8 @@ def capture_frequency_query(
     assert statements
     assert "token_hash = ANY" in statements[-1]
     assert "token = ANY" not in statements[-1]
+    assert "matching_tokens AS MATERIALIZED" in statements[-1]
+    assert "FROM matching_tokens" in statements[-1]
 
 
 async def test_classic_route_image_filter_scores_only_units_with_images(

apps/api/tests/contract/test_retrieval_lazy_tree_contract.py

@@ -5,6 +5,9 @@
 from shared.services.retrieval.nav.nav_hierarchy import NodeMeta, ProviderToolSpace
 from shared.services.retrieval.nav.nav_knowhere import KnowhereProvider, SectionRow
 from shared.services.retrieval.nav.nav_map_scores import _walk_tree
+from shared.services.retrieval.nav._compat import Chunk
+from shared.services.retrieval.nav.nav_compose import pack_nav_evidence
+from shared.services.retrieval.nav.nav_types import NavConfig, NavState
 
 
 class _MetadataForbiddenProvider(KnowhereProvider):
@@ -31,3 +34,72 @@ def test_tree_walk_reads_children_and_titles_without_materializing_metadata() ->
     assert children == {"root": ["child"], "child": []}
     assert leaves == {"child"}
     assert titles == {"root": "Root", "child": "Child"}
+
+
+def test_evidence_pack_reads_titles_without_materializing_subtree_metadata() -> None:
+    class _CountingMetadataProvider(_MetadataForbiddenProvider):
+        metadata_calls = 0
+
+        def node_meta(self, section_id: str) -> NodeMeta:
+            self.metadata_calls += 1
+            return super().node_meta(section_id)
+
+    provider = _CountingMetadataProvider(
+        doc_id="doc",
+        sections=[
+            SectionRow("root", None, "Root", "Root", 0, "", 0),
+            SectionRow("child", "root", "Root / Child", "Child", 1, "", 1),
+        ],
+        units=(),
+    )
+    toolspace = ProviderToolSpace(provider)
+    state = NavState(doc_id="doc", query="child")
+    chunk = Chunk(
+        node_id="child",
+        doc_id="doc",
+        text="evidence",
+        line_ids=(1,),
+        section_id="child",
+    )
+
+    result = pack_nav_evidence(
+        [(chunk, 1.0)],
+        toolspace,
+        state,
+        NavConfig(),
+        budget_chars=100,
+    )
+
+    assert result.evidence_text == "[E1]\n[§ Child]\nevidence"
+    assert provider.metadata_calls == 0
+
+
+def test_evidence_pack_identifies_header_owners_from_parent_chain() -> None:
+    provider = KnowhereProvider(
+        doc_id="doc",
+        sections=[
+            SectionRow("root", None, "Root", "Root", 0, "", 0),
+            SectionRow("parent", "root", "Root / Parent", "Parent", 1, "", 1),
+            SectionRow(
+                "child", "parent", "Root / Parent / Child", "Child", 2, "", 2
+            ),
+        ],
+        units=(),
+    )
+    toolspace = ProviderToolSpace(provider)
+    state = NavState(doc_id="doc", query="child")
+    chunks = [
+        Chunk("parent", "doc", "parent evidence", (1,), "parent"),
+        Chunk("child", "doc", "child evidence", (2,), "child"),
+    ]
+
+    result = pack_nav_evidence(
+        [(chunks[0], 1.0), (chunks[1], 0.9)],
+        toolspace,
+        state,
+        NavConfig(),
+        budget_chars=200,
+    )
+
+    assert result.kept_chunks == [chunks[1]]
+    assert result.evidence_text == "[E1]\n[§ Child]\nchild evidence"

apps/api/tests/contract/test_retrieval_map_unit_index_contract.py

@@ -15,6 +15,7 @@
 )
 from shared.services.retrieval.nav.nav_hierarchy import ProviderToolSpace
 from shared.services.retrieval.nav._compat import Chunk, EpisodeResult
+from shared.services.retrieval.nav import nav_knowhere
 from shared.services.retrieval.nav.nav_map_scores import (
     build_score_units,
     compute_corpus_map_and_unit_scores,
@@ -41,6 +42,77 @@
 _USER_ID = "local-dev-user"
 
 
+def test_read_only_score_loader_drives_frequency_lookup_from_token_hash(
+    monkeypatch,
+) -> None:
+    document_id = "doc-frequency"
+    job_result_id = "revision-frequency"
+    executions: list[tuple[str, object]] = []
+
+    class FakeCursor:
+        def __init__(self) -> None:
+            self.rows: list[tuple[object, ...]] = []
+
+        def execute(self, statement: str, parameters: object = None) -> None:
+            executions.append((statement, parameters))
+            if "document_map_unit_indexes" in statement:
+                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:
+                self.rows = [("unit-frequency", "path", "retrieval", 1)]
+            else:
+                self.rows = []
+
+        def fetchall(self) -> list[tuple[object, ...]]:
+            return self.rows
+
+        def close(self) -> None:
+            return None
+
+    class FakeConnection:
+        def __init__(self) -> None:
+            self.cursor_instance = FakeCursor()
+
+        def set_session(self, *, readonly: bool, autocommit: bool) -> None:
+            assert readonly is True
+            assert autocommit is True
+
+        def cursor(self) -> FakeCursor:
+            return self.cursor_instance
+
+        def close(self) -> None:
+            return None
+
+    connection = FakeConnection()
+    monkeypatch.setattr(nav_knowhere, "_connect", lambda _dsn: connection)
+    store = nav_knowhere.ReadOnlyChunkStore(
+        dsn="postgresql://test",
+        revisions={document_id: job_result_id},
+    )
+
+    corpus = store.load_persisted_score_corpus(
+        [document_id],
+        {document_id: ["section-frequency"]},
+        ["retrieval"],
+    )
+
+    assert corpus is not None
+    frequency_executions = [
+        (statement, parameters)
+        for statement, parameters in executions
+        if "matching_tokens AS MATERIALIZED" in statement
+    ]
+    assert len(frequency_executions) == 1
+    statement, parameters = frequency_executions[0]
+    assert "FROM matching_tokens" in statement
+    assert "token_hash = ANY" in statement
+    assert isinstance(parameters, list)
+    assert parameters[0] == [
+        "6e51d6a3d90b6a3243d38e6da6b3f31f49867c1360beba83da8ca9630f9672c7"
+    ]
+
+
 class _IncompleteIndexStore:
     """Minimal lazy store whose missing index returns no persisted scores."""