fix: harden retrieval serving consistency and incomplete-index fallback

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

@@ -7,7 +7,7 @@
 from app.api.dependencies.current_user import with_current_user
 from app.services.rate_limit.data_structures import CurrentUser
 from fastapi import APIRouter, Depends
-from pydantic import BaseModel, Field, field_validator
+from pydantic import BaseModel, Field, field_validator, model_validator
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from shared.core.database import get_db
@@ -60,10 +60,17 @@ class RetrievalQueryRequest(BaseModel):
     )
     channels: list[str] = Field(
         default_factory=list,
-        description="Channels to run (empty=all). Options: path, content, term",
+        description=(
+            "Deprecated and unsupported by the persisted map-unit route. "
+            "Leave empty; explicit channel selection is rejected."
+        ),
     )
     channel_weights: dict[str, float] = Field(
-        default_factory=dict, description="Per-channel weight overrides"
+        default_factory=dict,
+        description=(
+            "Deprecated and unsupported by the persisted map-unit route. "
+            "Leave empty; explicit overrides are rejected."
+        ),
     )
     rerank: bool = Field(False, description="Enable LLM reranking after RRF fusion")
     threshold: float = Field(0.0, ge=0.0, description="Minimum RRF score threshold")
@@ -112,6 +119,20 @@ def validate_chunk_types(cls, v: list[str] | None) -> list[str] | None:
     def normalize_namespace(cls, namespace: str | None) -> str:
         return normalize_retrieval_namespace(namespace)
 
+    @model_validator(mode="after")
+    def reject_unsupported_channel_controls(self) -> "RetrievalQueryRequest":
+        if self.channels:
+            raise ValueError(
+                "channels is deprecated and unsupported; omit it and use the "
+                "persisted path/content map-unit scorer"
+            )
+        if self.channel_weights:
+            raise ValueError(
+                "channel_weights is deprecated and unsupported; omit it and use "
+                "the persisted path/content map-unit scorer"
+            )
+        return self
+
 
 class RetrievalQueryResponse(BaseModel):
     namespace: str

apps/api/tests/contract/test_retrieval_contract.py

@@ -306,6 +306,28 @@ async def test_should_return_request_validation_failure_for_an_invalid_channel(
     assert "Invalid channel" in cast(str, violations[0]["description"])
 
 
+async def test_should_reject_legacy_channel_controls(
+    developer_api_client_factory: Callable[
+        [], AbstractAsyncContextManager[AsyncClient]
+    ],
+) -> None:
+    async with developer_api_client_factory() as api_client:
+        response = await api_client.post(
+            "/api/v1/retrieval/query",
+            json={
+                "namespace": "default",
+                "query": "alpha",
+                "channels": ["content"],
+            },
+        )
+
+    assert response.status_code == 400
+    response_json = cast(dict[str, object], response.json())
+    error = cast(dict[str, object], response_json["error"])
+    assert error["code"] == "INVALID_ARGUMENT"
+    assert "deprecated and unsupported" in str(error)
+
+
 async def test_should_exclude_matching_document_ids_from_the_response(
     developer_api_client_factory: Callable[
         [], AbstractAsyncContextManager[AsyncClient]

apps/api/tests/contract/test_retrieval_lazy_snapshot_quality_contract.py

@@ -165,10 +165,7 @@ def test_lazy_provider_preserves_score_units_and_scores() -> None:
     lazy_scores = compute_corpus_map_and_unit_scores(
         lazy, doc_ids=["doc"], query="alpha retrieval"
     )
-    assert eager_scores[1] == {}
-    assert lazy_scores[1] == {}
-    assert all(score == 0.0 for score in eager_scores[0].values())
-    assert all(score == 0.0 for score in lazy_scores[0].values())
+    assert eager_scores == lazy_scores
 
     lazy_provider = lazy._provider
     self_units = getattr(lazy_provider, "self_units")
@@ -245,7 +242,7 @@ def build_stats(search_field: str) -> PersistedBm25Stats:
     assert scored["beta evidence"]["unit-b"] > scored["beta evidence"]["unit-a"]
 
 
-def test_missing_index_does_not_read_chunk_payloads() -> None:
+def test_missing_index_falls_back_to_legacy_payload_scoring() -> None:
     _eager, lazy, store = _providers()
     queries: list[str] = ["alpha retrieval", "supporting image"]
     store.section_loads = 0
@@ -256,8 +253,8 @@ def test_missing_index_does_not_read_chunk_payloads() -> None:
     )
 
     assert set(actual) == set(queries)
-    assert all(unit_scores == {} for _map_scores, unit_scores in actual.values())
-    assert store.section_loads == 0
+    assert any(unit_scores for _map_scores, unit_scores in actual.values())
+    assert store.section_loads > 0
 
 
 def test_missing_index_is_empty_across_documents() -> None:
@@ -276,8 +273,8 @@ def test_missing_index_is_empty_across_documents() -> None:
     )
 
     assert actual == expected
-    assert all(unit_scores == {} for _map_scores, unit_scores in actual.values())
-    assert store.section_loads == 0
+    assert any(unit_scores for _map_scores, unit_scores in actual.values())
+    assert store.section_loads > 0
 
 
 def test_native_chunk_store_strips_async_driver_from_database_url(

apps/api/tests/contract/test_retrieval_map_unit_index_contract.py

@@ -276,8 +276,8 @@ def reject_payload_read(
 
     assert any(score > 0.0 for score in actual_scores[1].values())
     assert select_map_highlights(actual_scores[1], k=3)
-    assert fallback_scores[1] == {}
-    assert all(score == 0.0 for score in fallback_scores[0].values())
+    assert any(score > 0.0 for score in fallback_scores[1].values())
+    assert any(score > 0.0 for score in fallback_scores[0].values())
 
 
 async def test_lazy_snapshot_defers_selected_asset_reference_metadata(
@@ -476,8 +476,8 @@ def test_incomplete_index_returns_empty_scores() -> None:
         lazy, doc_ids=["doc-a", "doc-b"], query="alpha beta"
     )
 
-    assert actual[1] == {}
-    assert expected[1] == {}
+    assert set(actual[1]) == {"leaf-a", "leaf-b"}
+    assert set(expected[1]) == {"leaf-a", "leaf-b"}
     assert all(score == 0.0 for score in actual[0].values())
     assert store.persisted_loads == 1
 

packages/shared-python/shared/services/retrieval/cache_service.py

@@ -10,13 +10,45 @@
 
 _RETRIEVAL_CACHE_TTL_SECONDS = 300
 _VERSION_FALLBACK = 0
+_INDEX_READINESS_TTL_SECONDS = 60
 
 
 def _namespace_version_key(*, user_id: str, namespace: str) -> str:
     namespace = normalize_retrieval_namespace(namespace)
     return f"retrieval:version:{user_id}:{namespace}"
 
 
+def _namespace_index_readiness_key(*, user_id: str, namespace: str) -> str:
+    namespace = normalize_retrieval_namespace(namespace)
+    return f"retrieval:index-readiness:{user_id}:{namespace}"
+
+
+async def record_retrieval_index_readiness(
+    *,
+    user_id: str,
+    namespace: str,
+    ready: bool,
+    expected_revisions: int,
+    indexed_revisions: int,
+) -> None:
+    """Publish a short-lived index readiness signal for operators and callers.
+
+    Redis is deliberately only a status cache. PostgreSQL generations and
+    serving rows remain the source of truth, and retrieval must continue to
+    work if Redis is unavailable.
+    """
+    redis_service = RedisServiceFactory.get_service()
+    await redis_service.set(
+        _namespace_index_readiness_key(user_id=user_id, namespace=namespace),
+        {
+            "ready": bool(ready),
+            "expected_revisions": int(expected_revisions),
+            "indexed_revisions": int(indexed_revisions),
+        },
+        ex=_INDEX_READINESS_TTL_SECONDS,
+    )
+
+
 def _normalize_exclude_sections(exclude_sections: list[dict[str, str]]) -> list[str]:
     normalized: list[str] = []
     for item in exclude_sections:

packages/shared-python/shared/services/retrieval/nav/nav_knowhere.py

@@ -300,6 +300,7 @@ def load_persisted_score_corpus(
         comes from ``document_map_unit_indexes`` (written at index time).
         """
         from shared.services.retrieval.nav.persisted_score_load import (
+            average_idf_from_namespace_stats,
             build_channel_bm25_stats,
             combine_average_idf,
         )
@@ -359,12 +360,51 @@ def load_persisted_score_corpus(
                     revision_key
                 ]
             else:
-                average_idf_path = combine_average_idf(
-                    [(float(row[4] or 0.0), int(row[3] or 0)) for row in index_rows]
-                )
-                average_idf_content = combine_average_idf(
-                    [(float(row[5] or 0.0), int(row[3] or 0)) for row in index_rows]
-                )
+                total_unit_count = sum(int(row[3] or 0) for row in index_rows)
+                try:
+                    cur.execute(
+                        "SELECT tokens.channel, tokens.token, "
+                        "COUNT(DISTINCT tokens.map_unit_id) "
+                        "FROM document_map_unit_tokens AS tokens "
+                        "JOIN document_map_units AS units ON units.id = 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 "
+                        "WHERE tokens.channel = ANY(%s) "
+                        "GROUP BY tokens.channel, tokens.token",
+                        [*revision_params, ["path", "content"]],
+                    )
+                    namespace_token_dfs: dict[str, list[int]] = {
+                        "path": [],
+                        "content": [],
+                    }
+                    for channel, _token, document_frequency in cur.fetchall():
+                        if str(channel) in namespace_token_dfs:
+                            namespace_token_dfs[str(channel)].append(
+                                int(document_frequency)
+                            )
+                    average_idf_path = average_idf_from_namespace_stats(
+                        unit_count=total_unit_count,
+                        token_document_frequencies=namespace_token_dfs["path"],
+                    )
+                    average_idf_content = average_idf_from_namespace_stats(
+                        unit_count=total_unit_count,
+                        token_document_frequencies=namespace_token_dfs["content"],
+                    )
+                except Exception as exc:
+                    _logger.warning(
+                        "exact namespace IDF load failed; using revision averages: %s",
+                        exc,
+                    )
+                    average_idf_path = combine_average_idf(
+                        [(float(row[4] or 0.0), int(row[3] or 0)) for row in index_rows]
+                    )
+                    average_idf_content = combine_average_idf(
+                        [
+                            (float(row[5] or 0.0), int(row[3] or 0))
+                            for row in index_rows
+                        ]
+                    )
                 self._score_average_idf_cache[revision_key] = (
                     average_idf_path,
                     average_idf_content,

packages/shared-python/shared/services/retrieval/nav/nav_map_scores.py

@@ -8,12 +8,89 @@
     build_content_search_text,
     build_path_search_text,
     build_term_search_text,
+    PersistedScoreCorpus,
+    PersistedScoreUnit,
     score_persisted_corpus_many,
 )
+from .persisted_score_load import (
+    average_idf_from_unit_dfs,
+    build_channel_bm25_stats,
+)
 
 _logger = logging.getLogger(__name__)
 
 
+def _build_legacy_score_corpus(ts: Any, doc_ids: Sequence[str]) -> PersistedScoreCorpus:
+    """Build the retired in-memory scorer input when persisted indexes are absent."""
+    raw_units: List[dict] = []
+    for doc_id in doc_ids:
+        raw_units.extend(build_score_units(ts, doc_id))
+    frequencies: Dict[Tuple[str, str], Dict[str, int]] = {}
+    unit_rows: List[dict] = []
+    path_dfs: Dict[str, int] = {}
+    content_dfs: Dict[str, int] = {}
+    for unit in raw_units:
+        unit_id = str(unit.get("chunk_id") or "").strip()
+        if not unit_id:
+            continue
+        path_tokens = str(unit.get("path_search_text") or "").split()
+        content_tokens = str(unit.get("content_search_text") or "").split()
+        path_freq: Dict[str, int] = {}
+        content_freq: Dict[str, int] = {}
+        for token in path_tokens:
+            path_freq[token] = path_freq.get(token, 0) + 1
+        for token in content_tokens:
+            content_freq[token] = content_freq.get(token, 0) + 1
+        frequencies[(unit_id, "path")] = path_freq
+        frequencies[(unit_id, "content")] = content_freq
+        for token in path_freq:
+            path_dfs[token] = path_dfs.get(token, 0) + 1
+        for token in content_freq:
+            content_dfs[token] = content_dfs.get(token, 0) + 1
+        unit_rows.append(
+            {
+                "unit_id": unit_id,
+                "path_length": len(path_tokens),
+                "content_length": len(content_tokens),
+            }
+        )
+    unit_count = len(unit_rows)
+    return PersistedScoreCorpus(
+        units=[
+            PersistedScoreUnit(
+                unit_id=str(row["unit_id"]),
+                path_length=int(row["path_length"]),
+                content_length=int(row["content_length"]),
+                path_frequencies=frequencies[(str(row["unit_id"]), "path")],
+                content_frequencies=frequencies[(str(row["unit_id"]), "content")],
+            )
+            for row in unit_rows
+        ],
+        path_stats=build_channel_bm25_stats(
+            unit_rows=unit_rows,
+            map_unit_id_field="unit_id",
+            length_field="path_length",
+            channel="path",
+            query_tokens=list(path_dfs),
+            frequencies=frequencies,
+            average_idf=average_idf_from_unit_dfs(
+                unit_count=unit_count, token_document_frequency=path_dfs
+            ),
+        ),
+        content_stats=build_channel_bm25_stats(
+            unit_rows=unit_rows,
+            map_unit_id_field="unit_id",
+            length_field="content_length",
+            channel="content",
+            query_tokens=list(content_dfs),
+            frequencies=frequencies,
+            average_idf=average_idf_from_unit_dfs(
+                unit_count=unit_count, token_document_frequency=content_dfs
+            ),
+        ),
+    )
+
+
 def _children_ids(ts: Any, section_id: str, doc_id: str) -> List[str]:
     children_fn = getattr(ts, "_children_for_section_path", None)
     if not callable(children_fn):
@@ -346,6 +423,13 @@ def compute_corpus_map_and_unit_scores_many(
         time.perf_counter() - loader_started,
         persisted_corpus is not None,
     )
+    if persisted_corpus is None:
+        _logger.warning(
+            "retrieval map index unavailable; using bounded legacy in-memory scorer "
+            "documents=%d",
+            len(valid_doc_ids),
+        )
+        persisted_corpus = _build_legacy_score_corpus(ts, valid_doc_ids)
     score_started = time.perf_counter()
     unit_scores_by_query = (
         score_persisted_corpus_many(persisted_corpus, unique_queries)

packages/shared-python/shared/services/retrieval/nav/persisted_score_load.py

@@ -38,6 +38,29 @@ def combine_average_idf(parts: Sequence[tuple[float, int]]) -> float:
     )
 
 
+def average_idf_from_namespace_stats(
+    *,
+    unit_count: int,
+    token_document_frequencies: Sequence[int],
+) -> float:
+    """Compute the exact namespace-level average IDF used by rank_bm25.
+
+    Namespace token statistics already contain one document frequency per
+    token. Computing the mean from those rows avoids the incorrect
+    per-revision-average approximation when a namespace contains revisions
+    with different token distributions.
+    """
+    if unit_count <= 0:
+        return 0.0
+    idfs = [
+        math.log(unit_count - int(frequency) + 0.5)
+        - math.log(int(frequency) + 0.5)
+        for frequency in token_document_frequencies
+        if 0 < int(frequency) <= unit_count
+    ]
+    return sum(idfs) / len(idfs) if idfs else 0.0
+
+
 def build_channel_bm25_stats(
     *,
     unit_rows: Sequence[Mapping[str, Any]],