From 1155cd7ba6441ba19f9375e78ab5eed4f479f023 Mon Sep 17 00:00:00 2001
From: chengke <404835780@qq.com>
Date: Tue, 1 Sep 2026 10:17:56 +0800
Subject: [PATCH] refactor: remove RetrievalServingRevisionStat and related
 statistics handling

This commit removes the RetrievalServingRevisionStat model and its associated logic from the document lifecycle and backfill processes. The changes include the removal of calls to rebuild namespace serving statistics and the deletion of related database entries, streamlining the document archiving process. Additionally, the backfill script has been updated to reflect these changes, ensuring it no longer attempts to rebuild statistics for archived documents.
---
 CONTEXT.md                                    |   2 +-
 ...d4e_drop_retrieval_namespace_statistics.py | 119 +++++
 .../services/documents/lifecycle_service.py   |  22 -
 apps/api/scripts/backfill_map_unit_indexes.py |  10 +-
 .../tests/contract/test_documents_contract.py |  82 ----
 ...mically-publish-retrieval-serving-index.md |   2 +-
 ...-coherent-retrieval-serving-generations.md |   4 +-
 docs/design/retrieval-serving-index-plan.md   | 425 ------------------
 .../shared/models/database/document.py        |  99 ----
 .../retrieval/namespace_map_snapshot.py       |   8 +-
 .../services/retrieval/publication_service.py |  13 -
 .../services/retrieval/serving_manifest.py    | 203 +--------
 12 files changed, 129 insertions(+), 860 deletions(-)
 create mode 100644 apps/api/alembic/versions/9f0a1b2c3d4e_drop_retrieval_namespace_statistics.py
 delete mode 100644 docs/design/retrieval-serving-index-plan.md

diff --git a/CONTEXT.md b/CONTEXT.md
index b67e16d0..8609579f 100644
--- a/CONTEXT.md
+++ b/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
diff --git a/apps/api/alembic/versions/9f0a1b2c3d4e_drop_retrieval_namespace_statistics.py b/apps/api/alembic/versions/9f0a1b2c3d4e_drop_retrieval_namespace_statistics.py
new file mode 100644
index 00000000..55069014
--- /dev/null
+++ b/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"],
+        )
diff --git a/apps/api/app/services/documents/lifecycle_service.py b/apps/api/app/services/documents/lifecycle_service.py
index 3ea08f05..fd97a9ba 100644
--- a/apps/api/app/services/documents/lifecycle_service.py
+++ b/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,
diff --git a/apps/api/scripts/backfill_map_unit_indexes.py b/apps/api/scripts/backfill_map_unit_indexes.py
index 6eed78ef..ebf1a04d 100644
--- a/apps/api/scripts/backfill_map_unit_indexes.py
+++ b/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,
diff --git a/apps/api/tests/contract/test_documents_contract.py b/apps/api/tests/contract/test_documents_contract.py
index a71d4029..3e33f202 100644
--- a/apps/api/tests/contract/test_documents_contract.py
+++ b/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
diff --git a/docs/adr/0006-atomically-publish-retrieval-serving-index.md b/docs/adr/0006-atomically-publish-retrieval-serving-index.md
index 6f0b0800..49e3e472 100644
--- a/docs/adr/0006-atomically-publish-retrieval-serving-index.md
+++ b/docs/adr/0006-atomically-publish-retrieval-serving-index.md
@@ -2,5 +2,5 @@
 
 - Status: Accepted
 - Context: Retrieval will use a persistent derived serving index to avoid rebuilding a large namespace on every first request. A document revision without a complete index would have unpredictable latency and could produce inconsistent scoring metadata.
-- Decision: Build the serving manifest and scoring statistics in the same database transaction as the document revision. Write the completeness marker last. If serving-index construction fails, roll back the publication and retry the job; do not expose an active revision with a partial serving index.
+- Decision: Build the map-unit index and serving manifest in the same database transaction as the document revision. Write the completeness marker last. If serving-index construction fails, roll back the publication and retry the job; do not expose an active revision with a partial serving index.
 - Consequences: Active revisions have a simple completeness invariant and predictable first-request behavior. Publication takes more work and storage, and an index failure can delay publication, but retrieval can retain a guarded legacy fallback for migrations or already-existing incomplete revisions.
diff --git a/docs/adr/0007-use-coherent-retrieval-serving-generations.md b/docs/adr/0007-use-coherent-retrieval-serving-generations.md
index 89d2ca05..f875480d 100644
--- a/docs/adr/0007-use-coherent-retrieval-serving-generations.md
+++ b/docs/adr/0007-use-coherent-retrieval-serving-generations.md
@@ -1,6 +1,6 @@
 # Use coherent retrieval-serving generations
 
 - Status: Accepted
-- Context: A namespace can contain many active document revisions, and publication can replace them while a retrieval request is loading serving metadata and scoring statistics.
+- Context: A namespace can contain many active document revisions, and publication can replace them while a retrieval request is loading serving metadata.
 - Decision: Assign each namespace a serving generation. Retrieval captures one generation and verifies it across serving reads; if it changes, retry once and use the exact legacy path if consistency cannot be established.
-- Consequences: Retrieval never combines incompatible revision metadata and scoring statistics. Publication and retrieval need a small amount of generation bookkeeping, and rare concurrent updates may cause a retry or slower fallback.
+- Consequences: Retrieval never combines incompatible revision metadata. Publication and retrieval need a small amount of generation bookkeeping, and rare concurrent updates may cause a retry or slower fallback.
diff --git a/docs/design/retrieval-serving-index-plan.md b/docs/design/retrieval-serving-index-plan.md
deleted file mode 100644
index 5e66f6d6..00000000
--- a/docs/design/retrieval-serving-index-plan.md
+++ /dev/null
@@ -1,425 +0,0 @@
-# Retrieval-serving index: online performance plan
-
-**Status:** Proposed for review
-**Reviewed against:** current `knowhere` retrieval and publication code, 2026-08-29
-**Scope:** first-request retrieval performance; LLM planner/harvest/control time is excluded
-
-## 1. Goal and non-goals
-
-The target is predictable, bounded **Retrieval Non-LLM Work** on the current
-production-sized namespace (about 643 active documents, 50k sections, and 60k
-chunks). The measured baseline is roughly 160-170 seconds before map-nav's LLM
-episode begins. We do not require a one-second absolute target in this phase;
-we require that the request path avoid repeated full-corpus work and have a
-clear linear/bounded complexity profile.
-
-The target includes:
-
-- snapshot or serving-index loading;
-- classic or map-nav lexical scoring;
-- ranking;
-- selected-result hydration;
-- citation and asset-reference assembly.
-
-It excludes planner, harvest, control, and answer-generation model time. Those
-stages must continue to have separate timings.
-
-The plan does not change prompts, models, tokenization, BM25 formulas, RRF
-weights, cache semantics, citation rules, or public HTTP response shapes.
-It does not add LLM-response caching, process-wide serving caches, or startup
-prewarming. The acceptance benchmark is a cold, uncached retrieval request,
-and planner/harvest/control model time remains a separately reported
-dependency. Episode-local reuse is allowed only while one request is active.
-
-## 2. Verified current behavior
-
-The current code has two retrieval routes in
-`shared/services/retrieval/execution/routes.py`:
-
-- `use_agentic=false` runs `bottom_discovery()` and then ranking/hydration.
-- The default map-nav route calls `load_nav_snapshot(..., lazy=True)`, runs the
-  synchronous navigation episode, then opens a fresh database context for
-  reference resolution and result assembly.
-
-`load_nav_snapshot()` currently:
-
-1. Reads active documents and their current job-result IDs.
-2. Loads all matching sections into memory.
-3. Loads all chunk identities and `connect_to` metadata into memory.
-4. Uses a lazy store for selected chunk content and asset paths.
-
-The current persisted map scorer in
-`shared/services/retrieval/nav/nav_knowhere.py` still loads all eligible map
-units, then reads query frequencies and term scores from the persisted tables.
-The scorer itself is fast; the broad database projection is not.
-
-`bottom_discovery()` currently executes path, content, and term channels
-sequentially. `term_channel()` uses substring predicates over lowercased text,
-without a trigram index.
-
-Publication currently writes sections/chunks and then calls
-`replace_document_map_units()` in the same SQLAlchemy transaction. The existing
-`document_map_unit_indexes` row is a per-document-revision completeness marker.
-The existing backfill script only rebuilds that map-unit index; it does not yet
-build the proposed serving manifest or namespace statistics.
-
-The working tree already contains an uncommitted keyset-pagination change and
-the `3f4a5b6c7d8e` section-order migration. Keep those changes separate from
-the serving-index implementation and from unrelated documentation edits.
-
-## 3. Architecture decision
-
-Use PostgreSQL as the source of truth and add a persistent, revision-pinned
-serving read model. Do not add OpenSearch, Elasticsearch, Tantivy, or another
-search service in this version. PostgreSQL's existing FTS plus `pg_trgm` keeps
-the current scoring and tie-breaking behavior directly testable.
-
-### 3.1 Revision serving manifest
-
-Add one compressed manifest row per `(document_id, job_result_id)`. The payload
-contains ordered metadata only:
-
-- document, revision, source filename, and job identity;
-- section IDs, parent IDs, paths, titles, levels, summaries, and sort order;
-- chunk IDs, section IDs, types, sort order, and `connect_to` target IDs;
-- map-unit row IDs, unit IDs, unit kinds, token lengths, and sort order;
-- root-asset IDs and remounted asset owners.
-
-It must not contain full chunk content or asset file paths. Those remain in the
-canonical tables and are loaded lazily for selected evidence.
-
-Store the payload as canonical JSON compressed with the standard-library zlib
-implementation, with a format version and checksum. The serving loader must
-reject an unknown version, checksum mismatch, or incomplete payload.
-
-### 3.2 Serving generations and statistics
-
-Add namespace-scoped generation metadata and persistent scoring statistics:
-
-- `retrieval_namespace_generations`: current generation per user/namespace;
-- `retrieval_serving_revision_stats`: compressed per-revision contributions;
-- `retrieval_namespace_stats`: aggregate unit counts, total lengths, vocabulary
-  frequency histograms, and generation;
-- `retrieval_namespace_token_stats`: queryable document frequency per channel
-  and token hash.
-
-The generation is a consistency marker, not a replacement for revision IDs.
-Retrieval captures active revision IDs and one generation. If generation changes
-while the snapshot is being captured or before scoring starts, retry once; if
-consistency still cannot be proven, use the exact legacy reader.
-
-Every retrieval route must carry that capture as an immutable revision pin set:
-`{document_id, namespace, job_result_id}` plus the captured generation. The pin
-set is the source of truth for the request after capture. Downstream queries
-must constrain sections, map units, chunks, connected assets, ranking lookups,
-reference resolution, and result assembly by the pinned `job_result_id`; they
-must not re-join through the live `Document.current_job_result_id`. A generation
-change after scoring has started must never cause a mix of old and new rows:
-finish against the captured pins (or return an exact legacy result), and only
-retry before work that depends on the snapshot begins.
-Snapshot admission is therefore decided at capture time. If a later archive
-must suppress an in-flight result, discard/retry the whole request; do not
-replace its pinned revision with the document's new current revision.
-
-Cache hits occur before route execution, so every operation that changes the
-serving generation (publication, republish, archive, or namespace move) must
-also advance the namespace retrieval-cache version, or store the generation in
-the cache entry and reject mismatches. This keeps cached responses from
-outliving the generation they represent without changing the public response
-shape.
-
-### 3.3 Indexes
-
-Additive migrations should provide:
-
-- a token-first covering index for map-unit token candidates. The existing
-  `idx_document_map_unit_tokens_lookup` is token-first but does not cover the
-  selected columns; the pending `2e3f4a5b6c7d` migration adds a unit-first
-  covering index for a different access pattern, so the serving reader may
-  need one additional token-first covering index;
-- a revision/section lookup index for map units;
-- a trigram GIN index on `document_map_units.term_search_text_lower`;
-- a generated lowercased term field and trigram GIN index for
-  `document_chunks.term_search_text`;
-- the existing chunk ordering index plus the pending token-covering and
-  section-order migrations (`2e3f4a5b6c7d` and `3f4a5b6c7d8e`).
-
-Enable PostgreSQL's built-in `pg_trgm` extension. No separate search service is
-required.
-
-## 4. Retrieval changes
-
-Capture the revision pin set and generation at retrieval-route entry, before
-the small-corpus count or route selection. Pass that capture into whichever
-route is selected; a route-local capture is allowed only when it is performed
-as the same snapshot transaction. This prevents the count/load pair in the
-small-corpus optimization from straddling a publication.
-
-### 4.1 Fast map-nav snapshot
-
-Extend `load_nav_snapshot()` to try the serving manifest first:
-
-1. Capture active documents, current revision IDs, and namespace generation in a
-   short read-only transaction, returning the immutable revision pin set with
-   the snapshot.
-2. Fetch one manifest row per active revision.
-3. Decode and validate manifests.
-4. Apply the existing document and section exclusion predicates.
-5. Build the current `LazyKnowhereProvider` and `LazyChunkRefIndex` from the
-   decoded metadata.
-6. Pin the lazy chunk store to the captured revision IDs.
-7. Verify generation stability before returning the snapshot.
-
-For an unfiltered map-nav request, route selection may count chunks from these
-same validated manifests instead of scanning `document_chunks`; filtered and
-classic requests retain the exact SQL counter. The count shortcut must fall
-back when any manifest is missing or invalid.
-
-If any manifest is absent or invalid, use the existing legacy snapshot loader.
-The legacy loader must return the same revision pin set and apply the same
-downstream predicates. This fallback is automatic and exact; it is not a public
-feature flag.
-
-The serving path must preserve current ordering, duplicate bare/document-scoped
-reference keys, root-asset remounting, section filtering, and revision pinning.
-Keep the pin set available through the complete map-nav request. After the LLM
-episode, either materialize selected rows (including connected assets) from the
-pinned lazy store before closing it, or pass the pin set to
-`resolve_workflow_references()` and `assemble_retrieval_results()`. Their SQL
-must select the captured `(document_id, job_result_id)` rows directly, so a
-republish during the episode cannot make final citations resolve against the
-new current revision.
-
-### 4.2 Exact persisted map scoring
-
-Keep `PersistedScoreCorpus` and the existing scorer unchanged wherever possible.
-Replace only the data-loading strategy:
-
-- Prepare the immutable, revision-pinned unit projection and namespace scoring
-  statistics once per retrieval episode. Checklist relight waves must reuse that
-  projection; they may fetch or compute only query-specific postings/scores.
-  A wave must not issue another full-namespace unit/statistics load for the same
-  pin set. Instrument the loader call count and include it in the benchmark
-  report so repeated projection loads cannot hide behind separate wave timings.
-- use manifest map-unit metadata to represent all units, including zero-score
-  units; the serving reader should not re-query `document_map_units` for these
-  IDs, lengths, or section membership;
-- query token postings only for tokens in the request;
-- filter postings by captured revisions and allowed sections;
-- discover term-channel candidates through the trigram index while retaining the
-  current exact substring/token-hit scoring. The candidate predicate must use
-  the trigram-indexed `LIKE '%term%'` form (with the same lowercased query and
-  tokens), then apply the existing exact score expression; do not scan every
-  unit's term text in Python.
-- obtain normal-corpus lengths, document frequencies, and IDF-flooring data from
-  persistent statistics;
-- preserve the existing lexical sort key and RRF ranking.
-
-Queries with document or section exclusions must remain exact. If adjusted
-statistics cannot be calculated with certainty, use the legacy scorer for that
-request rather than approximating them.
-
-### 4.3 Classic retrieval — one pinned revision snapshot
-
-Keep the existing channel implementations and result projection. In
-`bottom_discovery()`:
-
-- capture one revision pin set and generation before starting any channel;
-- execute enabled channels concurrently;
-- give each channel its own short-lived database session;
-- pass the same pin set to every channel and constrain every channel query to
-  those revisions;
-- preserve channel limits, Python BM25, term scoring, RRF merge, score
-  normalization, and all-or-error behavior;
-- use the new trigram index only to narrow term candidates.
-
-Do not share one `AsyncSession` across concurrent channel tasks.
-Ranking lookups, duplicate suppression, connected-target hydration, and final
-assembly must receive the same pin set as discovery. The classic result must
-therefore contain rows from one revision per document even if publication
-replaces a document while one of the channel sessions is running. The
-small-corpus optimization must use this same captured snapshot/pin contract (or
-the exact legacy equivalent), rather than loading all rows through live current
-revision joins.
-
-## 5. Publication and lifecycle behavior
-
-Refactor publication so the same build pass produces:
-
-- canonical sections/chunks;
-- existing map-unit rows and completeness marker;
-- the revision serving manifest;
-- revision statistics and namespace-statistics deltas.
-
-All of this happens synchronously in the existing publication transaction. The
-completeness marker and generation update are written last. If serving-index
-construction fails, the publication transaction rolls back.
-
-New publication remains online during backfill. First publication, republish,
-archive, and namespace-move paths must update statistics under the same
-namespace generation row lock. The lock covers the active revision set,
-namespace membership, revision contributions, and the generation increment, so
-readers and writers have one lifecycle ordering.
-
-Backfill must rebuild the complete derived serving state (map units, manifest,
-revision contribution, and namespace-statistics delta), not only the existing
-map-unit index. It must select only documents with `status = 'active'`, a non-null
-`current_job_result_id`, and the intended user/namespace. Immediately before
-writing a contribution, it must hold the namespace lock and re-read the
-document, then require all of the following to remain true: active status,
-unchanged user/namespace, and `current_job_result_id` equal to the captured
-revision. Otherwise it skips that revision without adding statistics. This
-active-status predicate is required in the selector as well as in the
-commit-time guard; update `apps/api/scripts/backfill_map_unit_indexes.py` to
-include it in the existing selector.
-
-Archiving must atomically remove or invalidate that document revision's serving
-statistics contribution while holding the same lock and advance the namespace
-generation. `archive` currently changes `status` without clearing
-`current_job_result_id`, so checking the revision pointer alone is insufficient
-and would allow an in-flight backfill to re-add an archived revision.
-
-## 6. Online rollout
-
-There is no planned downtime, runtime feature flag, or production shadow-read
-mode.
-
-1. Deploy additive schema/index migrations, beginning with the pending
-   `2e3f4a5b6c7d` and `3f4a5b6c7d8e` migrations.
-2. Deploy code that automatically uses the serving reader only for complete,
-   valid revisions and otherwise uses the legacy reader.
-3. Run an explicit, idempotent, bounded backfill for existing active revisions.
-4. Keep retrieval and publication online while backfill runs.
-5. Verify manifest checksums, revision coverage, namespace statistics, and
-   generation consistency.
-6. Run strict legacy-versus-serving differential checks before considering the
-   rollout complete.
-
-If backfill is incomplete, affected revisions continue on the exact legacy
-path. If online serving data is corrupted, reject it, alert, repair it with the
-backfill/rebuild script, and do not serve partial data.
-
-Before enabling the serving reader for a namespace, record an inventory of
-active `(document_id, current_job_result_id)` pairs, manifest completeness, and
-expected per-revision and aggregate unit counts. After backfill, reconcile those
-same values and verify that every aggregate includes only active, namespace-
-member revisions. Abort the fast-path rollout on any missing/extra revision,
-checksum failure, count mismatch, archived contribution, or generation
-discontinuity.
-
-Any migration, backfill, or other database write—especially against
-production—requires explicit approval immediately before execution. Read-only
-inspection and benchmarking may proceed without that approval.
-
-## 6.1 DevOps operations runbook
-
-DevOps owns the production rollout mechanics; application code does not run a
-startup backfill or create serving tables implicitly. Execute the following in
-order:
-
-1. **Preflight (read-only):** confirm the target account, database, migration
-   head, available disk, connection headroom, and a recent rollback point. Record
-   the active `(document_id, current_job_result_id)` inventory for each namespace
-   that will be backfilled.
-2. **Schema rollout:** with explicit approval immediately beforehand, apply the
-   additive migrations in dependency order: `2e3f4a5b6c7d`,
-   `3f4a5b6c7d8e`, `4a5b6c7d8e9f`, then `5b6c7d8e9f0a`. Run the trigram-index
-   migration during a low-traffic window and monitor for blocking locks.
-3. **Application rollout:** deploy the API and worker versions containing the
-   serving reader and atomic publication changes. Verify health, error rate, and
-   legacy fallback before starting the backfill.
-4. **Bounded backfill:** with separate approval, run
-   `uv run python apps/api/scripts/backfill_map_unit_indexes.py --apply` from a
-   controlled operator environment. Limit concurrency, pause on database
-   saturation, and resume safely; the operation is idempotent and stale or
-   inactive revisions must be skipped.
-5. **Reconciliation:** compare the preflight inventory with serving manifests,
-   checksums, per-revision unit counts, namespace aggregates, and generation
-   values. Confirm aggregates contain only active documents still belonging to
-   the namespace. Investigate every missing, extra, stale, or invalid revision.
-6. **Acceptance:** run the production read-only legacy-versus-serving
-   differential harness and record latency, selected IDs, order, scores,
-   citations, section paths, asset references, and fallback behavior. Declare
-   the rollout complete only after zero semantic mismatches.
-
-If migration or backfill must be stopped, leave the serving tables in place and
-stop the operator job. The reader will continue using the exact legacy path for
-incomplete revisions. Roll back application code first if necessary; do not
-drop serving tables or indexes as an emergency rollback action. Repair a failed
-revision by rerunning the bounded backfill after the cause is understood.
-
-## 7. Contract tests and benchmarks
-
-Use contract tests only. Add contracts for:
-
-- manifest round-trip, checksum, version, and revision pinning;
-- eager versus serving snapshot equivalence;
-- exclusions, duplicate chunk IDs, document-scoped references, and root assets;
-- exact Latin/CJK, empty, no-hit, phrase, token-only, and negative-IDF cases;
-- incomplete serving data falling back to legacy;
-- publication replacement, archive deltas, concurrent generation changes, and
-  stale backfill protection;
-- cache invalidation racing with a generation change, proving an old cached
-  response is not returned for a newer serving generation;
-- map-nav republish during the LLM episode, proving final hydration and
-  connected-asset resolution stay on the captured revisions;
-- classic publication replacement during concurrent channels, ranking, and
-  final assembly, proving every returned row shares the channel's pin set;
-- archive/backfill races proving archived or namespace-moved revisions never
-  contribute to serving statistics;
-- concurrent classic channels preserving IDs, order, scores, citations, and
-  fallback behavior.
-
-Race tests must use barriers or an equivalent deterministic hook to force a
-republish during the map-nav episode, a publication between classic channel
-sessions, an archive during backfill, and a namespace move during backfill.
-Each test must assert both the returned evidence and the persisted statistics,
-not merely that the request completed.
-
-The validation harness must also inspect the generated SQL/query plans (or an
-equivalent query-boundary assertion) to prove pinned reads do not use live
-`Document.current_job_result_id` joins. Run cache-version/generation races and
-verify an old cached response is rejected after a lifecycle change.
-
-Run a differential harness against the production read-only database and
-compare selected IDs, ordering, rounded scores, citations, section paths,
-asset references, and fallback behavior.
-
-Benchmark fresh processes and uncached queries. Report separately:
-
-- serving capture/decode;
-- map index projection and scoring;
-- episode-local projection reuse (number of full projection loads and per-wave
-  query-only scoring time);
-- classic discovery;
-- ranking;
-- hydration/assembly;
-- total Retrieval Non-LLM Work;
-- planner/harvest/control LLM time.
-
-The complexity check is explicit: one request may perform one full pinned
-snapshot/projection pass, relight work should scale with query postings rather
-than reloading the corpus, and hydration should scale with selected evidence
-(`top_k`/references), not namespace size. Navigation wave count must not
-multiply full-corpus database loads.
-
-Record peak resident memory for a fresh worker during the same benchmark and
-repeat it with the expected concurrent-request level. Memory is reported as an
-operational trade-off rather than a latency acceptance gate for this phase;
-before production rollout, any episode-local or process-local reuse still needs
-an explicit byte/item budget and an agreed worker ceiling.
-
-The fast path is accepted only after zero semantic mismatches and evidence that
-the cold request performs one bounded serving projection, does not repeat
-full-corpus loads per navigation wave, and meets an agreed latency budget for
-the current production-sized corpus.
-
-## 8. Main tradeoffs and risks
-
-- Publication becomes slower and uses more storage because derived data is built
-  synchronously.
-- Existing documents need an explicit backfill before they use the fast path.
-- A serving-index inconsistency causes a slower legacy request, not approximate
-  evidence.
-- PostgreSQL remains a scaling dependency; a future search-engine migration
-  would require a new semantic-parity review.
diff --git a/packages/shared-python/shared/models/database/document.py b/packages/shared-python/shared/models/database/document.py
index 739221c4..3d630b91 100644
--- a/packages/shared-python/shared/models/database/document.py
+++ b/packages/shared-python/shared/models/database/document.py
@@ -443,105 +443,6 @@ class RetrievalServingRevisionManifest(Base):
     )
 
 
-class RetrievalServingRevisionStat(Base):
-    """Compressed scoring contribution for one document revision."""
-
-    __tablename__ = "retrieval_serving_revision_stats"
-
-    id: Mapped[str] = mapped_column(
-        String(100), primary_key=True, default=lambda: f"rss_{uuid4().hex}"
-    )
-    user_id: Mapped[str] = mapped_column(Text, nullable=False)
-    namespace: Mapped[str] = mapped_column(String(255), nullable=False)
-    document_id: Mapped[str] = mapped_column(
-        String(36), ForeignKey("documents.document_id", ondelete="CASCADE"), nullable=False
-    )
-    job_result_id: Mapped[str] = mapped_column(
-        String(36), ForeignKey("job_results.id", ondelete="CASCADE"), nullable=False
-    )
-    format_version: Mapped[int] = mapped_column(Integer, nullable=False)
-    payload_zlib: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
-    checksum: Mapped[str] = mapped_column(String(64), nullable=False)
-    created_at: Mapped[datetime] = mapped_column(
-        DateTime, default=utc_now_naive, nullable=False
-    )
-
-    __table_args__ = (
-        UniqueConstraint(
-            "document_id",
-            "job_result_id",
-            name="uq_retrieval_serving_revision_stats_revision",
-        ),
-        Index(
-            "idx_retrieval_serving_revision_stats_scope",
-            "user_id",
-            "namespace",
-            "document_id",
-            "job_result_id",
-        ),
-    )
-
-
-class RetrievalNamespaceStat(Base):
-    """Compressed aggregate scoring statistics for one namespace generation."""
-
-    __tablename__ = "retrieval_namespace_stats"
-
-    id: Mapped[str] = mapped_column(
-        String(100), primary_key=True, default=lambda: f"rns_{uuid4().hex}"
-    )
-    user_id: Mapped[str] = mapped_column(Text, nullable=False)
-    namespace: Mapped[str] = mapped_column(String(255), nullable=False)
-    generation: Mapped[int] = mapped_column(BigInteger, nullable=False)
-    payload_zlib: Mapped[bytes] = mapped_column(LargeBinary, nullable=False)
-    checksum: Mapped[str] = mapped_column(String(64), nullable=False)
-    updated_at: Mapped[datetime] = mapped_column(
-        DateTime, default=utc_now_naive, onupdate=utc_now_naive, nullable=False
-    )
-
-    __table_args__ = (
-        UniqueConstraint(
-            "user_id",
-            "namespace",
-            name="uq_retrieval_namespace_stats_scope",
-        ),
-    )
-
-
-class RetrievalNamespaceTokenStat(Base):
-    """Document frequency for one token/channel in a namespace generation."""
-
-    __tablename__ = "retrieval_namespace_token_stats"
-
-    id: Mapped[str] = mapped_column(
-        String(100), primary_key=True, default=lambda: f"rnt_{uuid4().hex}"
-    )
-    user_id: Mapped[str] = mapped_column(Text, nullable=False)
-    namespace: Mapped[str] = mapped_column(String(255), nullable=False)
-    generation: Mapped[int] = mapped_column(BigInteger, nullable=False)
-    channel: Mapped[str] = mapped_column(String(32), nullable=False)
-    token_hash: Mapped[str] = mapped_column(String(64), nullable=False)
-    document_frequency: Mapped[int] = mapped_column(Integer, nullable=False)
-
-    __table_args__ = (
-        UniqueConstraint(
-            "user_id",
-            "namespace",
-            "channel",
-            "token_hash",
-            name="uq_retrieval_namespace_token_stats_key",
-        ),
-        Index(
-            "idx_retrieval_namespace_token_stats_lookup",
-            "user_id",
-            "namespace",
-            "generation",
-            "channel",
-            "token_hash",
-        ),
-    )
-
-
 class RetrievalNamespaceMapSnapshot(Base):
     """Persisted namespace-level MAP (sections + chunk index + map units).
 
diff --git a/packages/shared-python/shared/services/retrieval/namespace_map_snapshot.py b/packages/shared-python/shared/services/retrieval/namespace_map_snapshot.py
index bd8b7767..41fa6c6b 100644
--- a/packages/shared-python/shared/services/retrieval/namespace_map_snapshot.py
+++ b/packages/shared-python/shared/services/retrieval/namespace_map_snapshot.py
@@ -2,9 +2,8 @@
 
 Callers must already hold the namespace generation lock (see
 ``serving_generation.lock_namespace_generation``) before calling either
-function here, exactly as ``rebuild_namespace_serving_statistics`` requires.
-Each call only touches one document's subtree; every other document's
-subtree in the payload is left byte-for-byte unchanged.
+function here. Each call only touches one document's subtree; every other
+document's subtree in the payload is left byte-for-byte unchanged.
 """
 
 from __future__ import annotations
@@ -81,8 +80,7 @@ def remove_document_from_namespace_map_snapshot(
 def _target_generation(db: Session, *, user_id: str, namespace: str) -> int:
     """Namespace generation this snapshot is prepared for (current + 1).
 
-    Mirrors ``rebuild_namespace_serving_statistics``: callers advance the
-    generation after this write, in the same transaction.
+    Callers advance the generation after this write, in the same transaction.
     """
     generation = db.execute(
         select(RetrievalNamespaceGeneration)
diff --git a/packages/shared-python/shared/services/retrieval/publication_service.py b/packages/shared-python/shared/services/retrieval/publication_service.py
index 84b3b2d2..7ee5dec2 100644
--- a/packages/shared-python/shared/services/retrieval/publication_service.py
+++ b/packages/shared-python/shared/services/retrieval/publication_service.py
@@ -38,9 +38,6 @@
     advance_namespace_generation,
     lock_namespace_generation,
 )
-from shared.services.retrieval.serving_manifest import (
-    rebuild_namespace_serving_statistics,
-)
 
 
 def utc_now_naive() -> datetime:
@@ -183,17 +180,7 @@ def _publish_document_state_for_job(
         )
 
         db.flush()
-        rebuild_namespace_serving_statistics(
-            db,
-            user_id=scope.user_id,
-            namespace=scope.namespace,
-        )
         if existing_namespace and str(existing_namespace) != scope.namespace:
-            rebuild_namespace_serving_statistics(
-                db,
-                user_id=scope.user_id,
-                namespace=str(existing_namespace),
-            )
             remove_document_from_namespace_map_snapshot(
                 db,
                 user_id=scope.user_id,
diff --git a/packages/shared-python/shared/services/retrieval/serving_manifest.py b/packages/shared-python/shared/services/retrieval/serving_manifest.py
index 467d55c9..c66ae579 100644
--- a/packages/shared-python/shared/services/retrieval/serving_manifest.py
+++ b/packages/shared-python/shared/services/retrieval/serving_manifest.py
@@ -4,8 +4,6 @@
 
 import hashlib
 import json
-import logging
-import time
 import zlib
 from typing import Any
 
@@ -15,20 +13,13 @@
 from shared.models.database.document import (
     Document,
     DocumentChunk,
-    DocumentMapUnit,
-    DocumentMapUnitToken,
     DocumentSection,
-    RetrievalNamespaceGeneration,
-    RetrievalNamespaceStat,
-    RetrievalNamespaceTokenStat,
     RetrievalServingRevisionManifest,
-    RetrievalServingRevisionStat,
 )
 from shared.models.database.job_result import JobResult
 from shared.services.retrieval.publication_models import DocumentPublicationScope
 
 SERVING_MANIFEST_FORMAT_VERSION = 1
-_logger = logging.getLogger(__name__)
 
 
 def build_revision_serving_payload(
@@ -121,82 +112,25 @@ def build_revision_serving_payload(
     }
 
 
-def build_revision_statistics_payload(
-    db: Session,
-    *,
-    scope: DocumentPublicationScope,
-) -> dict[str, Any]:
-    """Build compressed scoring contributions for one revision.
-
-    Stores aggregate token frequencies for namespace statistics rebuild.
-    Per-unit frequencies stay in ``document_map_unit_tokens`` and are loaded
-    at query time by the query tokens only.
-    """
-    units = list(
-        db.scalars(
-            select(DocumentMapUnit)
-            .where(DocumentMapUnit.document_id == scope.document_id)
-            .where(DocumentMapUnit.job_result_id == scope.job_result_id)
-        )
-    )
-    unit_ids = [unit.id for unit in units]
-    frequencies: dict[str, dict[str, int]] = {"path": {}, "content": {}}
-    if unit_ids:
-        for _map_unit_id, channel, token, frequency in db.execute(
-            select(
-                DocumentMapUnitToken.map_unit_id,
-                DocumentMapUnitToken.channel,
-                DocumentMapUnitToken.token,
-                DocumentMapUnitToken.frequency,
-            ).where(DocumentMapUnitToken.map_unit_id.in_(unit_ids))
-        ).all():
-            channel_key = str(channel)
-            if channel_key not in frequencies:
-                continue
-            token_key = str(token)
-            frequencies[channel_key][token_key] = frequencies[channel_key].get(
-                token_key, 0
-            ) + int(frequency)
-    return {
-        "document_id": scope.document_id,
-        "job_result_id": scope.job_result_id,
-        "unit_count": len(units),
-        "path_token_count": sum(int(unit.path_token_count or 0) for unit in units),
-        "content_token_count": sum(
-            int(unit.content_token_count or 0) for unit in units
-        ),
-        "token_frequencies": frequencies,
-    }
-
-
 def persist_revision_serving_state(
     db: Session,
     *,
     scope: DocumentPublicationScope,
 ) -> dict[str, Any]:
-    """Replace manifest and statistics rows for one revision atomically.
+    """Replace the serving manifest row for one revision atomically.
 
     Returns the manifest payload so callers can patch the namespace-level MAP
     snapshot without rebuilding it.
     """
     manifest_payload = build_revision_serving_payload(db, scope=scope)
-    statistics_payload = build_revision_statistics_payload(db, scope=scope)
     manifest_bytes, manifest_checksum, manifest_version = encode_serving_manifest(
         manifest_payload
     )
-    statistics_bytes, statistics_checksum, statistics_version = encode_serving_manifest(
-        statistics_payload
-    )
     db.execute(
         delete(RetrievalServingRevisionManifest)
         .where(RetrievalServingRevisionManifest.document_id == scope.document_id)
         .where(RetrievalServingRevisionManifest.job_result_id == scope.job_result_id)
     )
-    db.execute(
-        delete(RetrievalServingRevisionStat)
-        .where(RetrievalServingRevisionStat.document_id == scope.document_id)
-        .where(RetrievalServingRevisionStat.job_result_id == scope.job_result_id)
-    )
     db.add(
         RetrievalServingRevisionManifest(
             user_id=scope.user_id,
@@ -208,144 +142,9 @@ def persist_revision_serving_state(
             checksum=manifest_checksum,
         )
     )
-    db.add(
-        RetrievalServingRevisionStat(
-            user_id=scope.user_id,
-            namespace=scope.namespace,
-            document_id=scope.document_id,
-            job_result_id=scope.job_result_id,
-            format_version=statistics_version,
-            payload_zlib=statistics_bytes,
-            checksum=statistics_checksum,
-        )
-    )
     return manifest_payload
 
 
-def rebuild_namespace_serving_statistics(
-    db: Session,
-    *,
-    user_id: str,
-    namespace: str,
-) -> int:
-    """Recompute namespace aggregates from active current revisions.
-
-    Callers hold the namespace generation lock. The aggregate is prepared for
-    the generation that the caller will publish next.
-    """
-    started = time.perf_counter()
-    generation = db.execute(
-        select(RetrievalNamespaceGeneration)
-        .where(RetrievalNamespaceGeneration.user_id == user_id)
-        .where(RetrievalNamespaceGeneration.namespace == namespace)
-        .with_for_update()
-    ).scalar_one()
-    target_generation = int(generation.generation) + 1
-    revisions = {
-        (str(document_id), str(job_result_id))
-        for document_id, job_result_id in db.execute(
-            select(Document.document_id, Document.current_job_result_id)
-            .where(Document.user_id == user_id)
-            .where(Document.namespace == namespace)
-            .where(Document.status == "active")
-            .where(Document.current_job_result_id.is_not(None))
-        ).all()
-        if document_id and job_result_id
-    }
-    aggregate: dict[str, Any] = {
-        "document_count": 0,
-        "unit_count": 0,
-        "path_token_count": 0,
-        "content_token_count": 0,
-        "token_frequencies": {"path": {}, "content": {}},
-    }
-    document_frequencies: dict[tuple[str, str], int] = {}
-    for row in db.scalars(
-        select(RetrievalServingRevisionStat)
-        .where(RetrievalServingRevisionStat.user_id == user_id)
-        .where(RetrievalServingRevisionStat.namespace == namespace)
-    ):
-        if (row.document_id, row.job_result_id) not in revisions:
-            continue
-        payload = decode_serving_manifest(
-            row.payload_zlib,
-            checksum=row.checksum,
-            format_version=row.format_version,
-        )
-        aggregate["document_count"] += 1
-        aggregate["unit_count"] += int(payload.get("unit_count", 0))
-        aggregate["path_token_count"] += int(payload.get("path_token_count", 0))
-        aggregate["content_token_count"] += int(payload.get("content_token_count", 0))
-        token_frequencies = payload.get("token_frequencies", {})
-        if not isinstance(token_frequencies, dict):
-            continue
-        for channel, values in token_frequencies.items():
-            if channel not in aggregate["token_frequencies"] or not isinstance(
-                values, dict
-            ):
-                continue
-            for token, value in values.items():
-                token_key = str(token)
-                aggregate["token_frequencies"][channel][token_key] = aggregate[
-                    "token_frequencies"
-                ][channel].get(token_key, 0) + int(value)
-                if int(value) > 0:
-                    key = (str(channel), token_key)
-                    document_frequencies[key] = document_frequencies.get(key, 0) + 1
-
-    encoded, checksum, _version = encode_serving_manifest(aggregate)
-    namespace_stat = db.execute(
-        select(RetrievalNamespaceStat)
-        .where(RetrievalNamespaceStat.user_id == user_id)
-        .where(RetrievalNamespaceStat.namespace == namespace)
-    ).scalar_one_or_none()
-    if namespace_stat is None:
-        db.add(
-            RetrievalNamespaceStat(
-                user_id=user_id,
-                namespace=namespace,
-                generation=target_generation,
-                payload_zlib=encoded,
-                checksum=checksum,
-            )
-        )
-    else:
-        namespace_stat.generation = target_generation
-        namespace_stat.payload_zlib = encoded
-        namespace_stat.checksum = checksum
-    db.execute(
-        delete(RetrievalNamespaceTokenStat)
-        .where(RetrievalNamespaceTokenStat.user_id == user_id)
-        .where(RetrievalNamespaceTokenStat.namespace == namespace)
-    )
-    db.add_all(
-        [
-            RetrievalNamespaceTokenStat(
-                user_id=user_id,
-                namespace=namespace,
-                generation=target_generation,
-                channel=channel,
-                token_hash=hashlib.sha256(token.encode("utf-8")).hexdigest(),
-                document_frequency=frequency,
-            )
-            for (channel, token), frequency in document_frequencies.items()
-        ]
-    )
-    db.flush()
-    _logger.info(
-        "retrieval namespace statistics rebuilt user_id=%s namespace=%s "
-        "generation=%d documents=%d units=%d token_stats=%d seconds=%.3f",
-        user_id,
-        namespace,
-        target_generation,
-        aggregate["document_count"],
-        aggregate["unit_count"],
-        len(document_frequencies),
-        time.perf_counter() - started,
-    )
-    return target_generation
-
-
 def _connection_target_ids(metadata: Any) -> list[str]:
     if not isinstance(metadata, dict):
         return []