Staging

CONTEXT.md

@@ -228,7 +228,10 @@ generation and retries or falls back if publication changes it during capture.
 
 The compatibility requirement that a serving-index retrieval returns the same
 selected chunk IDs, ordering, rounded scores, citations, and asset references
-as the legacy retrieval path for the same request.
+as the legacy retrieval path for the same request. Every retrieval optimization
+must preserve this quality contract; a latency improvement without validated
+semantic parity is not shippable. Validation also compares source sections,
+evidence content, router, and stop reason for each pinned request.
 
 ### Retrieval Revision Pin
 

apps/api/alembic/versions/a0b1c2d3e4f5_add_token_leading_map_unit_covering_index.py

@@ -0,0 +1,44 @@
+"""Add the token-leading covering index for map-unit lookups."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+
+revision: str = "a0b1c2d3e4f5"
+down_revision: str | None = "9f0a1b2c3d4e"
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+_INDEX_NAME = "idx_document_map_unit_tokens_token_lookup"
+
+
+def upgrade() -> None:
+    """Create the additive index without taking a table-wide write lock."""
+    uses_external_transaction: bool = bool(
+        op.get_context().opts.get("knowhere_external_transaction", False)
+    )
+    statement: str = (
+        f"CREATE INDEX {{concurrently}}IF NOT EXISTS {_INDEX_NAME} "
+        "ON document_map_unit_tokens (channel, token_hash, map_unit_id) "
+        "INCLUDE (token, frequency)"
+    )
+    if uses_external_transaction:
+        op.execute(statement.format(concurrently=""))
+        return
+    with op.get_context().autocommit_block():
+        op.execute(statement.format(concurrently="CONCURRENTLY "))
+
+
+def downgrade() -> None:
+    """Remove only the index introduced by this migration."""
+    uses_external_transaction: bool = bool(
+        op.get_context().opts.get("knowhere_external_transaction", False)
+    )
+    if uses_external_transaction:
+        op.execute(f"DROP INDEX IF EXISTS {_INDEX_NAME}")
+        return
+    with op.get_context().autocommit_block():
+        op.execute(f"DROP INDEX CONCURRENTLY IF EXISTS {_INDEX_NAME}")

apps/api/alembic/versions/b1c2d3e4f5a6_add_channel_bm25_statistics.py

@@ -0,0 +1,42 @@
+"""Add persisted per-channel BM25 corpus statistics."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+
+
+revision: str = "b1c2d3e4f5a6"
+down_revision: str | None = "a0b1c2d3e4f5"
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Add nullable statistics so incomplete legacy rows keep the fallback."""
+    column_names: tuple[str, ...] = (
+        "path_document_count",
+        "path_total_length",
+        "content_document_count",
+        "content_total_length",
+    )
+    for column_name in column_names:
+        op.execute(
+            f"ALTER TABLE document_map_unit_indexes "
+            f"ADD COLUMN IF NOT EXISTS {column_name} INTEGER"
+        )
+
+
+def downgrade() -> None:
+    """Remove the additive statistics columns."""
+    column_names: tuple[str, ...] = (
+        "content_total_length",
+        "content_document_count",
+        "path_total_length",
+        "path_document_count",
+    )
+    for column_name in column_names:
+        op.execute(
+            f"ALTER TABLE document_map_unit_indexes DROP COLUMN IF EXISTS {column_name}"
+        )

apps/api/alembic/versions/c2d3e4f5a6b7_repair_token_leading_map_unit_covering_index.py

@@ -0,0 +1,62 @@
+"""Repair a missing or invalid token-leading map-unit covering index."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+from sqlalchemy import text
+
+
+revision: str = "c2d3e4f5a6b7"
+down_revision: str | None = "b1c2d3e4f5a6"
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+_INDEX_NAME = "idx_document_map_unit_tokens_token_lookup"
+_INDEX_COLUMNS = "(channel, token_hash, map_unit_id) INCLUDE (token, frequency)"
+
+
+def _is_index_ready() -> bool:
+    """Return whether the current schema contains a usable covering index."""
+    is_ready = op.get_bind().execute(
+        text(
+            "SELECT indexes.indisvalid AND indexes.indisready "
+            "FROM pg_index AS indexes "
+            "JOIN pg_class AS classes ON classes.oid = indexes.indexrelid "
+            "JOIN pg_namespace AS namespaces "
+            "ON namespaces.oid = classes.relnamespace "
+            "WHERE namespaces.nspname = current_schema() "
+            "AND classes.relname = :index_name"
+        ),
+        {"index_name": _INDEX_NAME},
+    ).scalar_one_or_none()
+    return bool(is_ready)
+
+
+def _repair_index(*, concurrently: bool) -> None:
+    """Replace a missing or unusable index using the allowed DDL mode."""
+    if _is_index_ready():
+        return
+    concurrent_clause: str = "CONCURRENTLY " if concurrently else ""
+    op.execute(f"DROP INDEX {concurrent_clause}IF EXISTS {_INDEX_NAME}")
+    op.execute(
+        f"CREATE INDEX {concurrent_clause}{_INDEX_NAME} "
+        f"ON document_map_unit_tokens {_INDEX_COLUMNS}"
+    )
+
+
+def upgrade() -> None:
+    """Ensure the additive covering index exists and is usable."""
+    uses_external_transaction: bool = bool(
+        op.get_context().opts.get("knowhere_external_transaction", False)
+    )
+    if uses_external_transaction:
+        _repair_index(concurrently=False)
+        return
+    with op.get_context().autocommit_block():
+        _repair_index(concurrently=True)
+
+
+def downgrade() -> None:
+    """Keep the index owned by the preceding additive migration."""

apps/api/alembic/versions/d3e4f5a6b7c8_merge_retrieval_index_heads.py

@@ -0,0 +1,21 @@
+"""Merge the retrieval index migration branches."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+revision: str = "d3e4f5a6b7c8"
+down_revision: tuple[str, str] = (
+    "0a1b2c3d4e5f",
+    "c2d3e4f5a6b7",
+)
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    """Merge migration heads without applying additional schema changes."""
+
+
+def downgrade() -> None:
+    """Split the migration graph back into its two parent heads."""

apps/api/app/services/document_ingestion/creation_service.py

@@ -5,7 +5,7 @@
 from dataclasses import dataclass
 from datetime import datetime, timezone
 from typing import cast
-from urllib.parse import urlparse
+from urllib.parse import unquote, urlparse
 
 from app.repositories.job_repository import JobRepository
 from app.services.document_ingestion.command import DocumentIngestionCommand
@@ -308,7 +308,7 @@ def _build_job_response(
 
 def _resolve_url_source_file_name(*, source_url: str, file_extension: str) -> str:
     parsed_url = urlparse(source_url)
-    url_basename = str(os.path.basename(parsed_url.path))
+    url_basename = unquote(str(os.path.basename(parsed_url.path)))
     if url_basename and os.path.splitext(url_basename)[1].lower() == file_extension:
         return url_basename
     if url_basename: