Staging

.github/workflows/build-images.yml

@@ -546,6 +546,8 @@ jobs:
           API_WEBHOOK_ENDPOINT: https://api-staging.knowhereto.ai/v1/internal/s3-events
           SNS_TOPIC_ARN: arn:aws:sns:us-east-1:107424103509:knowhere-staging-s3-events
           QSTASH_CALLBACK_BASE_URL: https://api-staging.knowhereto.ai/api/v1
+          API_CPU: "256"
+          API_MEMORY: "1024"
           WORKER_CPU: "2048"
           WORKER_MEMORY: "4096"
         shell: bash
@@ -772,6 +774,8 @@ jobs:
           API_WEBHOOK_ENDPOINT: https://api.knowhereto.ai/v1/internal/s3-events
           SNS_TOPIC_ARN: arn:aws:sns:us-east-1:107424103509:knowhere-prod-s3-events
           QSTASH_CALLBACK_BASE_URL: https://api.knowhereto.ai/api/v1
+          API_CPU: "512"
+          API_MEMORY: "2048"
           WORKER_CPU: "2048"
           WORKER_MEMORY: "4096"
         shell: bash

apps/api/tests/contract/test_retrieval_classic_map_unit_contract.py

@@ -6,7 +6,7 @@
 from uuid import uuid4
 
 from httpx import AsyncClient
-from sqlalchemy import select
+from sqlalchemy import Engine, event, select
 
 from shared.models.database.document import DocumentMapUnit
 from shared.services.retrieval.publication_content import (
@@ -88,6 +88,91 @@ async def test_classic_route_maps_winning_unit_to_one_chunk(
     }
 
 
+async def test_classic_route_uses_token_hash_lookup_for_frequency_query(
+    developer_api_client_factory: Callable[
+        [], AbstractAsyncContextManager[AsyncClient]
+    ],
+) -> None:
+    identifier = uuid4().hex[:8]
+    namespace = f"classic-token-hash-{identifier}"
+    statements: list[str] = []
+
+    def capture_frequency_query(
+        _connection: Any,
+        _cursor: Any,
+        statement: str,
+        _parameters: Any,
+        _context: Any,
+        _executemany: bool,
+    ) -> None:
+        if (
+            "FROM document_map_unit_tokens" in statement
+            and "frequency" in statement
+            and "token_hash" in statement
+        ):
+            statements.append(statement)
+
+    event.listen(Engine, "before_cursor_execute", capture_frequency_query)
+    try:
+        async with developer_api_client_factory() as api_client:
+            await _publish_document(
+                namespace=namespace,
+                source_file_name="token-hash.pdf",
+                chunks=[
+                    {
+                        "chunk_id": f"token-hash-{identifier}",
+                        "type": "text",
+                        "content": "token hash lookup marker",
+                        "path": "token-hash.pdf/Root/Section/body",
+                        "order": 1,
+                        "metadata": {},
+                    },
+                    {
+                        "chunk_id": f"token-hash-filler-a-{identifier}",
+                        "type": "text",
+                        "content": "unrelated filler a",
+                        "path": "token-hash.pdf/Root/Section/a",
+                        "order": 2,
+                        "metadata": {},
+                    },
+                    {
+                        "chunk_id": f"token-hash-filler-b-{identifier}",
+                        "type": "text",
+                        "content": "unrelated filler b",
+                        "path": "token-hash.pdf/Root/Section/b",
+                        "order": 3,
+                        "metadata": {},
+                    },
+                    {
+                        "chunk_id": f"token-hash-filler-c-{identifier}",
+                        "type": "text",
+                        "content": "unrelated filler c",
+                        "path": "token-hash.pdf/Root/Section/c",
+                        "order": 4,
+                        "metadata": {},
+                    },
+                ],
+            )
+            response = await api_client.post(
+                "/api/v1/retrieval/query",
+                json={
+                    "namespace": namespace,
+                    "query": "token hash lookup",
+                    "top_k": 1,
+                    "use_agentic": False,
+                },
+            )
+    finally:
+        event.remove(Engine, "before_cursor_execute", capture_frequency_query)
+
+    assert response.status_code == 200
+    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(
     developer_api_client_factory: Callable[
         [], AbstractAsyncContextManager[AsyncClient]

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,78 @@
 _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 "FROM document_map_unit_tokens" 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 "FROM document_map_unit_tokens" in statement
+    ]
+    assert len(frequency_executions) == 1
+    statement, parameters = frequency_executions[0]
+    assert "map_unit_id = ANY" in statement
+    assert "token_hash = ANY" in statement
+    assert isinstance(parameters, list)
+    assert parameters[0] == ["unit-frequency"]
+    assert parameters[1] == [
+        "6e51d6a3d90b6a3243d38e6da6b3f31f49867c1360beba83da8ca9630f9672c7"
+    ]
+
+
 class _IncompleteIndexStore:
     """Minimal lazy store whose missing index returns no persisted scores."""
 

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

@@ -36,10 +36,11 @@ def react_budget() -> int:
 Ordered query strategy (follow this order; skip a step only if already tried
 or not applicable to the given title / parent path). Pattern-level only —
 do not invent document-specific titles:
-1. Derive the search line from the given title by removing leading number /
-   letter / punctuation prefixes and trailing metadata qualifiers (document
-   identifiers/codes, revision labels, and similar). Keep the semantic title
-   body. Prefer that body over a metadata-only query when both are present.
+1. Derive the semantic title body from the given title: strip leading
+   number / letter / punctuation prefixes and trailing metadata qualifiers
+   (document identifiers/codes, revision labels, and similar). The given title
+   may be a merged heading (semantic body plus a code); still reduce it to the
+   semantic body first. Grep that body before any metadata-only query.
 2. When the parent path indicates appendices/annexes (or the title is a
    lettered appendix-style entry): grep the structural form
    "Appendix <letter>" using the letter taken from the title. Prefer this

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

@@ -195,15 +195,16 @@ def _try_outline_anchoring_route(
         if (judge_result.payload or {}).get("choice") != OUTLINE_CHOICE:
             return False
 
-    toc_with_level: list[dict[str, Any]] = []
-    for entry in kept:
-        row: dict[str, Any] = {
+    # Judge digest used full ``kept`` (incl. null pages). Anchoring / null-page
+    # ReAct only consume resolvable outline destinations (``paged_kept``).
+    toc_with_level: list[dict[str, Any]] = [
+        {
             "heading": entry["heading"],
             "level": entry["level"],
+            "physical_page": int(entry["page"]),
         }
-        if entry.get("page") is not None:
-            row["physical_page"] = int(entry["page"])
-        toc_with_level.append(row)
+        for entry in paged_kept
+    ]
 
     hierarchy = {
         "source": "pdf_outline",

apps/worker/app/services/document_agent/tools/find_toc_anchor_pages.py

@@ -83,12 +83,30 @@ def _merge_keyword_split_lines(
     return merged
 
 
+def _match_toc_keyword_in_line(normalized_line: str) -> str | None:
+    """Return the longest TOC keyword contained in a normalized line."""
+    if not normalized_line:
+        return None
+    return next(
+        (
+            keyword
+            for keyword in sorted(TOC_KEYWORDS, key=len, reverse=True)
+            if keyword in normalized_line
+        ),
+        None,
+    )
+
+
 def _find_toc_text_matches(lines: list[str]) -> list[dict[str, Any]]:
-    """Match TOC keywords only as whole lines after keyword-split repair."""
+    """Match TOC keywords as line-level containment after keyword-split repair.
+
+    Cross-line handling stays keyword-internal only (e.g. 目+录). Hit rule is
+    ``keyword in normalized_line`` (longest match wins), not whole-line equality.
+    """
     matches: list[dict[str, Any]] = []
     for raw_line, start_idx, end_idx in _merge_keyword_split_lines(lines):
-        keyword = normalize_match_text(raw_line)
-        if keyword not in TOC_KEYWORDS:
+        keyword = _match_toc_keyword_in_line(normalize_match_text(raw_line))
+        if keyword is None:
             continue
         matches.append(
             {
@@ -205,8 +223,9 @@ def _filter_recurring_elements(
 @register_tool(
     name="find.toc_anchor_pages",
     description=(
-        "Scan full PDF page text for whole-line TOC keywords, filter recurring "
-        "navigation elements, then render candidate PNGs for VLM confirmation."
+        "Scan full PDF page text for line-level TOC keywords (containment after "
+        "keyword-split repair), filter recurring navigation elements, then "
+        "render candidate PNGs for VLM confirmation."
     ),
     preconditions=(has_page_labels, has_page_full_text),
 )