diff --git a/app/api/routes_admin_documents.py b/app/api/routes_admin_documents.py index 072b8b3..5fcf965 100644 --- a/app/api/routes_admin_documents.py +++ b/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) diff --git a/app/db/admin_document_repository.py b/app/db/admin_document_repository.py index 671d4cb..8fc586f 100644 --- a/app/db/admin_document_repository.py +++ b/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( diff --git a/app/rag/vectorstore.py b/app/rag/vectorstore.py index 0c523ba..5f69fbe 100644 --- a/app/rag/vectorstore.py +++ b/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, diff --git a/app/workers/ingest_failures.py b/app/workers/ingest_failures.py new file mode 100644 index 0000000..8cfff85 --- /dev/null +++ b/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 ``:`` 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 ``": ..."`` 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) diff --git a/app/workers/ingest_worker.py b/app/workers/ingest_worker.py index 166d812..8bd566e 100644 --- a/app/workers/ingest_worker.py +++ b/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)} diff --git a/docs/TRACK_A_SETTINGS_SPEC.md b/docs/TRACK_A_SETTINGS_SPEC.md new file mode 100644 index 0000000..04573b5 --- /dev/null +++ b/docs/TRACK_A_SETTINGS_SPEC.md @@ -0,0 +1,119 @@ +# Track A — Settings slice build spec + +Self-contained spec to build the **Settings/Profile** backend (first Track A slice) end-to-end +against this repo (`Trust-Code-System/PetroBrain`, FastAPI). CI is currently **green** — keep it +green. The separate frontend repo is **PetroBrain Web** (Next.js, at +`C:\Users\Admin\Desktop\PetroBrain Web`), which consumes this backend through its `/api/pb/*` +proxy and reveals pages via `lib/featureFlags.ts`. + +## The contract (from PetroBrain Web `lib/account/client.ts` + `types.ts`) + +Build these exact backend paths (the frontend client is unchanged): + +| Method + path | Returns / body | Shape (TS) | +|---|---|---| +| `GET /profile` | `ProfileData` | `{ id, name, email, role, org?, avatarUrl? }` | +| `PATCH /profile` | `ProfileData` | body `{ name }` | +| `POST /profile/avatar` (multipart `file`) | `ProfileData` | — | +| `GET /org` | `OrgSettings` | `{ company, country, segment, reportingBoundary, units, gwpSet, frameworks[], assetCount? }` | +| `PATCH /org` | `OrgSettings` | partial | +| `GET /settings` | `UserSettings` | `{ units, language, notifications{product,reports,alerts}, opportunityAlerts? }` | +| `PATCH /settings` | `UserSettings` | partial | +| `GET /team` | `{ items: TeamMember[] }` | `TeamMember{ id, name, email, role, status? }` | +| `GET /memory` | `{ items: CopilotMemory[] }` | `CopilotMemory{ id, content, kind?, createdAt? }` | +| `PATCH /memory/{id}` | `CopilotMemory` | body `{ content }` | +| `DELETE /memory/{id}` | 204 | — | + +Enums: `units = oilfield|metric`, `language = en|pcm|yo|ha`, `gwpSet = ar5|ar6`, +`reportingBoundary = operational_control|financial_control|equity_share`, +`segment = upstream|midstream|downstream|integrated`. + +## Data-model decisions (grounded in the existing schema) + +- `users` (migration 004) has `email, role, status, allowed_assets` but **no `name`/`avatar`**. +- `tenants` has an `attributes` JSONB. +- `tenant_memories` (012) exists but is admin/tenant-scoped (`/admin/memory`). + +So: +1. **New migration `app/db/migrations/018_account.sql`** — table `account_profiles`: + `tenant_id TEXT, user_id TEXT, display_name TEXT, avatar_url TEXT, + settings JSONB NOT NULL DEFAULT '{}'::jsonb, created_utc, updated_utc, + PRIMARY KEY (tenant_id, user_id)`. Add `ENABLE` + `FORCE ROW LEVEL SECURITY` and a + `tenant_isolation_account_profiles` policy `USING/WITH CHECK + (current_setting('petrobrain.tenant_id') = tenant_id)` — copy the exact form from + `003_assets.sql`. (Migrations auto-apply via `pg.apply_migrations` globbing `*.sql`.) +2. **Profile** = join `users` (email, role) + `account_profiles` (display_name, avatar_url) + + tenant name (org). Lazily create a default `account_profiles` row on first read. + `name` defaults to the email local-part when `display_name` is null. +3. **Settings** = `account_profiles.settings` JSONB, merged over a default `UserSettings`. + PATCH does a shallow merge + revalidate enums. +4. **Org** = read/write `tenants.attributes` (keys: company, country, segment, + reportingBoundary, units, gwpSet, frameworks). `company` falls back to `tenants.name`. + `assetCount` = `SELECT count(*) FROM assets` (tenant-scoped). **PATCH /org gated to + `require_role("admin","tenant_owner","platform_admin")`.** +5. **Team** = `users` rows for the tenant → `TeamMember`. `name` from `account_profiles` if + present, else email local-part. `status`: map `active`→active, `invited`→invited. +6. **Memory** = the signed-in user's own copilot memories. Reuse `tenant_memory_repository` + filtered to `created_by == who.user_id`; map `body→content`, keep `kind`, `created_utc→createdAt`. + PATCH runs `check_memory_body` (app.core.memory_guard) before saving; DELETE archives/removes. +7. **Avatar** = validate (reuse the `_scan_upload`/object-store pattern from + `routes_admin_documents`), store via `app.storage.object_store.object_key_for` + + `get_object_store()`, set `account_profiles.avatar_url`, return `ProfileData`. *(Avatar may + ship as a follow-up commit if object-store wiring is heavy — the page degrades without it.)* + +## Files to add/change (backend) + +- `app/db/migrations/018_account.sql` — table + RLS (above). +- `app/db/account_repository.py` — `LocalJsonAccountRepository` + `PostgresAccountRepository` + + `get_account_repository()` factory (mirror `research_repository.py` structure exactly, + incl. the `builtins.list` annotation convention if you add a `list`-named method, and the + tenant-scoped `pg.tenant_connection`). Methods: `get_profile`, `upsert_profile`, + `get_settings`, `update_settings`. +- `app/api/routes_account.py` — `router = APIRouter(tags=["account"])` with the routes above, + each `Depends(get_principal)`. (No prefix — paths are top-level `/profile`, `/org`, etc.) +- `app/models/schemas.py` — pydantic request/response models matching the TS shapes. +- `app/main.py` — `app.include_router(routes_account.router)` (find where the other routers + are included and add it). +- Tests: `tests/test_account.py` (LocalJson + TestClient with `auth_helpers`, like + `test_auth_me.py`) and `tests/test_account_postgres.py` (PG-gated, NOSUPERUSER role pattern + from `test_assets_postgres.py`) — cover tenant isolation (a second tenant can't read/patch + the first's profile/settings), the default-on-first-read, and the `/org` admin gate. + +## Frontend reconciliation (PetroBrain Web) + +- Paths already match (`/profile`, `/org`, `/settings`, `/team`, `/memory`) — **no client + change needed** if you build the exact paths above. Verify `lib/account/client.ts`. +- **Reveal:** add `"/app/settings"` and `"/app/profile"` to `LIVE_APP_HREFS` in + `lib/featureFlags.ts` (and update the reveal-on-ship comment). Build + the page goes live. + +## Verification loop (must stay green — every step is merge-blocking) + +```bash +# backend (from PetroBrain/) +python -m ruff check . && python -m mypy app/ && python -m pytest tests/ -q +python tests/eval_harness.py # red-team safety eval must exit 0 +# frontend (from PetroBrain Web/) +npm run lint && npm run typecheck && npm test && npm run build +``` +Then: branch `feat/track-a-settings`, commit, push, open PR, confirm the full CI board green +(`backend`, `frontend`, `security scan`, `terraform`, `tier-b`), merge. Reveal the page in a +small frontend PR (or the same change set) and confirm Vercel deploys. + +## Gotchas (learned the hard way) +- A method named `list` on a class shadows builtin `list` in annotations under + `from __future__ import annotations` → use `builtins.list[...]` in that file (see the repos). +- `dict_rows=True` connections return dicts at runtime but mypy types them as tuples — `cast` + where needed. +- The red-team eval (`tests/eval_harness.py`) runs in CI with no API keys (deterministic) and + is **merge-blocking** — run it locally before pushing. +- `tenant_connection` sets/0-resets the `petrobrain.tenant_id` GUC; never query tenant data + outside it. RLS is `FORCE`d, so the app role can't see across tenants even with a bad WHERE. + +## Then: the remaining Track A slices (same loop) +1. **Analytics + Reports** — `GET /analytics/emissions|insights`, `GET /reports/summary`, + `POST /reports`, `GET|POST /reports/schedules`, `DELETE /reports/schedules/{id}`. +2. **Data Tools** — `POST /data/import`, `GET /data/template|export|quality`, `POST /data/batch`, + `GET /data/batch/{id}`. +3. **Emissions rework** — reconcile the frontend emissions client to the backend's existing + **inventory** model (`POST /emissions/inventory`, `GET /emissions/inventories`) instead of the + assumed `scope-summary/sources/financed/reports/reconciliation` shape. diff --git a/frontend/apps/web/app/admin/documents/components/DocumentsScreen.tsx b/frontend/apps/web/app/admin/documents/components/DocumentsScreen.tsx index aaf0ce3..ea4dede 100644 --- a/frontend/apps/web/app/admin/documents/components/DocumentsScreen.tsx +++ b/frontend/apps/web/app/admin/documents/components/DocumentsScreen.tsx @@ -11,6 +11,7 @@ import { BackLink, Badge, Button } from '@petrobrain/ui'; import { fetchAssets } from '@/lib/chat/assets'; import { useChatStore } from '@/lib/chat/store'; import { + deleteAdminDocument, getAdminDocument, listAdminDocuments, requeueAdminDocument, @@ -48,6 +49,7 @@ export function DocumentsScreen() { const [pending, setPending] = useState([]); const [requeuingId, setRequeuingId] = useState(null); + const [deletingId, setDeletingId] = useState(null); const [filters, setFilters] = useState({ status: 'all', type: 'all', @@ -131,6 +133,39 @@ export function DocumentsScreen() { onSettled: () => void queryClient.invalidateQueries({ queryKey: DOCUMENTS_QUERY_KEY }), }); + const deleteMutation = useMutation({ + mutationFn: (ingestId: string) => + deleteAdminDocument({ baseUrl: apiBaseUrl, token, ingestId }), + // Optimistic: drop the row immediately, restore it if the request fails. + onMutate: async (ingestId) => { + setDeletingId(ingestId); + await queryClient.cancelQueries({ queryKey: DOCUMENTS_QUERY_KEY }); + const previous = queryClient.getQueryData(DOCUMENTS_QUERY_KEY) ?? []; + queryClient.setQueryData(DOCUMENTS_QUERY_KEY, (old) => + (old ?? []).filter((r) => r.ingest_id !== ingestId), + ); + return { previous }; + }, + onError: (_err, _ingestId, ctx) => { + if (ctx?.previous) { + queryClient.setQueryData(DOCUMENTS_QUERY_KEY, ctx.previous); + } + }, + onSettled: () => { + setDeletingId(null); + void queryClient.invalidateQueries({ queryKey: DOCUMENTS_QUERY_KEY }); + }, + }); + + const handleDelete = (ingestId: string) => { + const row = (documentsQuery.data ?? []).find((r) => r.ingest_id === ingestId); + const label = row ? `"${row.title}"` : 'this document'; + if (typeof window !== 'undefined' && !window.confirm(`Delete ${label}? This removes the file and its indexed chunks and cannot be undone.`)) { + return; + } + deleteMutation.mutate(ingestId); + }; + const filteredRows = useMemo( () => filterRows(documentsQuery.data ?? [], filters), [documentsQuery.data, filters], @@ -225,6 +260,8 @@ export function DocumentsScreen() { emptyState={'No documents match the current filters. Drop a file above to seed the index.'} onRequeue={(ingestId) => requeueMutation.mutate(ingestId)} requeuingId={requeuingId} + onDelete={handleDelete} + deletingId={deletingId} /> diff --git a/frontend/apps/web/app/admin/documents/components/DocumentsTable.test.tsx b/frontend/apps/web/app/admin/documents/components/DocumentsTable.test.tsx index ef4c1ad..3a54f25 100644 --- a/frontend/apps/web/app/admin/documents/components/DocumentsTable.test.tsx +++ b/frontend/apps/web/app/admin/documents/components/DocumentsTable.test.tsx @@ -134,6 +134,50 @@ describe('DocumentsTable', () => { ); expect(screen.queryByRole('button', { name: 'Requeue' })).toBeNull(); }); + + it('shows Delete on every persisted row and fires onDelete with the ingest id', () => { + const onDelete = vi.fn(); + render( + , + ); + + const doneRow = screen.getByTestId('row-ing-done'); + expect(within(doneRow).getByRole('button', { name: 'Delete' })).toBeInTheDocument(); + + fireEvent.click(within(doneRow).getByRole('button', { name: 'Delete' })); + expect(onDelete).toHaveBeenCalledWith('ing-done'); + }); + + it('does not offer Delete for an optimistic (not-yet-persisted) row', () => { + render( + , + ); + expect(screen.queryByRole('button', { name: 'Delete' })).toBeNull(); + }); + + it('does not render the Delete action when no onDelete handler is given', () => { + render( + , + ); + expect(screen.queryByRole('button', { name: 'Delete' })).toBeNull(); + }); }); describe('filterRows', () => { diff --git a/frontend/apps/web/app/admin/documents/components/DocumentsTable.tsx b/frontend/apps/web/app/admin/documents/components/DocumentsTable.tsx index 8806e08..3651a34 100644 --- a/frontend/apps/web/app/admin/documents/components/DocumentsTable.tsx +++ b/frontend/apps/web/app/admin/documents/components/DocumentsTable.tsx @@ -13,6 +13,11 @@ function canRequeue(row: AdminDocumentRow): boolean { return REQUEUEABLE.has(row.status) && !row.ingest_id.startsWith('optimistic-'); } +/** Optimistic rows have no backend record yet, so they can't be deleted. */ +function canDelete(row: AdminDocumentRow): boolean { + return !row.ingest_id.startsWith('optimistic-'); +} + export interface DocumentsTableProps { rows: AdminDocumentRow[]; isLoading: boolean; @@ -22,6 +27,10 @@ export interface DocumentsTableProps { onRequeue?: (ingestId: string) => void; /** ingest_id currently being re-dispatched, for the per-row spinner. */ requeuingId?: string | null; + /** When provided, every persisted row shows a "Delete" action. */ + onDelete?: (ingestId: string) => void; + /** ingest_id currently being deleted, for the per-row spinner. */ + deletingId?: string | null; } export function DocumentsTable({ @@ -31,7 +40,10 @@ export function DocumentsTable({ emptyState, onRequeue, requeuingId, + onDelete, + deletingId, }: DocumentsTableProps) { + const showActions = Boolean(onRequeue || onDelete); if (isError) { return (

@@ -58,7 +70,7 @@ export function DocumentsTable({ Status Chunks Updated - {onRequeue ? Actions : null} + {showActions ? Actions : null} @@ -83,19 +95,32 @@ export function DocumentsTable({ {formatRelative(row.updated_utc)} - {onRequeue ? ( - - {canRequeue(row) ? ( - - ) : null} + {showActions ? ( + +

+ {onRequeue && canRequeue(row) ? ( + + ) : null} + {onDelete && canDelete(row) ? ( + + ) : null} +
) : null} diff --git a/frontend/apps/web/lib/admin-documents/api.ts b/frontend/apps/web/lib/admin-documents/api.ts index c442133..074b67e 100644 --- a/frontend/apps/web/lib/admin-documents/api.ts +++ b/frontend/apps/web/lib/admin-documents/api.ts @@ -48,6 +48,30 @@ export async function requeueAdminDocument( return (await resp.json()) as AdminDocumentRow; } +export interface DeleteAdminDocumentResult { + ingest_id: string; + deleted: boolean; + chunks_deleted: number; +} + +/** + * Permanently remove an uploaded document: its ingest record, the stored blob, + * and (when no sibling ingest shares its document_id) its indexed chunks. The + * backend gates on ``role=admin`` and scopes the delete to the caller's tenant. + */ +export async function deleteAdminDocument( + opts: RequestOpts & { ingestId: string }, +): Promise { + const init: RequestInit = { + method: 'DELETE', + headers: { Authorization: `Bearer ${opts.token}` }, + }; + if (opts.signal) init.signal = opts.signal; + const resp = await fetch(new URL(`/admin/documents/${opts.ingestId}`, opts.baseUrl), init); + if (!resp.ok) throw await apiError(resp); + return (await resp.json()) as DeleteAdminDocumentResult; +} + export interface RequeueStuckResult { requeued: number; results: { ingest_id: string; status: string; detail?: string }[]; diff --git a/tests/test_admin_document_upload.py b/tests/test_admin_document_upload.py index b8d99f2..d72de3a 100644 --- a/tests/test_admin_document_upload.py +++ b/tests/test_admin_document_upload.py @@ -421,6 +421,107 @@ def _boom(*args, **kwargs): assert all(r["status"] == "done" for r in admin_repo.list_records(tenant_id="tenant-a")) +def test_delete_removes_record_object_and_chunks(monkeypatch, admin_repo, memory_store): + chunk_deletes: list[tuple[str, str]] = [] + + async def _fake_delete_chunks(tenant_id, document_id): + chunk_deletes.append((tenant_id, document_id)) + return 3 + + monkeypatch.setattr(routes_admin_documents, "_delete_vector_chunks", _fake_delete_chunks) + + up = client.post( + "/admin/documents", + headers=_admin_headers(), + data={"metadata": json.dumps(_payload(document_id="SOP-DEL-1"))}, + files={"file": ("kick.md", _KICK_MD.encode("utf-8"), "text/markdown")}, + ) + assert up.json()["status"] == "done" + ingest_id = up.json()["ingest_id"] + object_key = admin_repo.get(tenant_id="tenant-a", ingest_id=ingest_id)["object_key"] + + rd = client.delete(f"/admin/documents/{ingest_id}", headers=_admin_headers()) + assert rd.status_code == 200, rd.text + body = rd.json() + assert body["deleted"] is True + assert body["chunks_deleted"] == 3 + + # Record gone, blob gone, chunks purged for this document_id. + assert admin_repo.get(tenant_id="tenant-a", ingest_id=ingest_id) is None + assert admin_repo.list_records(tenant_id="tenant-a") == [] + with pytest.raises(KeyError): + memory_store.get(object_key) + assert chunk_deletes == [("tenant-a", "SOP-DEL-1")] + + +def test_delete_keeps_chunks_when_a_sibling_shares_document_id(monkeypatch, admin_repo, memory_store): + chunk_deletes: list[tuple[str, str]] = [] + + async def _fake_delete_chunks(tenant_id, document_id): + chunk_deletes.append((tenant_id, document_id)) + return 0 + + monkeypatch.setattr(routes_admin_documents, "_delete_vector_chunks", _fake_delete_chunks) + + # Two uploads share the same document_id (chunks are keyed by document_id). + ids = [] + for _ in range(2): + up = client.post( + "/admin/documents", + headers=_admin_headers(), + data={"metadata": json.dumps(_payload(document_id="SOP-DUP-1"))}, + files={"file": ("kick.md", _KICK_MD.encode("utf-8"), "text/markdown")}, + ) + ids.append(up.json()["ingest_id"]) + + # Deleting the first must NOT purge chunks - the sibling still represents them. + rd = client.delete(f"/admin/documents/{ids[0]}", headers=_admin_headers()) + assert rd.status_code == 200 + assert rd.json()["chunks_deleted"] == 0 + assert chunk_deletes == [] + + # Deleting the last sibling now purges the chunks. + rd2 = client.delete(f"/admin/documents/{ids[1]}", headers=_admin_headers()) + assert rd2.status_code == 200 + assert chunk_deletes == [("tenant-a", "SOP-DUP-1")] + + +def test_delete_unknown_id_404s(): + rd = client.delete("/admin/documents/does-not-exist", headers=_admin_headers()) + assert rd.status_code == 404 + + +def test_delete_requires_admin_role(): + rd = client.delete( + "/admin/documents/whatever", + headers=auth_headers(role="engineer", allowed_assets=["Asset-A"]), + ) + assert rd.status_code == 403 + + +def test_delete_is_tenant_isolated(monkeypatch, admin_repo): + async def _noop(tenant_id, document_id): + return 0 + + monkeypatch.setattr(routes_admin_documents, "_delete_vector_chunks", _noop) + + up = client.post( + "/admin/documents", + headers=_admin_headers(), + data={"metadata": json.dumps(_payload(document_id="SOP-ISO-1"))}, + files={"file": ("kick.md", _KICK_MD.encode("utf-8"), "text/markdown")}, + ) + ingest_id = up.json()["ingest_id"] + + # Tenant B cannot delete tenant A's document. + rd = client.delete( + f"/admin/documents/{ingest_id}", + headers=_admin_headers(tenant_id="tenant-b", user_id="bob", allowed_assets=["*"]), + ) + assert rd.status_code == 404 + assert admin_repo.get(tenant_id="tenant-a", ingest_id=ingest_id) is not None + + def test_worker_marks_failed_when_object_missing(admin_repo, memory_store): # Create the record by uploading, then nuke the bytes so the worker fails. r = client.post( diff --git a/tests/test_ingest_failures.py b/tests/test_ingest_failures.py new file mode 100644 index 0000000..1d81da9 --- /dev/null +++ b/tests/test_ingest_failures.py @@ -0,0 +1,55 @@ +"""The ingestion failure_reason shown to admins must never leak raw provider +error text (billing state, provider identity, internal hosts, temp paths).""" +from __future__ import annotations + +from app.workers.ingest_failures import safe_failure_reason + +# The exact shape of the OpenAI quota error that leaked to the admin UI. +_OPENAI_429 = ( + "Error code: 429 - {'error': {'message': 'You exceeded your current quota, " + "please check your plan and billing details. For more information on this " + "error, read the docs: https://platform.openai.com/docs/guides/error-codes/" + "api-errors.', 'type': 'insufficient_quota', 'param': None, 'code': " + "'insufficient_quota'}}" +) + + +def _leaks_secrets(reason: str) -> bool: + lowered = reason.lower() + return any( + token in lowered + for token in ("quota", "billing", "openai", "platform.openai", "http", "{", "error code") + ) + + +def test_embed_quota_error_is_sanitized(): + reason = safe_failure_reason("embed", RuntimeError(_OPENAI_429)) + assert reason.startswith("embed:") + assert not _leaks_secrets(reason) + assert "retry" in reason.lower() + + +def test_embed_auth_error_is_sanitized(): + reason = safe_failure_reason("embed", RuntimeError("Error code: 401 - invalid_api_key")) + assert reason.startswith("embed:") + assert "api_key" not in reason.lower() + assert "administrator" in reason.lower() + + +def test_extract_empty_is_specific_but_safe(): + reason = safe_failure_reason("extract", ValueError("extracted document text is empty")) + assert reason.startswith("extract:") + assert "no readable text" in reason.lower() + + +def test_extract_generic_error_hides_internal_detail(): + reason = safe_failure_reason("extract", RuntimeError("/tmp/abc123/upload.pdf: bad xref at 0xDEAD")) + assert reason.startswith("extract:") + assert "/tmp" not in reason + assert "0xdead" not in reason.lower() + + +def test_dispatch_error_hides_transport_detail(): + reason = safe_failure_reason("dispatch", RuntimeError("No such transport: 'redis://secret-host:6379'")) + assert reason.startswith("dispatch:") + assert "secret-host" not in reason