feat: unify BM25 scoring with persistent namespace MAP

apps/api/alembic/versions/6c7d8e9f0a1b_add_retrieval_namespace_map_snapshots.py

@@ -0,0 +1,39 @@
+"""Add persisted namespace-level MAP snapshot table."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+
+from alembic import op
+import sqlalchemy as sa
+
+
+revision: str = "6c7d8e9f0a1b"
+down_revision: str | None = "5b6c7d8e9f0a"
+branch_labels: Sequence[str] | None = None
+depends_on: Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    if not sa.inspect(op.get_bind()).has_table("retrieval_namespace_map_snapshots"):
+        op.create_table(
+            "retrieval_namespace_map_snapshots",
+            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("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("updated_at", sa.DateTime(), nullable=False),
+            sa.PrimaryKeyConstraint("id"),
+            sa.UniqueConstraint(
+                "user_id",
+                "namespace",
+                name="uq_retrieval_namespace_map_snapshots_scope",
+            ),
+        )
+
+
+def downgrade() -> None:
+    op.drop_table("retrieval_namespace_map_snapshots", if_exists=True)

apps/api/alembic/versions/7d8e9f0a1b2c_add_map_unit_asset_presence.py

@@ -0,0 +1,63 @@
+"""Add has_image/has_table presence flags to document_map_units."""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "7d8e9f0a1b2c"
+down_revision = "6c7d8e9f0a1b"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    columns = {col["name"] for col in inspector.get_columns("document_map_units")}
+    if "has_image" not in columns:
+        op.add_column(
+            "document_map_units",
+            sa.Column(
+                "has_image", sa.Boolean(), nullable=False, server_default=sa.false()
+            ),
+        )
+    if "has_table" not in columns:
+        op.add_column(
+            "document_map_units",
+            sa.Column(
+                "has_table", sa.Boolean(), nullable=False, server_default=sa.false()
+            ),
+        )
+
+    inspector = sa.inspect(bind)
+    indexes = {item["name"] for item in inspector.get_indexes("document_map_units")}
+    if "idx_document_map_units_has_image" not in indexes:
+        op.create_index(
+            "idx_document_map_units_has_image",
+            "document_map_units",
+            ["document_id", "job_result_id"],
+            postgresql_where=sa.text("has_image = true"),
+        )
+    if "idx_document_map_units_has_table" not in indexes:
+        op.create_index(
+            "idx_document_map_units_has_table",
+            "document_map_units",
+            ["document_id", "job_result_id"],
+            postgresql_where=sa.text("has_table = true"),
+        )
+
+
+def downgrade() -> None:
+    inspector = sa.inspect(op.get_bind())
+    indexes = {item["name"] for item in inspector.get_indexes("document_map_units")}
+    if "idx_document_map_units_has_table" in indexes:
+        op.drop_index("idx_document_map_units_has_table", table_name="document_map_units")
+    if "idx_document_map_units_has_image" in indexes:
+        op.drop_index("idx_document_map_units_has_image", table_name="document_map_units")
+
+    columns = {col["name"] for col in inspector.get_columns("document_map_units")}
+    if "has_table" in columns:
+        op.drop_column("document_map_units", "has_table")
+    if "has_image" in columns:
+        op.drop_column("document_map_units", "has_image")

apps/api/alembic/versions/8e9f0a1b2c3d_add_map_unit_index_average_idf.py

@@ -0,0 +1,55 @@
+"""Add average_idf columns to document_map_unit_indexes.
+
+Stores the rank_bm25 Okapi average IDF per channel at index-write time so
+query scoring never scans all tokens to rebuild it.
+"""
+
+from __future__ import annotations
+
+import sqlalchemy as sa
+from alembic import op
+
+revision = "8e9f0a1b2c3d"
+down_revision = "7d8e9f0a1b2c"
+branch_labels = None
+depends_on = None
+
+
+def upgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    columns = {
+        col["name"] for col in inspector.get_columns("document_map_unit_indexes")
+    }
+    if "average_idf_path" not in columns:
+        op.add_column(
+            "document_map_unit_indexes",
+            sa.Column(
+                "average_idf_path",
+                sa.Float(),
+                nullable=False,
+                server_default="0",
+            ),
+        )
+    if "average_idf_content" not in columns:
+        op.add_column(
+            "document_map_unit_indexes",
+            sa.Column(
+                "average_idf_content",
+                sa.Float(),
+                nullable=False,
+                server_default="0",
+            ),
+        )
+
+
+def downgrade() -> None:
+    bind = op.get_bind()
+    inspector = sa.inspect(bind)
+    columns = {
+        col["name"] for col in inspector.get_columns("document_map_unit_indexes")
+    }
+    if "average_idf_content" in columns:
+        op.drop_column("document_map_unit_indexes", "average_idf_content")
+    if "average_idf_path" in columns:
+        op.drop_column("document_map_unit_indexes", "average_idf_path")

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

@@ -77,6 +77,14 @@ class RetrievalQueryRequest(BaseModel):
             "Set false to force classic 3-channel top-K retrieval."
         ),
     )
+    conversation_id: str | None = Field(
+        None,
+        max_length=255,
+        description=(
+            "Caller-supplied conversation identifier, threaded through for "
+            "retrieval tracing. Does not affect caching or result content."
+        ),
+    )
 
     @field_validator("channels")
     @classmethod
@@ -172,6 +180,7 @@ async def execute_retrieval_query(
         threshold=payload.threshold,
         internal_recall_k=payload.internal_recall_k,
         use_agentic=payload.use_agentic,
+        conversation_id=payload.conversation_id,
         llm_config=llm_config,
     )
 

apps/api/app/services/documents/lifecycle_service.py

@@ -20,6 +20,9 @@
     invalidate_retrieval_cache_namespaces,
 )
 from shared.services.retrieval.graph.service import DocumentGraphService, GraphScope
+from shared.services.retrieval.namespace_map_snapshot import (
+    remove_document_from_namespace_map_snapshot,
+)
 from shared.services.retrieval.serving_generation import (
     advance_namespace_generation,
     lock_namespace_generation,
@@ -483,6 +486,14 @@ async def archive_document(
                 namespace=previous_namespace,
             )
         )
+        await db.run_sync(
+            lambda sync_db: remove_document_from_namespace_map_snapshot(
+                sync_db,
+                user_id=user_id,
+                namespace=previous_namespace,
+                document_id=document_id,
+            )
+        )
         await db.run_sync(
             lambda sync_db: advance_namespace_generation(
                 sync_db,