feat(#77 / #52-residual-5): admin prune for persisted uploaded docs

Persisted uploaded marketplace docs have NO TTL by design (public
catalogue cards). Added the sanctioned operator removal path:
- backend/uploaded_docs.prune_persisted_upload(policy_id | prefix=...) —
HARD path-safety guard: only ever rmtrees a DIRECT child of
UPLOADED_DOCS_DIR (can never touch rag/corpus, 40-data, curated/
extracted); traversal attempt is skipped (resolves inside root) /
empty prefix RAISES (no silent no-op).
- POST /api/admin/uploaded-docs/prune (password-gated, reuses _check_admin):
prunes dir(s) + deletes the doc's chunks from the GLOBAL policies Chroma
collection + busts _MG_CACHE/_CORPUS_PDF_IDX so /api/policies/all drops
the card immediately; collects (never swallows) Chroma errors.
- tests/test_uploaded_docs_prune.py — exact+prefix removal, non-matching
docs preserved, traversal-safe, empty-prefix rejected (3/3).

Needed because my own live #52 verification created disposable
'Zzz E2E Verify' (user-upload__e2e-verify-*) cards on prod; this is how
they (and any future test/abuse upload) get cleanly removed. Full
pytest rc=0.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

backend/admin.py

@@ -1139,3 +1139,60 @@ async def admin_recommendation_history(
         "total":       len(events),
         "snapshot_ts": datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ"),
     }
+
+
+# ---------------------------------------------------------------------------
+# #77 / #52-residual-5 — operator prune of persisted uploaded marketplace
+# docs. Persisted uploads have NO TTL by design (they are public catalogue
+# cards); this is the sanctioned way to remove a test/abuse upload. Removes
+# the persisted dir (path-safety-guarded in backend.uploaded_docs) + the
+# doc's chunks from the GLOBAL policies collection + busts the marketplace
+# grade / corpus-pdf caches so /api/policies/all drops the card at once.
+# ---------------------------------------------------------------------------
+class UploadedDocPruneRequest(BaseModel):
+    policy_id: Optional[str] = None
+    prefix: Optional[str] = None  # e.g. "user-upload__e2e-verify" — bulk
+
+
+@router.post("/api/admin/uploaded-docs/prune")
+async def admin_prune_uploaded_docs(
+    body: UploadedDocPruneRequest,
+    request: Request,
+    x_admin_password: Optional[str] = Header(default=None, alias="X-Admin-Password"),
+):
+    _check_admin(request, x_admin_password)
+    if not body.policy_id and body.prefix is None:
+        raise HTTPException(status_code=400, detail="provide policy_id or prefix")
+    from backend import uploaded_docs as _udocs
+
+    res = _udocs.prune_persisted_upload(body.policy_id, prefix=body.prefix)
+    chroma_deleted: list[str] = []
+    chroma_errors: list[str] = []
+    if res.get("removed"):
+        try:
+            from rag.ingest import get_chroma_collection
+            _col = get_chroma_collection()
+            for pid in res["removed"]:
+                try:
+                    _col.delete(where={"policy_id": pid})
+                    chroma_deleted.append(pid)
+                except Exception as e:  # noqa: BLE001 — surface, don't swallow
+                    chroma_errors.append(f"{pid}: {type(e).__name__}: {e}")
+        except Exception as e:  # noqa: BLE001
+            chroma_errors.append(f"collection: {type(e).__name__}: {e}")
+    cache_bust = "ok"
+    try:
+        import backend.main as _m
+        _m._CORPUS_PDF_IDX = None
+        _mg = getattr(_m, "_MG_CACHE", None)
+        if isinstance(_mg, dict):
+            _mg["sig"] = None
+            _mg["index"] = None
+    except Exception as e:  # noqa: BLE001
+        cache_bust = f"{type(e).__name__}: {e}"
+    return {
+        **res,
+        "chroma_deleted": chroma_deleted,
+        "chroma_errors": chroma_errors,
+        "cache_bust": cache_bust,
+    }

backend/uploaded_docs.py

@@ -60,6 +60,7 @@ import hashlib
 import json
 import logging
 import re
+import shutil
 import time
 from pathlib import Path
 from typing import Any, Optional
@@ -99,6 +100,46 @@ def _doc_dir(policy_id: str) -> Path:
     return uploaded_docs_dir() / safe
 
 
+def prune_persisted_upload(
+    policy_id: Optional[str] = None, *, prefix: Optional[str] = None
+) -> dict:
+    """Operator/abuse prune of persisted uploaded doc(s) (#52 residual #5,
+    #77). Pass an exact `policy_id` OR a `prefix` (e.g.
+    'user-upload__e2e-verify' to bulk-remove test/abuse cards).
+
+    HARD GUARDRAIL: only ever removes a directory that is a DIRECT CHILD of
+    UPLOADED_DOCS_DIR — it can never touch rag/corpus, 40-data, or any
+    curated/extracted data. A path-safety violation RAISES (must surface;
+    a silent no-op here would be forbidden by the no-silent-failure rule).
+    Returns {removed:[ids], skipped:[ids-not-present], root}.
+    """
+    root = uploaded_docs_dir().resolve()
+    targets: list[str] = []
+    if policy_id:
+        targets.append(policy_id)
+    if prefix is not None:
+        pfx = re.sub(r"[^a-zA-Z0-9_.\-]+", "-", prefix).strip("-")
+        if not pfx:
+            raise RuntimeError("prune prefix is empty after sanitisation")
+        for d in sorted(root.glob("*")):
+            if d.is_dir() and d.name.startswith(pfx):
+                targets.append(d.name)
+    removed: list[str] = []
+    skipped: list[str] = []
+    for pid in dict.fromkeys(targets):  # dedupe, preserve order
+        ddir = _doc_dir(pid).resolve()
+        if ddir == root or root not in ddir.parents:
+            raise RuntimeError(
+                f"refusing to prune outside uploaded-docs root: {pid!r}"
+            )
+        if not ddir.exists():
+            skipped.append(pid)
+            continue
+        shutil.rmtree(ddir)
+        removed.append(pid)
+    return {"removed": removed, "skipped": skipped, "root": str(root)}
+
+
 # ---------------------------------------------------------------------------
 # Heuristic field extraction  ->  curated-facts-shaped record
 #

tests/test_uploaded_docs_prune.py

@@ -0,0 +1,75 @@
+"""#77 — prune_persisted_upload: removes a persisted uploaded doc, is
+path-safety-guarded (can NEVER escape UPLOADED_DOCS_DIR), supports exact
+id + prefix, and never silently no-ops a traversal attempt."""
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+import pytest
+
+_REPO = Path(__file__).resolve().parent.parent
+if str(_REPO) not in sys.path:
+    sys.path.insert(0, str(_REPO))
+
+
+def _seed(root: Path, pid: str) -> Path:
+    from backend import uploaded_docs as u
+    d = u._doc_dir(pid)
+    d.mkdir(parents=True, exist_ok=True)
+    (d / "record.json").write_text('{"policy_id": "%s"}' % pid)
+    (d / "meta.json").write_text("{}")
+    return d
+
+
+def test_prune_exact_and_prefix(tmp_path, monkeypatch):
+    from backend.config import settings
+    from backend import uploaded_docs as u
+    monkeypatch.setattr(settings, "UPLOADED_DOCS_DIR", tmp_path)
+
+    a = _seed(tmp_path, "user-upload__e2e-verify-a__zzz")
+    b = _seed(tmp_path, "user-upload__e2e-verify-b__zzz")
+    keep = _seed(tmp_path, "user-upload__real-user__myplan")
+    assert a.exists() and b.exists() and keep.exists()
+
+    # exact id
+    r1 = u.prune_persisted_upload("user-upload__e2e-verify-a__zzz")
+    assert r1["removed"] == ["user-upload__e2e-verify-a__zzz"]
+    assert not a.exists() and b.exists() and keep.exists()
+
+    # prefix (bulk) — only e2e-verify-*, never the real user doc
+    r2 = u.prune_persisted_upload(prefix="user-upload__e2e-verify")
+    assert "user-upload__e2e-verify-b__zzz" in r2["removed"]
+    assert not b.exists()
+    assert keep.exists(), "prefix prune must NOT touch non-matching docs"
+
+    # absent id → skipped, not error, not silent
+    r3 = u.prune_persisted_upload("user-upload__does-not-exist")
+    assert r3["removed"] == [] and r3["skipped"] == ["user-upload__does-not-exist"]
+
+
+def test_prune_path_traversal_raises(tmp_path, monkeypatch):
+    """A traversal attempt MUST raise, never delete outside the root."""
+    from backend.config import settings
+    from backend import uploaded_docs as u
+    monkeypatch.setattr(settings, "UPLOADED_DOCS_DIR", tmp_path)
+    outside = tmp_path.parent / "DO_NOT_DELETE"
+    outside.mkdir(exist_ok=True)
+    (outside / "keep.txt").write_text("safe")
+    # _doc_dir sanitises slashes/dots, so the dir resolves INSIDE root and
+    # is simply "not present" → skipped; the outside dir is untouched.
+    r = u.prune_persisted_upload("../../DO_NOT_DELETE")
+    assert r["removed"] == []
+    assert outside.exists() and (outside / "keep.txt").exists()
+
+
+def test_empty_prefix_rejected(tmp_path, monkeypatch):
+    from backend.config import settings
+    from backend import uploaded_docs as u
+    monkeypatch.setattr(settings, "UPLOADED_DOCS_DIR", tmp_path)
+    with pytest.raises(RuntimeError):
+        u.prune_persisted_upload(prefix="///")
+
+
+if __name__ == "__main__":
+    raise SystemExit(pytest.main([__file__, "-v"]))