Staging

CONTEXT.md

@@ -221,7 +221,7 @@ the serving-index latency target until the derived data is repaired.
 ### Retrieval Serving Generation
 
 The namespace-scoped version that identifies one coherent set of active
-document revisions and their serving-index statistics. Retrieval captures one
+document revisions and their serving-index data. Retrieval captures one
 generation and retries or falls back if publication changes it during capture.
 
 ### Retrieval Semantic Parity

apps/api/alembic/versions/9f0a1b2c3d4e_drop_retrieval_namespace_statistics.py

@@ -0,0 +1,119 @@
+"""Drop unused retrieval namespace statistics tables.
+
+Query-time BM25 scoring reads only ``document_map_unit_tokens``,
+``document_map_units``, and ``document_map_unit_indexes``. The per-revision and
+namespace statistics tables were only written by publication/backfill and never
+read by the retrieval path, so they are removed here.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "9f0a1b2c3d4e"
+down_revision: str | None = "8e9f0a1b2c3d"
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+__all__ = [
+    "revision",
+    "down_revision",
+    "branch_labels",
+    "depends_on",
+    "upgrade",
+    "downgrade",
+]
+
+
+def upgrade() -> None:
+    op.drop_index(
+        "idx_retrieval_namespace_token_stats_lookup",
+        table_name="retrieval_namespace_token_stats",
+        if_exists=True,
+    )
+    op.drop_table("retrieval_namespace_token_stats", if_exists=True)
+    op.drop_table("retrieval_namespace_stats", if_exists=True)
+    op.drop_index(
+        "idx_retrieval_serving_revision_stats_scope",
+        table_name="retrieval_serving_revision_stats",
+        if_exists=True,
+    )
+    op.drop_table("retrieval_serving_revision_stats", if_exists=True)
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    if not inspector.has_table("retrieval_serving_revision_stats"):
+        op.create_table(
+            "retrieval_serving_revision_stats",
+            sa.Column("id", sa.String(length=100), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("namespace", sa.String(length=255), nullable=False),
+            sa.Column("document_id", sa.String(length=36), nullable=False),
+            sa.Column("job_result_id", sa.String(length=36), nullable=False),
+            sa.Column("format_version", sa.Integer(), nullable=False),
+            sa.Column("payload_zlib", sa.LargeBinary(), nullable=False),
+            sa.Column("checksum", sa.String(length=64), nullable=False),
+            sa.Column("created_at", sa.DateTime(), nullable=False),
+            sa.ForeignKeyConstraint(
+                ["document_id"], ["documents.document_id"], ondelete="CASCADE"
+            ),
+            sa.ForeignKeyConstraint(
+                ["job_result_id"], ["job_results.id"], ondelete="CASCADE"
+            ),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "document_id",
+                "job_result_id",
+                name="uq_retrieval_serving_revision_stats_revision",
+            ),
+        )
+        op.create_index(
+            "idx_retrieval_serving_revision_stats_scope",
+            "retrieval_serving_revision_stats",
+            ["user_id", "namespace", "document_id", "job_result_id"],
+        )
+    if not inspector.has_table("retrieval_namespace_stats"):
+        op.create_table(
+            "retrieval_namespace_stats",
+            sa.Column("id", sa.String(length=100), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("namespace", sa.String(length=255), nullable=False),
+            sa.Column("generation", sa.BigInteger(), nullable=False),
+            sa.Column("payload_zlib", sa.LargeBinary(), nullable=False),
+            sa.Column("checksum", sa.String(length=64), nullable=False),
+            sa.Column("updated_at", sa.DateTime(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "user_id", "namespace", name="uq_retrieval_namespace_stats_scope"
+            ),
+        )
+    if not inspector.has_table("retrieval_namespace_token_stats"):
+        op.create_table(
+            "retrieval_namespace_token_stats",
+            sa.Column("id", sa.String(length=100), nullable=False),
+            sa.Column("user_id", sa.Text(), nullable=False),
+            sa.Column("namespace", sa.String(length=255), nullable=False),
+            sa.Column("generation", sa.BigInteger(), nullable=False),
+            sa.Column("channel", sa.String(length=32), nullable=False),
+            sa.Column("token_hash", sa.String(length=64), nullable=False),
+            sa.Column("document_frequency", sa.Integer(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "user_id",
+                "namespace",
+                "channel",
+                "token_hash",
+                name="uq_retrieval_namespace_token_stats_key",
+            ),
+        )
+        op.create_index(
+            "idx_retrieval_namespace_token_stats_lookup",
+            "retrieval_namespace_token_stats",
+            ["user_id", "namespace", "generation", "channel", "token_hash"],
+        )

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/app/services/documents/lifecycle_service.py

@@ -8,13 +8,11 @@
 
 from app.repositories.document_repository import DocumentRepository
 from loguru import logger
-from sqlalchemy import delete
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from shared.models.database.document import (
     DocumentChunk,
     DocumentSection,
-    RetrievalServingRevisionStat,
 )
 from shared.services.retrieval.cache_service import (
     invalidate_retrieval_cache_namespaces,
@@ -27,9 +25,6 @@
     advance_namespace_generation,
     lock_namespace_generation,
 )
-from shared.services.retrieval.serving_manifest import (
-    rebuild_namespace_serving_statistics,
-)
 from shared.services.storage.result_storage import ResultStorage, get_result_storage
 
 _DOCUMENT_CHUNK_ASSET_URL_EXPIRES_SECONDS = 7 * 24 * 60 * 60
@@ -469,23 +464,6 @@ async def archive_document(
             )
         )
         await self._repository.archive_document(db, document=document)
-        current_revision = document.current_job_result_id
-        if current_revision:
-            await db.run_sync(
-                lambda sync_db: sync_db.execute(
-                    delete(RetrievalServingRevisionStat).where(
-                        RetrievalServingRevisionStat.document_id == document_id,
-                        RetrievalServingRevisionStat.job_result_id == current_revision,
-                    )
-                )
-            )
-        await db.run_sync(
-            lambda sync_db: rebuild_namespace_serving_statistics(
-                sync_db,
-                user_id=user_id,
-                namespace=previous_namespace,
-            )
-        )
         await db.run_sync(
             lambda sync_db: remove_document_from_namespace_map_snapshot(
                 sync_db,

apps/api/scripts/backfill_map_unit_indexes.py

@@ -1,8 +1,8 @@
 """Backfill persisted MAP-NAV lexical indexes for existing revisions.
 
 Rebuilds, per active revision: the map-unit index, the revision serving
-manifest, that document's subtree in the namespace MAP snapshot, namespace
-statistics, and the namespace generation. The migrations that create these
+manifest, that document's subtree in the namespace MAP snapshot, and the
+namespace generation. The migrations that create these
 derived tables leave them empty intentionally. Run this command after
 deployment with ``--apply`` so each revision is rebuilt and committed
 independently; without ``--apply`` it is a read-only inventory.
@@ -69,7 +69,6 @@ def _bootstrap_python_path() -> None:
 from shared.services.retrieval.serving_manifest import (
     decode_serving_manifest,
     persist_revision_serving_state,
-    rebuild_namespace_serving_statistics,
 )
 
 
@@ -346,11 +345,6 @@ def backfill_map_unit_indexes(*, apply: bool, document_id: str = "") -> int:
             patch_namespace_map_snapshot(
                 db, scope=scope, manifest_payload=manifest_payload
             )
-            rebuild_namespace_serving_statistics(
-                db,
-                user_id=scope.user_id,
-                namespace=scope.namespace,
-            )
             advance_namespace_generation(
                 db,
                 user_id=scope.user_id,

apps/api/tests/contract/test_documents_contract.py

@@ -13,7 +13,6 @@
 
 from tests.support.contract_database import ContractDatabase
 from shared.testing.contract_runtime import get_contract_database_url
-from shared.services.retrieval.serving_manifest import encode_serving_manifest
 
 
 async def _create_contract_engine() -> AsyncEngine:
@@ -1333,84 +1332,3 @@ async def test_should_archive_a_document_via_the_legacy_archive_route(
     assert response_json["archived_at"]
     assert persisted_document["status"] == "archived"
     assert persisted_document["archived_at"] is not None
-
-
-@pytest.mark.asyncio
-async def test_archive_removes_revision_serving_stats_and_advances_generation(
-    developer_api_client_factory: Callable[
-        [], AbstractAsyncContextManager[AsyncClient]
-    ],
-) -> None:
-    document_id = f"doc_{uuid4().hex[:12]}"
-    namespace = f"archive-serving-{uuid4().hex[:8]}"
-    async with developer_api_client_factory() as api_client:
-        revision = await _insert_document_revision_with_chunks(
-            document_id=document_id,
-            namespace=namespace,
-            chunks=[
-                {
-                    "id": f"dchk_{uuid4().hex[:12]}",
-                    "chunk_id": "archive-serving-chunk",
-                    "chunk_type": "text",
-                    "content": "serving contribution",
-                    "source_chunk_path": "Archive/Serving",
-                    "metadata": {},
-                }
-            ],
-        )
-        payload_bytes, checksum, version = encode_serving_manifest(
-            {
-                "document_id": document_id,
-                "job_result_id": revision["job_result_id"],
-                "unit_count": 1,
-                "path_token_count": 1,
-                "content_token_count": 2,
-                "token_frequencies": {
-                    "path": {"archive": 1},
-                    "content": {"serving": 1},
-                },
-            }
-        )
-        await ContractDatabase.execute(
-            """
-            INSERT INTO retrieval_serving_revision_stats (
-                id, user_id, namespace, document_id, job_result_id,
-                format_version, payload_zlib, checksum, created_at
-            ) VALUES (
-                :id, :user_id, :namespace, :document_id, :job_result_id,
-                :format_version, :payload_zlib, :checksum, NOW()
-            )
-            """,
-            {
-                "id": f"rss_{uuid4().hex[:12]}",
-                "user_id": "local-dev-user",
-                "namespace": namespace,
-                "document_id": document_id,
-                "job_result_id": revision["job_result_id"],
-                "format_version": version,
-                "payload_zlib": payload_bytes,
-                "checksum": checksum,
-            },
-        )
-        response = await api_client.post(f"/api/v1/documents/{document_id}/archive")
-
-    assert response.status_code == 200
-    remaining_revision_stats = await ContractDatabase.fetch_one(
-        """
-        SELECT id
-        FROM retrieval_serving_revision_stats
-        WHERE document_id = :document_id AND job_result_id = :job_result_id
-        """,
-        {"document_id": document_id, "job_result_id": revision["job_result_id"]},
-    )
-    namespace_stats = await ContractDatabase.fetch_one(
-        """
-        SELECT generation
-        FROM retrieval_namespace_stats
-        WHERE user_id = :user_id AND namespace = :namespace
-        """,
-        {"user_id": "local-dev-user", "namespace": namespace},
-    )
-    assert remaining_revision_stats is None
-    assert namespace_stats is not None
-    assert int(namespace_stats["generation"]) >= 1

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]