fix(kb): legacy pipeline stands down for a promoted knowledge base

backend/Dockerfile.kb-sync

@@ -37,6 +37,12 @@ COPY backend/src/apis/shared/dynamo_errors.py ${LAMBDA_TASK_ROOT}/apis/shared/dy
 COPY backend/src/apis/shared/sync_policies/ ${LAMBDA_TASK_ROOT}/apis/shared/sync_policies/
 COPY backend/src/apis/shared/oauth/ ${LAMBDA_TASK_ROOT}/apis/shared/oauth/
 COPY backend/src/apis/shared/embeddings/ ${LAMBDA_TASK_ROOT}/apis/shared/embeddings/
+# kb_backend/ — the worker reaches documents/services/cleanup_service.py,
+# which reads records.resolve_engine and calls ManagedKbBackend to remove a
+# deleted document from a promoted knowledge base. Module-level imports in
+# that package are stdlib only (test_kb_backend_boundary.py), so this adds
+# files, not dependencies.
+COPY backend/src/apis/shared/kb_backend/ ${LAMBDA_TASK_ROOT}/apis/shared/kb_backend/
 # assistants/ — document_service verifies assistant ownership via
 # get_assistant before soft-deleting; the worker's miss-eviction path
 # (_delete_missing_web_document) goes through it.

backend/Dockerfile.rag-ingestion

@@ -168,9 +168,15 @@ COPY backend/src/apis/app_api/documents/ingestion/ ${LAMBDA_TASK_ROOT}
 # tag notices changes.
 #   embeddings/   — ingestion/embeddings/bedrock_embeddings.py re-exports it
 #   timestamps.py — ingestion/status.py imports utc_now_iso
+#   kb_backend/   — handler.py reads `records.resolve_engine` to skip documents
+#                   whose knowledge base has been promoted to the managed engine.
+#                   Cheap to carry: the package's module-level imports are stdlib
+#                   only (enforced by tests/architecture/test_kb_backend_boundary.py),
+#                   so this adds files, not dependencies.
 COPY backend/src/apis/shared/__init__.py ${LAMBDA_TASK_ROOT}/apis/shared/__init__.py
 COPY backend/src/apis/shared/timestamps.py ${LAMBDA_TASK_ROOT}/apis/shared/timestamps.py
 COPY backend/src/apis/shared/embeddings/ ${LAMBDA_TASK_ROOT}/apis/shared/embeddings/
+COPY backend/src/apis/shared/kb_backend/ ${LAMBDA_TASK_ROOT}/apis/shared/kb_backend/
 RUN touch ${LAMBDA_TASK_ROOT}/apis/__init__.py
 
 CMD [ "handler.lambda_handler" ]

backend/src/apis/app_api/documents/ingestion/handler.py

@@ -106,12 +106,53 @@ def _detect_mime_type(content_type: Optional[str], filename: str) -> str:
         pass  # dotenv not installed, running in Lambda
 
 
+def _resolve_engine(assistant_id: str) -> str:
+    """Which engine serves this knowledge base: ``managed`` or ``s3vectors``.
+
+    Delegates to ``records.resolve_engine`` rather than reading the attribute
+    here, so "absence means legacy" has exactly one definition in the codebase —
+    the same call the managed ingestion consumer makes for the mirror-image
+    decision.
+
+    **Failures resolve to legacy, deliberately.** The two ways to be wrong are
+    not symmetric:
+
+    * Wrong towards legacy on a promoted knowledge base: the document is indexed
+      twice and the two writers race on status. Wasteful and untidy, but the
+      document still ends up correct, because the managed consumer is also
+      running and it owns the terminal state.
+    * Wrong towards skipping on a legacy knowledge base: nothing indexes the
+      document at all. It sits un-ingested with no error, and the only path back
+      is a re-upload.
+
+    The second is much worse, so an unreadable record runs the legacy pipeline.
+    This is the same convention as ``resolver.load_record``, which treats a
+    failed read and an absent record identically on the grounds that "an absent
+    opinion is the legacy opinion".
+    """
+    from apis.shared.kb_backend.records import ENGINE_LEGACY, get_kb_record, resolve_engine
+
+    try:
+        return resolve_engine(get_kb_record(assistant_id, assistant_id))
+    except Exception as exc:  # noqa: BLE001 — see the docstring: legacy is the safe default
+        logger.warning(
+            f"could not resolve the retrieval engine for assistant {assistant_id}; "
+            f"running the legacy pipeline, which is the safe default: {exc}"
+        )
+        return ENGINE_LEGACY
+
+
 async def async_lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str, Any]:
     """
     Async implementation of the Lambda handler
     """
     from status import create_status_manager
 
+    # Function-local like every other import in this module: the Lambda image
+    # keeps its cold start down by not importing anything at module scope that a
+    # given invocation might not need.
+    from apis.shared.kb_backend.records import ENGINE_MANAGED
+
     # Initialize status manager
     status_manager = create_status_manager()
 
@@ -121,6 +162,49 @@ async def async_lambda_handler(event: Dict[str, Any], context: Any) -> Dict[str,
         # 1. Parse event and extract metadata
         event_data = _parse_s3_event(event)
 
+        # 1a. Routing exclusivity (design §537, Requirement 10.5). A promoted
+        # knowledge base is served by the managed backend, and the managed
+        # ingestion consumer is triggered by the same S3 upload through
+        # EventBridge. So for those documents this pipeline must do NOTHING:
+        # not parse, not embed, and above all not write document status.
+        #
+        # Both halves matter, and the status half is the one that bit us:
+        #
+        #   * Indexing anyway spends Docling time and S3 Vectors storage on
+        #     vectors that retrieval will never read, because `resolve_backend`
+        #     sends a promoted record to the managed adapter with no fallback.
+        #   * Writing status anyway means two writers own one field and the last
+        #     one wins by luck. Measured in dev: a PDF was marked `complete` by
+        #     this pipeline 65 s before the managed KB could answer for it, and
+        #     an image-only PDF that Docling could not parse at all was marked
+        #     `failed` while the managed KB was serving it perfectly. Reverse the
+        #     finishing order and a good document reads `failed` forever.
+        #
+        # Deliberately keyed on the *engine*, not on "has a KB record" or on
+        # `migrationState`: during `shadow` and `verify` the legacy path is still
+        # authoritative and must keep working (Requirements 16.1, 16.6). Only
+        # `promote` writes `retrievalEngine`, so only a promoted knowledge base
+        # is skipped here.
+        engine = _resolve_engine(event_data["assistant_id"])
+        if engine == ENGINE_MANAGED:
+            logger.info(
+                f"skipping document {event_data['document_id']} for assistant "
+                f"{event_data['assistant_id']}: it is served by the {engine!r} "
+                f"engine, so the managed ingestion consumer owns this document "
+                f"and its status"
+            )
+            return {
+                "statusCode": 200,
+                "body": json.dumps(
+                    {
+                        "message": "Skipped: knowledge base is served by the managed engine",
+                        "assistant_id": event_data["assistant_id"],
+                        "document_id": event_data["document_id"],
+                        "engine": engine,
+                    }
+                ),
+            }
+
         # 2. Update status to 'chunking'
         logger.info("Starting document processing")
         await status_manager.mark_chunking(assistant_id=event_data["assistant_id"], document_id=event_data["document_id"])

backend/src/apis/app_api/documents/services/cleanup_service.py

@@ -78,6 +78,16 @@ async def cleanup_document_resources(
         logger.error(f"Unexpected error in vector deletion for {document_id}: {e}", exc_info=True)
         vectors_deleted = False
 
+    try:
+        managed_deleted = await _delete_managed_documents_with_retries(
+            assistant_id, document_id, max_retries, base_delay
+        )
+    except Exception as e:
+        logger.error(
+            f"Unexpected error in managed-KB deletion for {document_id}: {e}", exc_info=True
+        )
+        managed_deleted = False
+
     try:
         s3_deleted = await _delete_s3_with_retries(
             s3_key, max_retries, base_delay
@@ -86,7 +96,7 @@ async def cleanup_document_resources(
         logger.error(f"Unexpected error in S3 deletion for {document_id}: {e}", exc_info=True)
         s3_deleted = False
 
-    all_succeeded = vectors_deleted and s3_deleted
+    all_succeeded = vectors_deleted and managed_deleted and s3_deleted
 
     if all_succeeded:
         try:
@@ -239,6 +249,98 @@ async def _delete_vectors_with_retries(
     return False
 
 
+async def _delete_managed_documents_with_retries(
+    assistant_id: str,
+    document_id: str,
+    max_retries: int,
+    base_delay: float,
+) -> bool:
+    """Remove the document from the managed knowledge base, if there is one.
+
+    The mirror of the legacy vector deletion above, and the reason it exists:
+    before this, deletion removed the S3 Vectors copy and the ``DOC#`` row but
+    never touched the managed knowledge base. On a promoted knowledge base the
+    content therefore stayed indexed forever. The corpus could only grow, at
+    $5.00/GB-month, and each orphan kept consuming a slot in every ``top_k``
+    before the status filter dropped it — so answers quietly got thinner while
+    nothing errored.
+
+    Returns ``True`` when there is nothing to do
+    -------------------------------------------
+    A knowledge base that is not promoted has no managed copy, so "no managed
+    copy to delete" is success, not failure. Returning ``False`` there would
+    block ``hard_delete_document`` for every legacy document in the system.
+
+    Why a failure here must block the hard delete
+    ---------------------------------------------
+    The caller only hard-deletes the ``DOC#`` row when every phase succeeds, and
+    that row is what the fail-closed status filter joins against. Leaving it in
+    place is what keeps a still-indexed managed chunk from being served while
+    this deletion is retried. Reporting success on a failed managed delete would
+    remove the row *and* leave the content — the one combination that turns a
+    storage leak into a disclosure.
+
+    Deletion is idempotent
+    ----------------------
+    ``DeleteKnowledgeBaseDocuments`` on an absent document is not an error, so a
+    retry after a partial failure is safe and needs no bookkeeping.
+    """
+    from apis.shared.kb_backend.records import ENGINE_MANAGED, get_kb_record, resolve_engine
+
+    try:
+        # App_KB_Id == assistant_id in this phase, so one value serves both.
+        engine = resolve_engine(get_kb_record(assistant_id, assistant_id))
+    except Exception as e:
+        # Unlike the ingestion gate — where an unreadable record resolves to
+        # legacy so the upload still gets indexed — an unreadable record HERE
+        # must fail. If this knowledge base is in fact promoted, reporting
+        # success would hard-delete the row that keeps its chunks unserved.
+        logger.error(
+            f"could not resolve the retrieval engine for assistant {assistant_id} "
+            f"while deleting {document_id}; treating the managed deletion as failed "
+            f"so the document record survives for a retry: {e}"
+        )
+        return False
+
+    if engine != ENGINE_MANAGED:
+        return True
+
+    from apis.shared.kb_backend.managed_backend import ManagedKbBackend, ManagedKbNotProvisioned
+
+    backend = ManagedKbBackend()
+    for attempt in range(max_retries):
+        try:
+            await backend.delete_document(assistant_id, document_id)
+            logger.info(
+                f"deleted document {document_id} from the managed knowledge base "
+                f"for assistant {assistant_id}"
+            )
+            return True
+        except ManagedKbNotProvisioned:
+            # Promoted but carrying no AWS identifiers is not a state a real
+            # migration produces — `promote` runs after provisioning. Nothing was
+            # ever indexed, so there is nothing to remove and no reason to retry.
+            logger.warning(
+                f"assistant {assistant_id} names the managed engine but has no AWS "
+                f"identifiers; nothing to delete for {document_id}"
+            )
+            return True
+        except Exception as e:
+            delay = base_delay * (2 ** attempt) + random.uniform(0, 0.1)
+            logger.warning(
+                f"Managed-KB deletion attempt {attempt + 1}/{max_retries} failed for "
+                f"{document_id}: {e}, retrying in {delay:.2f}s"
+            )
+            if attempt < max_retries - 1:
+                await asyncio.sleep(delay)
+
+    logger.error(
+        f"Managed-KB deletion failed after {max_retries} attempts for {document_id}; "
+        f"the document record is being kept so the status filter continues to hide it"
+    )
+    return False
+
+
 async def _delete_s3_with_retries(
     s3_key: str,
     max_retries: int,