feat(rag): add a real reranker to RAG retrieval

backend/alembic/versions/0056_knowledge_base_rerank.py

@@ -0,0 +1,61 @@
+"""Which reranker - and whose key - a collection reranks search results with.
+
+Retrieval fetches candidates by vector similarity and returns the top ones. A
+reranker is a second pass: a model scores each candidate against the query
+directly and reorders them, which is more accurate than the distance the vector
+index sorts by. It is optional and off by default.
+
+Two nullable columns, mirroring the embedding pair. `rerank_model` is the
+reranker's name; `rerank_secret_id` is the organization vault key that pays for
+it. Reranking runs only when *both* are set - either NULL leaves retrieval
+exactly as it was, so existing rows and unconfigured deployments are unchanged
+by this migration. SET NULL on delete for the same reason the embedding key is:
+losing the key drops reranking, it does not take document search down. Unlike
+the embedding key there is no deployment fallback - a reranker with no key is
+simply off.
+
+Revision ID: 0056_knowledge_base_rerank
+Revises: 0055_sandbox_operations
+Create Date: 2026-08-18
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0056_knowledge_base_rerank"
+down_revision: str | None = "0055_sandbox_operations"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "knowledge_bases",
+        sa.Column("rerank_model", sa.String(length=128), nullable=True),
+    )
+    op.add_column(
+        "knowledge_bases",
+        sa.Column("rerank_secret_id", sa.UUID(), nullable=True),
+    )
+    op.create_foreign_key(
+        op.f("knowledge_bases_rerank_secret_id_fkey"),
+        "knowledge_bases",
+        "organization_secrets",
+        ["rerank_secret_id"],
+        ["id"],
+        ondelete="SET NULL",
+    )
+
+
+def downgrade() -> None:
+    op.drop_constraint(
+        op.f("knowledge_bases_rerank_secret_id_fkey"),
+        "knowledge_bases",
+        type_="foreignkey",
+    )
+    op.drop_column("knowledge_bases", "rerank_secret_id")
+    op.drop_column("knowledge_bases", "rerank_model")

backend/alembic/versions/0057_ingestion_spend_source.py

@@ -0,0 +1,43 @@
+"""Tag each non-run RAG spend row as indexing or retrieval.
+
+`ingestion_spend` began as indexing alone, then a metered `POST /rag/search`
+landed its embedding and rerank cost in the same table - both are RAG spend
+outside any agent run. Left undistinguished, a search inflated the dashboard's
+"indexing" subtotal. `source` tells them apart; both still count toward the
+monthly budget, only the reporting split reads the column.
+
+Every row that predates the column is indexing, so `server_default` backfills
+them to `'ingestion'` without a data migration.
+
+Revision ID: 0057_ingestion_spend_source
+Revises: 0056_knowledge_base_rerank
+Create Date: 2026-08-20
+
+"""
+
+from collections.abc import Sequence
+
+import sqlalchemy as sa
+
+from alembic import op
+
+revision: str = "0057_ingestion_spend_source"
+down_revision: str | None = "0056_knowledge_base_rerank"
+branch_labels: str | Sequence[str] | None = None
+depends_on: str | Sequence[str] | None = None
+
+
+def upgrade() -> None:
+    op.add_column(
+        "ingestion_spend",
+        sa.Column(
+            "source",
+            sa.String(length=16),
+            nullable=False,
+            server_default="ingestion",
+        ),
+    )
+
+
+def downgrade() -> None:
+    op.drop_column("ingestion_spend", "source")

backend/app/agents/capabilities/budget/__init__.py

@@ -9,6 +9,7 @@
     SpendLedger,
     SpendLimit,
     SpendShare,
+    book_ambient_spend,
     booked_to,
     metered_by,
     price_request,
@@ -26,6 +27,7 @@
     "SpendLedger",
     "SpendLimit",
     "SpendShare",
+    "book_ambient_spend",
     "booked_to",
     "metered_by",
     "price_request",

backend/app/agents/capabilities/budget/_capability.py

@@ -434,6 +434,24 @@ def record_ambient_usage(
         ledger.record(model_name, usage, provider)
 
 
+def book_ambient_spend(entry: SpendEntry) -> None:
+    """Book one already-priced entry to whichever ledger is active, if any is.
+
+    The sibling of :func:`record_ambient_usage` for spend that is not
+    token-priced. `record` prices through `genai-prices`, which knows chat and
+    embedding models and nothing else, so a reranker call routed through it
+    would book `cost_usd=0, priced=False`. A reranker computes its own cost from
+    a published per-search price and hands the finished :class:`SpendEntry` here
+    instead, so the entry lands `priced=True` with a real number.
+
+    A no-op when nothing is metering, for the same reason the sibling is: a
+    provider should not refuse to work because no ledger is open.
+    """
+    ledger = _active_ledger.get()
+    if ledger is not None:
+        ledger.book(entry)
+
+
 def usage_counts(usage: RunUsage) -> tuple[int, int, int, int]:
     """The four counters a price is computed from, read off the run's usage.
 

backend/app/agents/capabilities/knowledge/_search.py

@@ -3,11 +3,13 @@
 import contextvars
 import logging
 from typing import TYPE_CHECKING, Any
+from uuid import UUID
 
 from app.core.config import settings
 from app.core.exceptions import AppException, ExternalServiceError
 from app.services.embedding_resolution import embeddings_for_collection
 from app.services.rag.embeddings import EmbeddingService
+from app.services.rag.reranker import build_reranker
 from app.services.rag.retrieval import RetrievalService
 from app.services.rag.vectorstore import PgVectorStore
 
@@ -32,7 +34,13 @@ def get_retrieval_service() -> "BaseRetrievalService":
         rag_settings, embedding_service, resolver=embeddings_for_collection
     )
     _vector_store = vector_store
-    _retrieval_service = RetrievalService(vector_store, rag_settings)
+    # The reranker resolver is wired here too, not only on the /rag/search
+    # route: an agent's knowledge search reranks when its collection is
+    # configured, and the run's open ledger books the cost - which is the
+    # agent-run half of "spend recorded on both paths".
+    _retrieval_service = RetrievalService(
+        vector_store, rag_settings, reranker_resolver=build_reranker
+    )
     return _retrieval_service
 
 
@@ -93,6 +101,8 @@ async def search_knowledge_base(
     query: str,
     kb_collection_names: list[str] | None = None,
     top_k: int = 5,
+    *,
+    organization_id: UUID | None,
 ) -> str:
     """Search the knowledge base and return formatted results.
 
@@ -102,6 +112,9 @@ async def search_knowledge_base(
             agent's spec. Never supplied by the LLM directly - injected via
             PydanticAI Deps or the _active_kb_collections ContextVar.
         top_k: Number of top results to retrieve (default: 5).
+        organization_id: The organization the run acts for, so a collection name
+            shared across tenants resolves this one's embedding and rerank config
+            rather than another's (#913).
     """
     resolved = kb_collection_names if kb_collection_names else (_active_kb_collections.get() or [])
     if not resolved:
@@ -111,10 +124,18 @@ async def search_knowledge_base(
     one_collection = len(resolved) == 1
     try:
         if one_collection:
-            results = await service.retrieve(query=query, collection_name=resolved[0], limit=top_k)
+            results = await service.retrieve(
+                query=query,
+                collection_name=resolved[0],
+                limit=top_k,
+                organization_id=organization_id,
+            )
         else:
             results = await service.retrieve_multi(
-                query=query, collection_names=resolved, limit=top_k
+                query=query,
+                collection_names=resolved,
+                limit=top_k,
+                organization_id=organization_id,
             )
     except AppException:
         # Already an account of what is wrong and what to do about it - an

backend/app/agents/capabilities/knowledge/_toolset.py

@@ -40,6 +40,9 @@ async def search_documents(
                 # model chooses *what* to search, never *where*.
                 kb_collection_names=ctx.deps.kb_collection_names,
                 top_k=top_k or default_top_k,
+                # The run's own organization, so a collection name shared with
+                # another tenant resolves this agent's config, not theirs (#913).
+                organization_id=ctx.deps.organization_id,
             )
         except Exception as exc:
             raise ModelRetry("Knowledge base temporarily unavailable, please try again.") from exc

backend/app/api/deps.py

@@ -913,6 +913,8 @@ async def verify_api_key(
 from app.services.embedding_resolution import embeddings_for_collection
 from app.services.rag.ingestion import IngestionService
 from app.services.rag.documents import DocumentProcessor
+from app.services.knowledge_search import KnowledgeSearchService
+from app.services.rag.reranker import build_reranker
 from app.services.rag.retrieval import RetrievalService
 from app.services.rag.vectorstore import PgVectorStore
 from app.services.rag.vectorstore import BaseVectorStore
@@ -958,13 +960,31 @@ async def get_vectorstore(
 
 
 def get_retrieval_service(vector_store: VectorStoreSvc) -> RetrievalService:
-    """Create RetrievalService instance."""
-    return RetrievalService(vector_store=vector_store, settings=settings.rag)
+    """Create RetrievalService instance.
+
+    The reranker resolver is `build_reranker`, the one composition point shared
+    with the agent-run knowledge tool, so both paths rerank the same way.
+    """
+    return RetrievalService(
+        vector_store=vector_store,
+        settings=settings.rag,
+        reranker_resolver=build_reranker,
+    )
 
 
 RetrievalSvc = Annotated[RetrievalService, Depends(get_retrieval_service)]
 
 
+def get_knowledge_search_service(
+    db: DBSession, retrieval: RetrievalSvc, access: CollectionAccessSvc
+) -> KnowledgeSearchService:
+    """Create KnowledgeSearchService instance."""
+    return KnowledgeSearchService(db, retrieval, access)
+
+
+KnowledgeSearchSvc = Annotated[KnowledgeSearchService, Depends(get_knowledge_search_service)]
+
+
 def get_document_processor() -> DocumentProcessor:
     """Create DocumentProcessor instance."""
     return DocumentProcessor(settings=settings.rag)
@@ -976,9 +996,14 @@ def get_document_processor() -> DocumentProcessor:
 def get_ingestion_service(
     processor: DocumentProcessorSvc,
     vector_store: VectorStoreSvc,
+    org: ActiveOrg,
 ) -> IngestionService:
-    """Create IngestionService instance."""
-    return IngestionService(processor=processor, vector_store=vector_store)
+    """Create IngestionService instance, scoped to the active organization.
+
+    The organization is what lets the store resolve this tenant's embedding key
+    on a collection name another tenant may share, rather than another's (#913).
+    """
+    return IngestionService(processor=processor, vector_store=vector_store, organization_id=org.id)
 
 
 IngestionSvc = Annotated[IngestionService, Depends(get_ingestion_service)]

backend/app/api/routes/v1/rag.py

@@ -48,9 +48,9 @@
     CurrentAppAdmin,
     IngestionSvc,
     KnowledgeBaseSvc,
+    KnowledgeSearchSvc,
     RAGDocumentSvc,
     RAGSyncSvc,
-    RetrievalSvc,
     SyncSourceSvc,
     VectorStoreSvc,
     require,
@@ -159,7 +159,7 @@ async def create_collection(
     is refused rather than quietly aliased onto their vector table.
     """
     await access.claim(ctx, name)
-    await vector_store.create_collection(name)
+    await vector_store.create_collection(name, organization_id=ctx.organization_id)
     await kb_svc.create_for_rag_collection(
         name, user_id=ctx.subject_id, organization_id=ctx.organization_id
     )
@@ -216,7 +216,9 @@ async def get_collection_info(
 ) -> Any:
     """Retrieve stats for a specific collection."""
     collection = await access.readable(ctx, name)
-    return await vector_store.get_collection_info(collection.collection_name)
+    return await vector_store.get_collection_info(
+        collection.collection_name, organization_id=collection.organization_id
+    )
 
 
 @router.get(
@@ -242,33 +244,17 @@ async def list_documents(
 )
 async def search_documents(
     request: RAGSearchRequest,
-    retrieval_service: RetrievalSvc,
-    access: CollectionAccessSvc,
+    service: KnowledgeSearchSvc,
     ctx: Auth,
 ) -> Any:
     """Search for relevant document chunks. Supports multi-collection search.
 
     Every collection named is resolved before the first vector is read, and one
     the caller cannot reach refuses the whole search rather than being dropped
-    from it - see `CollectionAccessService.readable_all`.
+    from it. The embedding and any rerank the search runs are metered against
+    the caller's organization - see `KnowledgeSearchService`.
     """
-    names = request.collection_names or [request.collection_name]
-    collections = [kb.collection_name for kb in await access.readable_all(ctx, names)]
-    if len(collections) > 1:
-        results = await retrieval_service.retrieve_multi(
-            query=request.query,
-            collection_names=collections,
-            limit=request.limit,
-            min_score=request.min_score,
-        )
-    else:
-        results = await retrieval_service.retrieve(
-            query=request.query,
-            collection_name=collections[0],
-            limit=request.limit,
-            min_score=request.min_score,
-            filter=request.filter or "",
-        )
+    results = await service.search(ctx, request)
     api_results = [RAGSearchResult(**hit.model_dump()) for hit in results]
     return RAGSearchResponse(results=api_results)
 

backend/app/commands/rag.py

@@ -73,7 +73,11 @@ def get_rag_services() -> tuple[
     )
     processor = DocumentProcessor(settings=settings)
     retrieval = RetrievalService(vector_store=vector_store, settings=settings)
-    ingestion = IngestionService(processor=processor, vector_store=vector_store)
+    # CLI admin context: no single tenant, so resolution falls back to the name.
+    # Uploads through the API are org-scoped.
+    ingestion = IngestionService(
+        processor=processor, vector_store=vector_store, organization_id=None
+    )
     return settings, vector_store, processor, retrieval, ingestion
 
 
@@ -93,7 +97,7 @@ async def list_collections_async(vector_store: BaseVectorStore) -> None:
 
     for name in collection_names:
         try:
-            info_obj = await vector_store.get_collection_info(name)
+            info_obj = await vector_store.get_collection_info(name, organization_id=None)
             click.echo(f"  {name}")
             click.echo(f"    Vectors: {info_obj.total_vectors:,}")
             click.echo(f"    Dimension: {info_obj.dim}")
@@ -344,6 +348,7 @@ async def search_async(
         query=query,
         collection_name=collection,
         limit=top_k,
+        organization_id=None,
     )
 
     if not results:
@@ -474,7 +479,7 @@ async def stats_async(settings: RAGSettings, vector_store: BaseVectorStore) -> N
         total_vectors = 0
         for name in collection_names:
             try:
-                info_obj = await vector_store.get_collection_info(name)
+                info_obj = await vector_store.get_collection_info(name, organization_id=None)
                 click.echo(f"  {name}:")
                 click.echo(f"    Vectors: {info_obj.total_vectors:,}")
                 total_vectors += info_obj.total_vectors