feat: admin-document ingest-failure tracking

app/api/routes_admin_documents.py

@@ -27,6 +27,7 @@
 from app.security.malware import MalwareDetected, MalwareScanUnavailable, scan_bytes
 from app.storage.object_store import get_object_store, object_key_for
 from app.workers.extractors import supported_extension
+from app.workers.ingest_failures import safe_failure_reason
 from app.workers.ingest_worker import ingest_document_task
 
 
@@ -137,6 +138,54 @@ async def get_document(ingest_id: str, who: Principal = Depends(_admin_only)):
     return _to_status(record)
 
 
+@router.delete("/{ingest_id}")
+async def delete_document(ingest_id: str, who: Principal = Depends(_admin_only)):
+    """Remove an uploaded document: its ingest record, the stored blob, and -
+    when no sibling ingest still represents the same ``document_id`` - its
+    indexed vector chunks. Tenant-scoped; returns 404 if the id is unknown.
+    """
+    repo = _repository()
+    record = repo.get(tenant_id=who.tenant_id, ingest_id=ingest_id)
+    if record is None:
+        raise HTTPException(status_code=404, detail="admin document ingest not found")
+
+    # Drop the record first so a concurrent list reflects the deletion promptly.
+    repo.delete(tenant_id=who.tenant_id, ingest_id=ingest_id)
+
+    # Best-effort blob cleanup. Each ingest owns its own object key, so this is
+    # always safe to remove regardless of siblings.
+    object_key = record.get("object_key")
+    if object_key:
+        try:
+            _object_store().delete(object_key)
+        except Exception as exc:  # noqa: BLE001 - never block delete on storage
+            logger.warning("object delete failed for ingest %s: %s", ingest_id, exc)
+
+    # Purge vector chunks only when no other ingest still represents this
+    # document_id (chunks are keyed by document_id, not ingest_id, so deleting
+    # them would otherwise strip a surviving duplicate's search results).
+    document_id = record.get("document_id")
+    siblings = any(
+        r.get("document_id") == document_id
+        for r in repo.list_records(tenant_id=who.tenant_id)
+    )
+    chunks_deleted = 0
+    if document_id and not siblings:
+        try:
+            chunks_deleted = await _delete_vector_chunks(who.tenant_id, document_id)
+        except Exception as exc:  # noqa: BLE001 - chunks may be re-cleaned later
+            logger.warning(
+                "vector chunk delete failed for document %s: %s", document_id, exc
+            )
+
+    _audit_admin_doc(
+        "admin_document_delete", who, ingest_id,
+        _metadata_from_record(record), record["filename"],
+        response={"ingest_id": ingest_id, "deleted": True, "chunks_deleted": chunks_deleted},
+    )
+    return {"ingest_id": ingest_id, "deleted": True, "chunks_deleted": chunks_deleted}
+
+
 # Statuses that can be re-dispatched. "extracting"/"embedding" are in-flight in
 # async mode (don't double-run); "done" is already indexed (re-upload to refresh).
 _REQUEUEABLE = {"queued", "failed"}
@@ -217,12 +266,15 @@ def _dispatch_ingest(repo, *, tenant_id: str, ingest_id: str) -> None:
     try:
         ingest_document_task.apply(args=(tenant_id, ingest_id), throw=False)
     except Exception as exc:  # noqa: BLE001 - inline execution itself failed
+        logger.warning(
+            "ingest_dispatch_failed for ingest %s: %s", ingest_id, exc,
+        )
         try:
             repo.update_status(
                 tenant_id=tenant_id,
                 ingest_id=ingest_id,
                 status="failed",
-                failure_reason=f"dispatch: {exc}",
+                failure_reason=safe_failure_reason("dispatch", exc),
             )
         except Exception:  # noqa: BLE001 - never mask the original failure
             pass
@@ -257,6 +309,31 @@ def _object_store():
     return get_object_store()
 
 
+async def _delete_vector_chunks(tenant_id: str, document_id: str) -> int:
+    """Delete a document's chunks from the pgvector store.
+
+    Opens a short-lived asyncpg pool (mirrors the ingest worker's wiring) and
+    closes it afterwards. Tests monkeypatch this hook to avoid needing Postgres.
+    When persistence is local-json (demo/dev), there is no vector DB to clean,
+    so this is a no-op.
+    """
+    settings = get_settings()
+    if settings.persistence_backend != "postgres":
+        return 0
+
+    import asyncpg
+
+    from app.rag.vectorstore import VectorStore
+
+    url = settings.database_url.replace("postgresql+asyncpg://", "postgresql://", 1)
+    pool = await asyncpg.create_pool(dsn=url, min_size=1, max_size=2)
+    try:
+        store = VectorStore(pool)
+        return await store.delete_document(tenant_id=tenant_id, document_id=document_id)
+    finally:
+        await pool.close()
+
+
 def _parse_metadata(raw: str) -> AdminDocumentMetadata:
     try:
         payload = json.loads(raw)

app/db/admin_document_repository.py

@@ -138,6 +138,19 @@ def get(self, *, tenant_id: str, ingest_id: str) -> dict[str, Any] | None:
                 return row
         return None
 
+    def delete(self, *, tenant_id: str, ingest_id: str) -> bool:
+        """Drop a single ingest record. Returns True if a row was removed."""
+        with self._lock:
+            rows = self._read_all_locked()
+            kept = [
+                r for r in rows
+                if not (r.get("tenant_id") == tenant_id and r.get("ingest_id") == ingest_id)
+            ]
+            if len(kept) == len(rows):
+                return False
+            self._write_all_locked(kept)
+            return True
+
     def list_records(self, *, tenant_id: str) -> list[dict[str, Any]]:
         rows = [_summary(r) for r in self._read_all() if r.get("tenant_id") == tenant_id]
         return sorted(rows, key=lambda r: r["created_utc"], reverse=True)
@@ -248,6 +261,15 @@ def get(self, *, tenant_id: str, ingest_id: str) -> dict[str, Any] | None:
             ).fetchone()
         return _serialize_admindoc(row) if row else None
 
+    def delete(self, *, tenant_id: str, ingest_id: str) -> bool:
+        with self._conn(tenant_id) as conn:
+            row = conn.execute(
+                "DELETE FROM admin_documents WHERE tenant_id = %s AND ingest_id = %s "
+                "RETURNING ingest_id",
+                (tenant_id, ingest_id),
+            ).fetchone()
+        return row is not None
+
     def list_records(self, *, tenant_id: str) -> list[dict[str, Any]]:
         with self._conn(tenant_id) as conn:
             rows = conn.execute(

app/rag/vectorstore.py

@@ -53,6 +53,29 @@ async def upsert(self, rows: list[dict[str, Any]]) -> int:
                 ])
         return len(rows)
 
+    async def delete_document(self, *, tenant_id: str, document_id: str) -> int:
+        """Remove every chunk for ``document_id`` within the tenant.
+
+        Returns the number of rows deleted. The tenant filter is mandatory and
+        first (mirrors :meth:`hybrid_search`) so a delete can never reach across
+        tenants even if the GUC/RLS backstop were misconfigured.
+        """
+        tenant_id = _require_tenant_id(tenant_id)
+        if not isinstance(document_id, str) or not document_id.strip():
+            raise ValueError("document_id is required to delete chunks")
+        async with self.pool.acquire() as con:
+            async with con.transaction():
+                await _set_tenant_context(con, tenant_id)
+                status = await con.execute(
+                    "DELETE FROM doc_chunks WHERE tenant_id = $1 AND document_id = $2",
+                    tenant_id, document_id,
+                )
+        # asyncpg returns a command tag like "DELETE 3"; parse the row count.
+        try:
+            return int(status.split()[-1])
+        except (AttributeError, ValueError, IndexError):
+            return 0
+
     async def hybrid_search(self, tenant_id: str, query_text: str,
                             query_embedding: list[float], top_k: int,
                             asset: str | None = None,

app/workers/ingest_failures.py

@@ -0,0 +1,54 @@
+"""User-safe failure messages for the document ingestion pipeline.
+
+Raw provider / parser exceptions routinely carry detail that must NOT reach a
+tenant's screen: an OpenAI ``429`` body advertises our billing page and admits
+the shared key is out of quota, vLLM/asyncpg errors leak internal hostnames,
+and pdf parsers leak temp file paths. Those belong in the server log only.
+
+``safe_failure_reason`` maps an exception to a short, stable, non-sensitive
+string (keeping the ``<stage>:`` prefix the UI and tests rely on) so the
+``failure_reason`` persisted on a document and shown in the admin table never
+exposes third-party error text. Callers should log the full exception
+separately for operators.
+"""
+from __future__ import annotations
+
+_QUOTA_OR_RATE = (
+    "429", "quota", "insufficient_quota", "rate limit", "rate_limit",
+    "too many requests", "billing",
+)
+_AUTH = (
+    "401", "403", "api key", "api_key", "unauthorized", "authentication",
+    "invalid_api_key", "permission",
+)
+
+
+def safe_failure_reason(stage: str, exc: BaseException) -> str:
+    """Return a user-safe ``"<stage>: ..."`` reason for a pipeline failure."""
+    text = str(exc).lower()
+    if stage == "embed":
+        if _contains(text, _QUOTA_OR_RATE):
+            return (
+                "embed: embedding service is temporarily unavailable "
+                "(rate limited or at capacity). Please retry shortly."
+            )
+        if _contains(text, _AUTH):
+            return "embed: embedding service is misconfigured. Contact your administrator."
+        return (
+            "embed: could not generate embeddings for this document. "
+            "Please retry; contact support if this keeps happening."
+        )
+    if stage == "extract":
+        if "empty" in text:
+            return "extract: no readable text was found in the document."
+        return (
+            "extract: could not read the document. Confirm it is a valid, "
+            "non-corrupt PDF, DOCX, or text file and try again."
+        )
+    if stage == "dispatch":
+        return "dispatch: ingestion could not be queued. Please retry shortly."
+    return f"{stage}: processing failed. Please retry."
+
+
+def _contains(text: str, needles: tuple[str, ...]) -> bool:
+    return any(n in text for n in needles)

app/workers/ingest_worker.py

@@ -16,6 +16,8 @@
 import threading
 from typing import Any
 
+import structlog
+
 from app.config import get_settings
 from app.db.admin_document_repository import (
     LocalJsonAdminDocumentRepository,
@@ -28,6 +30,9 @@
 from app.storage.object_store import get_object_store
 from app.workers.celery_app import celery_app
 from app.workers.extractors import extract_text
+from app.workers.ingest_failures import safe_failure_reason
+
+logger = structlog.get_logger(__name__)
 
 
 @celery_app.task(name="petrobrain.ingest_document", bind=True, max_retries=0)
@@ -72,12 +77,19 @@ async def _run(*, tenant_id: str, ingest_id: str) -> dict[str, Any]:
         text = extract_text(raw, record["filename"])
         if not text.strip():
             raise ValueError("extracted document text is empty")
-    except Exception as exc:  # noqa: BLE001 - surface the reason on the record
+    except Exception as exc:  # noqa: BLE001 - surface a SAFE reason on the record
+        # Full detail to the server log only; the persisted/displayed reason is
+        # sanitized so raw parser/temp-path errors never reach the admin UI.
+        logger.warning(
+            "ingest_extract_failed",
+            tenant_id=tenant_id, ingest_id=ingest_id, error=str(exc),
+            error_type=type(exc).__name__,
+        )
         repo.update_status(
             tenant_id=tenant_id,
             ingest_id=ingest_id,
             status="failed",
-            failure_reason=f"extract: {exc}",
+            failure_reason=safe_failure_reason("extract", exc),
         )
         return {"status": "failed", "ingest_id": ingest_id, "reason": str(exc)}
 
@@ -98,11 +110,19 @@ async def _run(*, tenant_id: str, ingest_id: str) -> dict[str, Any]:
             store, embedder, text=text, metadata=metadata
         )
     except Exception as exc:  # noqa: BLE001
+        # An OpenAI 429 body ("You exceeded your current quota ...") or any other
+        # provider error must NOT be persisted/shown verbatim - it leaks billing
+        # state and our provider choice. Log it for ops; store a safe reason.
+        logger.warning(
+            "ingest_embed_failed",
+            tenant_id=tenant_id, ingest_id=ingest_id, error=str(exc),
+            error_type=type(exc).__name__,
+        )
         repo.update_status(
             tenant_id=tenant_id,
             ingest_id=ingest_id,
             status="failed",
-            failure_reason=f"embed: {exc}",
+            failure_reason=safe_failure_reason("embed", exc),
         )
         return {"status": "failed", "ingest_id": ingest_id, "reason": str(exc)}