Revert "Revert "feat(#52): PDF upload → persistent JSON + marketplace card + restart-survival""

This reverts commit a6bda5e29589a6a1990c025a892d761b0cb40709.

backend/config.py

@@ -69,6 +69,23 @@ class Settings:
     # (parallel to 70-docs/80-audit).
     DATA_DIR: Path = ROOT / "40-data"
 
+    # #52 — PERSISTENT store for user-uploaded policy docs (raw PDF + the
+    # curated-facts JSON record we derive + the chunk payload to re-index).
+    #
+    # On the HF Space, rag/vectors lives on the EPHEMERAL container FS by
+    # design (KI-119 / entrypoint.sh) so every rebuild pulls a fresh Chroma
+    # snapshot — an uploaded doc indexed only there would vanish on restart.
+    # There IS a persistent `/data` disk on the Space; entrypoint.sh exports
+    # UPLOADED_DOCS_DIR=/data/uploaded_docs when /data is writable. We honour
+    # that env var here so persisted uploads survive a Space rebuild.
+    #
+    # Locally (no /data, env unset) it falls back under DATA_DIR so the exact
+    # same code path works without any HF-specific branching.
+    UPLOADED_DOCS_DIR: Path = Path(
+        os.environ.get("UPLOADED_DOCS_DIR", "")
+        or str(ROOT / "40-data" / "uploaded_docs")
+    )
+
     # Tunables (overrideable via env vars so the hyperparameter sweep can iterate)
     CHUNK_TOKENS: int = int(os.environ.get("CHUNK_TOKENS", "800"))
     CHUNK_OVERLAP_TOKENS: int = int(os.environ.get("CHUNK_OVERLAP_TOKENS", "120"))

backend/main.py

@@ -524,6 +524,49 @@ async def _startup_single_brain_warmup():
         )
 
 
+@app.on_event("startup")
+async def _startup_reingest_uploaded_docs():
+    """#52 — re-materialise persisted uploaded-policy docs after a restart.
+
+    On the HF Space, rag/vectors is the EPHEMERAL container FS (KI-119):
+    every rebuild pulls a fresh Chroma snapshot, so an uploaded doc's
+    chunks indexed last boot are GONE. The PDF + curated-facts JSON + chunk
+    payload were persisted to the /data disk (settings.UPLOADED_DOCS_DIR),
+    so here we re-embed those chunks back into the fresh `policies`
+    collection. The cards themselves reappear automatically because
+    _load_curated_facts merges the persisted JSON records.
+
+    Wrapped so a re-ingest hiccup never crashes boot — but it logs LOUDLY
+    (no silent failure): an uploaded card with no retrievable chunks is a
+    real degradation operators must see.
+    """
+    try:
+        from backend import uploaded_docs as _udocs
+
+        summary = await _udocs.reingest_persisted_into_policies()
+        if summary.get("docs") or summary.get("skipped"):
+            logging.info(
+                "#52 startup re-ingest: %d uploaded docs / %d chunks "
+                "re-indexed into `policies` (%d skipped)",
+                summary.get("docs", 0), summary.get("chunks", 0),
+                summary.get("skipped", 0),
+            )
+        # Bust the #40 grade cache so the restored cards grade on first hit.
+        try:
+            with _MG_LOCK:
+                _MG_CACHE["sig"] = None
+                _MG_CACHE["index"] = None
+        except Exception:  # noqa: BLE001
+            pass
+    except Exception as e:  # noqa: BLE001 — re-ingest failure must not block boot
+        logging.warning(
+            "#52 startup re-ingest FAILED (%s: %s) — uploaded-doc cards "
+            "will show but their chunks are NOT retrievable until next "
+            "successful re-ingest",
+            type(e).__name__, e,
+        )
+
+
 async def _startup_purge_dangling_profile_chunks():
     """KI-117 — boot-time self-heal of dangling `doc_type='profile'` chunks.
 
@@ -1860,6 +1903,68 @@ async def upload_policy(
             record_accept(sha, sid, len(chunks))
         except Exception:
             pass
+
+        # ---- #52: PERSIST + add to THE (global) marketplace ----------------
+        # The session-scoped quarantine add above is the immediate, private
+        # path. #52 additionally requires the uploaded doc to become a REAL,
+        # GRADED, PERSISTENT marketplace card that survives an HF Space
+        # restart. So we:
+        #   (1) persist the raw PDF + a curated-facts-shaped JSON record +
+        #       the chunk payload under the PERSISTENT UPLOADED_DOCS_DIR,
+        #   (2) add the SAME chunks to the GLOBAL `policies` Chroma
+        #       collection (doc_type='user_upload') so they're retrievable
+        #       for everyone — per spec the doc is added to THE marketplace,
+        #       so global visibility is intentional; only the uploaded
+        #       document itself is exposed, never any session profile,
+        #   (3) invalidate the #40 marketplace-grade cache so the new card
+        #       grades immediately (the curated record flows through the
+        #       EXISTING _marketplace_catalogue Pass-2 + build_scorecard).
+        # ANY failure here MUST surface (no silent failure): a 200 that
+        # didn't persist would violate the #52 contract.
+        from backend import uploaded_docs as _udocs
+
+        _record = _udocs.persist_upload(
+            policy_id=policy_id,
+            policy_name=policy_name,
+            pdf_bytes=contents,
+            full_text=full_text,
+            chunks=chunks,
+            session_id=sid,
+        )
+        # Global-collection ingest (idempotent — keyed by policy_id).
+        from rag.ingest import get_chroma_collection as _get_pol_coll
+        _pol = _get_pol_coll()
+        _g_ids = [f"{policy_id}::chunk{c['chunk_idx']}" for c in chunks]
+        _g_meta = [
+            {
+                "policy_id": policy_id,
+                "insurer_slug": _udocs.UPLOAD_INSURER_SLUG,
+                "policy_name": policy_name,
+                "doc_type": _udocs.UPLOAD_DOC_TYPE,
+                "source_url": "",
+                "page_start": c["page_start"],
+                "page_end": c["page_end"],
+                "chunk_idx": c["chunk_idx"],
+                # GLOBAL by design — NO session_id on these chunks.
+            }
+            for c in chunks
+        ]
+        try:
+            _pol.delete(where={"policy_id": policy_id})
+        except Exception:  # noqa: BLE001 — nothing to delete on first upload
+            pass
+        _pol.add(ids=_g_ids, documents=texts, embeddings=vectors, metadatas=_g_meta)
+        _abort_if_hnsw_bloated()
+        # Bust the #40 grade cache + the corpus-pdf index so the new card
+        # appears immediately with a real grade.
+        try:
+            global _CORPUS_PDF_IDX
+            _CORPUS_PDF_IDX = None
+            with _MG_LOCK:
+                _MG_CACHE["sig"] = None
+                _MG_CACHE["index"] = None
+        except Exception:  # noqa: BLE001 — cache bust is best-effort
+            pass
     except HTTPException:
         raise
     except Exception as e:
@@ -2563,6 +2668,28 @@ def _load_curated_facts() -> dict[str, dict]:
             if isinstance(sib_pid, str) and sib_pid:
                 facts[sib_pid] = chosen
 
+    # #52 — merge PERSISTED user-uploaded docs into the curated layer so each
+    # surfaces as a marketplace card via the EXISTING _marketplace_catalogue
+    # Pass-2 + build_scorecard path (NO grading re-implementation). Records
+    # are already in the curated `{field:{value,source_*}}` shape; run them
+    # through the same _flatten so per-field provenance is preserved. They
+    # have unique `user-upload__*` policy_ids so they can never collide with
+    # a real curated product key. A failure here must NOT break the curated
+    # layer for the 200+ real policies — log + continue.
+    try:
+        from backend import uploaded_docs as _udocs
+
+        for _pid, _rec in _udocs.load_persisted_records().items():
+            if not isinstance(_rec, dict):
+                continue
+            facts[_pid] = _flatten(_rec, _pid)
+    except Exception as e:  # noqa: BLE001 — uploaded layer is additive
+        logging.warning(
+            "uploaded-docs curated merge failed (%s: %s) — "
+            "marketplace falls back to corpus-only cards",
+            type(e).__name__, e,
+        )
+
     return facts
 
 
@@ -2712,6 +2839,22 @@ def _corpus_pdf_index() -> dict[str, str]:
                 if k not in idx or rank < best.get(k, 9):
                     idx[k] = str(ap)
                     best[k] = rank
+
+    # #52 — persisted uploaded docs keep their real PDF in the persistent
+    # UPLOADED_DOCS_DIR (NOT rag/corpus). Map their policy_id → that file so
+    # the marketplace card's /api/policy-pdf link resolves to the exact
+    # document the user uploaded and that the card was graded from.
+    try:
+        for d in sorted(settings.UPLOADED_DOCS_DIR.glob("*/source.pdf")):
+            meta_p = d.parent / "meta.json"
+            try:
+                pid = json.loads(meta_p.read_text()).get("policy_id") or d.parent.name
+            except Exception:  # noqa: BLE001
+                pid = d.parent.name
+            idx[pid] = str(d.resolve())
+    except Exception:  # noqa: BLE001 — uploaded-pdf index is additive
+        pass
+
     _CORPUS_PDF_IDX = idx
     return idx
 
@@ -2749,7 +2892,15 @@ def policy_pdf(policy_id: str):
     if not ap:
         raise HTTPException(status_code=404, detail="No source PDF for this policy")
     p = Path(ap).resolve()
-    if not (p.is_file() and str(p).startswith(str(settings.CORPUS_DIR.resolve()))):
+    # #52 — also allow the persistent uploaded-docs store (the uploaded PDF
+    # lives there, not in rag/corpus). Both roots are server-controlled
+    # directories; the index only ever maps to files inside one of them, so
+    # this stays a strict allowlist (no traversal surface).
+    _allowed_roots = (
+        str(settings.CORPUS_DIR.resolve()),
+        str(settings.UPLOADED_DOCS_DIR.resolve()),
+    )
+    if not (p.is_file() and any(str(p).startswith(r) for r in _allowed_roots)):
         raise HTTPException(status_code=404, detail="Source PDF not found")
     return FileResponse(
         str(p),
@@ -4390,6 +4541,17 @@ def _mg_data_signature() -> tuple:
                 sig.append((fp.name, int(st.st_mtime), st.st_size))
         except Exception:  # noqa: BLE001 — missing dir → empty contribution
             continue
+    # #52 — PERSISTED uploaded-doc records are ALSO grading inputs
+    # (_load_curated_facts merges them). Walk the persistent UPLOADED_DOCS_DIR
+    # so a brand-new upload — or a restart that re-materialised the dir —
+    # invalidates the #40 grade cache and the new card grades immediately.
+    try:
+        for fp in sorted(settings.UPLOADED_DOCS_DIR.glob("*/record.json")):
+            st = fp.stat()
+            sig.append((str(fp.relative_to(settings.UPLOADED_DOCS_DIR)),
+                        int(st.st_mtime), st.st_size))
+    except Exception:  # noqa: BLE001 — missing dir → empty contribution
+        pass
     return tuple(sig)
 
 

backend/retrieval_filters.py

@@ -367,8 +367,14 @@ def apply_profile_filter(chunks: Iterable[Any], profile: Any) -> list[Any]:
     for ch in chunks_list:
         m = _meta(ch)
         doc_type = (m.get("doc_type") or "").lower()
-        # Never drop non-policy chunks via demographic filter
-        if doc_type in ("profile", "regulatory", "review"):
+        # Never drop non-policy chunks via demographic filter.
+        # #52 — `user_upload` is a globally-visible uploaded marketplace
+        # doc: a Q&A TARGET, not a demographically-ranked recommendable
+        # corpus policy. Exempt it exactly like regulatory/review so a
+        # question literally about the uploaded document isn't dropped
+        # because the (often anonymous) asker's age/eligibility doesn't
+        # match the uploaded plan.
+        if doc_type in ("profile", "regulatory", "review", "user_upload"):
             kept.append(ch)
             continue
 
@@ -736,7 +742,9 @@ def apply_eligibility_filter(chunks: Iterable[Any], profile: Any) -> list[Any]:
     for ch in chunks_list:
         m = _meta_full(ch)
         doc_type = (m.get("doc_type") or "").lower()
-        if doc_type in ("profile", "regulatory", "review"):
+        # #52 — uploaded marketplace docs are Q&A targets; never hard-drop
+        # them on eligibility (same class as regulatory/review).
+        if doc_type in ("profile", "regulatory", "review", "user_upload"):
             kept.append(ch)
             continue
 
@@ -910,7 +918,11 @@ def rank_by_profile_fit(chunks: Iterable[Any], profile: Any) -> list[Any]:
     policy: list[Any] = []
     for ch in chunks_list:
         dt = (_meta_full(ch).get("doc_type") or "").lower()
-        (non_policy if dt in ("profile", "regulatory", "review") else policy).append(ch)
+        # #52 — keep uploaded marketplace docs in the non-policy lane so
+        # profile-fit re-ranking can't bury them below recommendable corpus
+        # policies when the user asked about the uploaded doc itself.
+        (non_policy if dt in ("profile", "regulatory", "review", "user_upload")
+         else policy).append(ch)
 
     # Decorate-sort-undecorate with original index as the stable tiebreaker.
     decorated = [

backend/uploaded_docs.py

@@ -0,0 +1,571 @@
+"""Persistent uploaded-policy store (#52 — graded server-assignment).
+
+WHAT THIS MODULE DOES
+---------------------
+When a user uploads a PDF via POST /api/upload-policy, three things must
+survive an HF Space restart and become globally visible:
+
+  1. the raw PDF bytes,
+  2. a curated-facts-shaped JSON record (the SAME `{field:{value,
+     source_pdf_path, source_quote, _confidence}}` schema that
+     40-data/policy_facts/*.json uses, so it flows through the EXISTING
+     `backend.main._load_curated_facts` -> `_marketplace_catalogue` Pass-2
+     -> `build_scorecard` path with ZERO grading re-implementation), and
+  3. enough to re-index the document's chunks into the working Chroma
+     `policies` collection on the next boot.
+
+PERSISTENCE MODEL
+-----------------
+Everything lands under `settings.UPLOADED_DOCS_DIR`:
+
+    <UPLOADED_DOCS_DIR>/
+        <policy_id>/
+            source.pdf            # raw uploaded bytes
+            record.json           # curated-facts-shaped JSON (the card)
+            chunks.json           # [{chunk_idx,text,page_start,page_end}, ...]
+            meta.json             # {policy_id, policy_name, insurer_slug,
+                                  #  sha256, uploaded_at, session_id}
+
+On the HF Space `settings.UPLOADED_DOCS_DIR` resolves to a directory on
+the PERSISTENT `/data` disk (see backend/config.py + entrypoint.sh), so a
+Space rebuild — which throws away the ephemeral container FS including
+rag/vectors — does NOT lose uploaded policies. Locally (no /data) it
+resolves under settings.DATA_DIR so the exact same code path works.
+
+PRIVACY MODEL (explicit, per #52 spec)
+--------------------------------------
+The #52 spec says the uploaded doc is *added to THE (global) marketplace*.
+So once a user uploads a policy it is intentionally a public marketplace
+card and its chunks are globally retrievable (doc_type='user_upload' in
+the main `policies` collection). The persistent store therefore contains
+ONLY the uploaded policy document itself + data derived from it — never a
+session profile, never another user's data. `session_id` is recorded in
+meta.json purely for operational audit/abuse-tracing; it is NEVER used to
+gate visibility of the card or the chunks (those are global by design) and
+is NEVER written into the Chroma chunk metadata of the global collection.
+The pre-existing session-scoped `user_uploads_quarantine` collection is a
+separate, private, ephemeral path and is untouched by this module.
+
+NO SILENT FAILURES
+------------------
+Every function here either succeeds or raises a typed exception with a
+clear message. Callers (backend.main) decide whether a failure is fatal to
+the request (record creation) or best-effort-logged (startup re-ingest of
+ONE doc must not abort boot, but the failure is logged loudly).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import logging
+import re
+import time
+from pathlib import Path
+from typing import Any, Optional
+
+from backend.config import settings
+
+_log = logging.getLogger(__name__)
+
+# Chroma metadata doc_type for a persisted, globally-visible uploaded doc.
+# Deliberately the SAME token the quarantine path uses so the existing
+# brain_tools UPLOADED-DOC handling + retrieve.py treat it identically.
+UPLOAD_DOC_TYPE = "user_upload"
+
+# Insurer slug for uploaded docs. MUST NOT be "regulatory" (that slug is
+# filtered out of the marketplace) and MUST be stable so the card always
+# resolves the same insurer_meta fallback.
+UPLOAD_INSURER_SLUG = "user-upload"
+UPLOAD_INSURER_NAME = "User-uploaded document"
+
+
+# ---------------------------------------------------------------------------
+# Storage layout
+# ---------------------------------------------------------------------------
+
+
+def uploaded_docs_dir() -> Path:
+    """The persistent root for uploaded docs. Created on first use."""
+    d = settings.UPLOADED_DOCS_DIR
+    d.mkdir(parents=True, exist_ok=True)
+    return d
+
+
+def _doc_dir(policy_id: str) -> Path:
+    # policy_id is already a tight slug (see backend.main.upload_policy:
+    # user-upload__<sid12>__<fileslug>) but defend against path traversal.
+    safe = re.sub(r"[^a-zA-Z0-9_.\-]+", "-", policy_id).strip("-") or "user-upload"
+    return uploaded_docs_dir() / safe
+
+
+# ---------------------------------------------------------------------------
+# Heuristic field extraction  ->  curated-facts-shaped record
+#
+# The repo's LLM extractor (rag/extract.py) needs network + the NIM brain.
+# That is correct for the corpus build but unusable inside a request (and
+# untestable offline). So we derive a REAL, sourced record deterministically
+# from the PDF's own text via regex over the IRDAI-standardised wording that
+# every Indian health policy uses. Each field we emit carries the verbatim
+# source_quote it was matched from — nothing is fabricated; a field is only
+# emitted when its evidence is literally present in the document.
+#
+# This populates well above the scorecard's MIN_GRADEABLE_COMPLETENESS_PCT
+# (9.0 == ~2 of 23 SCORED_FIELDS) so the card grades for real instead of
+# returning the data-starved "—"/0 sentinel. When the document genuinely
+# lacks structured terms, we DO NOT invent any — the card then honestly
+# shows the sentinel, which is the correct behaviour.
+# ---------------------------------------------------------------------------
+
+
+def _ctx(text: str, m: re.Match, pad: int = 90) -> str:
+    """Verbatim surrounding snippet for a regex match (the source_quote)."""
+    s = max(0, m.start() - pad)
+    e = min(len(text), m.end() + pad)
+    return re.sub(r"\s+", " ", text[s:e]).strip()[:300]
+
+
+def _fact(value: Any, quote: str, conf: str = "medium") -> dict:
+    """A curated-facts cell: {value, source_pdf_path, source_quote, _confidence}."""
+    return {
+        "value": value,
+        "source_pdf_path": "",  # filled by the caller with the persisted PDF path
+        "source_quote": quote,
+        "_confidence": conf,
+    }
+
+
+def extract_fields_from_text(full_text: str) -> dict[str, dict]:
+    """Regex-derive scorecard-relevant fields from policy text.
+
+    Returns a {field_name: <fact cell>} dict using the SAME canonical field
+    names backend.scorecard.SCORED_FIELDS / ALIASES read. Only fields with
+    literal textual evidence are emitted. Never raises (a totally
+    unparseable doc just yields {}).
+    """
+    t = full_text or ""
+    low = t.lower()
+    out: dict[str, dict] = {}
+
+    def add(field: str, value: Any, m: Optional[re.Match], conf: str = "medium"):
+        if value is None:
+            return
+        if field in out:
+            return
+        quote = _ctx(t, m) if m is not None else ""
+        out[field] = _fact(value, quote, conf)
+
+    # --- UIN (regulator identity; not a scored field but anchors the card) --
+    m = re.search(r"\b([A-Z]{3}[A-Z0-9]{10,22}V\d{6})\b", t)
+    if m:
+        add("uin_code", m.group(1), m, "high")
+
+    # --- Initial waiting period (days) -------------------------------------
+    m = re.search(
+        r"(\d{1,3})\s*days?[^.]{0,80}?(?:waiting period|from the (?:first )?"
+        r"(?:policy )?(?:commencement|inception)|shall be excluded)",
+        t, re.IGNORECASE,
+    ) or re.search(
+        r"(?:waiting period|initial waiting)[^.]{0,60}?(\d{1,3})\s*days?",
+        t, re.IGNORECASE,
+    )
+    if m:
+        d = int(m.group(1))
+        if 0 < d <= 90:
+            add("initial_waiting_period_days", d, m, "high")
+
+    # --- Pre-existing disease waiting (months) -----------------------------
+    m = re.search(
+        r"pre[\-\s]?existing[^.]{0,120}?(\d{1,2})\s*(?:months|month)",
+        t, re.IGNORECASE,
+    ) or re.search(
+        r"(\d{1,2})\s*months[^.]{0,80}?pre[\-\s]?existing",
+        t, re.IGNORECASE,
+    )
+    if m:
+        mo = int(m.group(1))
+        if 0 < mo <= 72:
+            add("pre_existing_disease_waiting_months", mo, m, "high")
+
+    # --- Specific-disease waiting (months) ---------------------------------
+    m = re.search(
+        r"(?:specific (?:disease|illness)|cataract|hernia)[^.]{0,120}?"
+        r"(\d{1,2})\s*months",
+        t, re.IGNORECASE,
+    )
+    if m:
+        mo = int(m.group(1))
+        if 0 < mo <= 48:
+            add("specific_disease_waiting_months", mo, m, "medium")
+
+    # --- Maternity waiting (months) ----------------------------------------
+    m = re.search(
+        r"maternity[^.]{0,120}?(\d{1,2})\s*months",
+        t, re.IGNORECASE,
+    )
+    if m:
+        mo = int(m.group(1))
+        if 0 < mo <= 48:
+            add("maternity_waiting_months", mo, m, "medium")
+
+    # --- Pre / post hospitalisation (days) ---------------------------------
+    m = re.search(r"pre[\-\s]?hospitali[sz]ation[^.]{0,60}?(\d{1,3})\s*days", t, re.IGNORECASE)
+    if m:
+        d = int(m.group(1))
+        if 0 < d <= 180:
+            add("pre_hospitalization_days", d, m, "high")
+    m = re.search(r"post[\-\s]?hospitali[sz]ation[^.]{0,60}?(\d{1,3})\s*days", t, re.IGNORECASE)
+    if m:
+        d = int(m.group(1))
+        if 0 < d <= 365:
+            add("post_hospitalization_days", d, m, "high")
+
+    # --- Co-payment (%) ----------------------------------------------------
+    m = re.search(r"co[\-\s]?pay(?:ment)?[^.]{0,80}?(\d{1,2})\s*%", t, re.IGNORECASE) \
+        or re.search(r"(\d{1,2})\s*%[^.]{0,40}?co[\-\s]?pay", t, re.IGNORECASE)
+    if m:
+        pct = int(m.group(1))
+        if 0 <= pct <= 50:
+            add("copayment_pct", pct, m, "medium")
+
+    # --- No-claim bonus (%) ------------------------------------------------
+    m = re.search(
+        r"(?:no[\-\s]?claim bonus|cumulative bonus|ncb)[^.]{0,80}?(\d{1,3})\s*%",
+        t, re.IGNORECASE,
+    )
+    if m:
+        pct = int(m.group(1))
+        if 0 < pct <= 200:
+            # MarketplacePolicy.no_claim_bonus_pct is Optional[int]; the
+            # scorecard reads it numerically either way. Emit int.
+            add("no_claim_bonus_pct", pct, m, "medium")
+
+    # --- Room rent capping -------------------------------------------------
+    m = re.search(
+        r"room rent[^.]{0,90}?(no (?:sub[\-\s]?limit|cap|capping|limit)|"
+        r"\d{1,2}\s*%\s*(?:of\s*(?:the\s*)?sum insured|of si)?|single private|"
+        r"twin sharing|shared accommodation)",
+        t, re.IGNORECASE,
+    )
+    if m:
+        cap = m.group(1).strip()
+        if re.search(r"no (sub[\-\s]?limit|cap|capping|limit)", cap, re.IGNORECASE):
+            cap = "No room rent cap"
+        add("room_rent_capping", cap, m, "medium")
+
+    # --- Network hospital count -------------------------------------------
+    m = re.search(
+        r"([\d,]{3,7})\+?\s*(?:network |empanelled |cashless )?hospitals?",
+        t, re.IGNORECASE,
+    )
+    if m:
+        try:
+            n = int(m.group(1).replace(",", ""))
+            if 50 <= n <= 50000:
+                add("network_hospital_count", n, m, "medium")
+        except ValueError:
+            pass
+
+    # --- Cashless supported -----------------------------------------------
+    if "cashless" in low:
+        m = re.search(r"cashless[^.]{0,80}", t, re.IGNORECASE)
+        add("cashless_treatment_supported", True, m, "medium")
+
+    # --- Max entry age (years) --------------------------------------------
+    m = re.search(
+        r"(?:maximum |max\.? )?entry age[^.]{0,40}?(\d{2,3})\s*years",
+        t, re.IGNORECASE,
+    ) or re.search(
+        r"entry age[^.]{0,40}?up to\s*(\d{2,3})\s*years", t, re.IGNORECASE,
+    )
+    if m:
+        age = int(m.group(1))
+        if 30 <= age <= 100:
+            add("max_entry_age", age, m, "medium")
+
+    # --- AYUSH coverage ----------------------------------------------------
+    if re.search(r"\bayush\b", low) or "ayurved" in low:
+        m = re.search(r"ayush[^.]{0,90}", t, re.IGNORECASE) or re.search(
+            r"ayurved[^.]{0,90}", t, re.IGNORECASE)
+        add("ayush_coverage", {"covered": True}, m, "medium")
+
+    # --- Maternity coverage (boolean-with-detail) -------------------------
+    if "maternity" in low:
+        m = re.search(r"maternity[^.]{0,120}", t, re.IGNORECASE)
+        covered = not bool(re.search(
+            r"maternity[^.]{0,40}(not covered|excluded|no cover)", t, re.IGNORECASE))
+        add("maternity_coverage", {"covered": covered}, m, "medium")
+
+    # --- Ambulance / day-care / restoration (presence booleans) -----------
+    if "ambulance" in low:
+        m = re.search(r"ambulance[^.]{0,90}", t, re.IGNORECASE)
+        add("ambulance_cover", {"covered": True}, m, "low")
+    if "day care" in low or "day-care" in low or "daycare" in low:
+        m = re.search(r"day[\-\s]?care[^.]{0,90}", t, re.IGNORECASE)
+        add("day_care_treatments_count", {"covered": True, "limit_text": "Day-care procedures covered"}, m, "low")
+    if "restoration" in low or "refill" in low or "reinstatement" in low:
+        m = re.search(r"(restoration|refill|reinstatement)[^.]{0,90}", t, re.IGNORECASE)
+        add("restoration_benefit", {"covered": True}, m, "low")
+
+    # --- Claim settlement ratio (insurer-level; commonly stated in CIS) ----
+    m = re.search(
+        r"claim settlement ratio[^.]{0,40}?(\d{2,3}(?:\.\d{1,2})?)\s*%",
+        t, re.IGNORECASE,
+    )
+    if m:
+        try:
+            csr = float(m.group(1))
+            if 30 <= csr <= 100:
+                add("claim_settlement_ratio", csr, m, "medium")
+        except ValueError:
+            pass
+
+    return out
+
+
+def _derive_policy_name(full_text: str, fallback: str) -> str:
+    """Best-effort human policy name from the document header."""
+    for line in (full_text or "").splitlines():
+        s = line.strip()
+        if not s:
+            continue
+        if re.search(r"(policy|plan|insurance|mediclaim|health)", s, re.IGNORECASE) \
+                and 6 <= len(s) <= 90:
+            return re.sub(r"\s+", " ", s)
+    return fallback
+
+
+# ---------------------------------------------------------------------------
+# Persisted record  (curated-facts JSON)  +  PDF  +  chunk payload
+# ---------------------------------------------------------------------------
+
+
+def build_record(
+    policy_id: str,
+    policy_name: str,
+    full_text: str,
+    persisted_pdf_path: str,
+) -> dict:
+    """Build the curated-facts-shaped JSON the marketplace Pass-2 consumes.
+
+    The returned dict is the EXACT shape `_load_curated_facts._flatten`
+    expects: scalar identity keys + per-field `{value, source_*}` cells.
+    """
+    fields = extract_fields_from_text(full_text)
+    rel_pdf = persisted_pdf_path
+    for cell in fields.values():
+        if isinstance(cell, dict) and "source_pdf_path" in cell:
+            cell["source_pdf_path"] = rel_pdf
+
+    record: dict[str, Any] = {
+        "policy_id": policy_id,
+        "policy_name": policy_name or _derive_policy_name(full_text, policy_id),
+        "insurer_slug": UPLOAD_INSURER_SLUG,
+        "_uploaded_doc": True,  # provenance flag (ignored by scorecard)
+    }
+    record.update(fields)
+    return record
+
+
+def persist_upload(
+    *,
+    policy_id: str,
+    policy_name: str,
+    pdf_bytes: bytes,
+    full_text: str,
+    chunks: list[dict],
+    session_id: str,
+) -> dict:
+    """Atomically persist the PDF + JSON record + chunk payload + meta.
+
+    Returns the built record dict. Raises RuntimeError on any failure (the
+    caller MUST surface this — a "successful" upload that didn't persist is
+    a silent failure and is forbidden by the #52 spec).
+    """
+    try:
+        ddir = _doc_dir(policy_id)
+        ddir.mkdir(parents=True, exist_ok=True)
+
+        pdf_path = ddir / "source.pdf"
+        pdf_path.write_bytes(pdf_bytes)
+
+        record = build_record(
+            policy_id, policy_name, full_text, persisted_pdf_path=str(pdf_path),
+        )
+
+        # Write to temp files then os.replace for crash-atomic visibility.
+        rec_tmp = ddir / "record.json.tmp"
+        rec_tmp.write_text(json.dumps(record, indent=2, ensure_ascii=False))
+        rec_tmp.replace(ddir / "record.json")
+
+        chunk_payload = [
+            {
+                "chunk_idx": c["chunk_idx"],
+                "text": c["text"],
+                "page_start": c["page_start"],
+                "page_end": c["page_end"],
+            }
+            for c in chunks
+        ]
+        ch_tmp = ddir / "chunks.json.tmp"
+        ch_tmp.write_text(json.dumps(chunk_payload, ensure_ascii=False))
+        ch_tmp.replace(ddir / "chunks.json")
+
+        meta = {
+            "policy_id": policy_id,
+            "policy_name": record["policy_name"],
+            "insurer_slug": UPLOAD_INSURER_SLUG,
+            "sha256": hashlib.sha256(pdf_bytes).hexdigest(),
+            "uploaded_at": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
+            "session_id": session_id,  # audit only — NEVER a visibility gate
+            "n_chunks": len(chunk_payload),
+        }
+        meta_tmp = ddir / "meta.json.tmp"
+        meta_tmp.write_text(json.dumps(meta, indent=2))
+        meta_tmp.replace(ddir / "meta.json")
+
+        _log.info(
+            "persisted uploaded doc %s (%d fields, %d chunks) -> %s",
+            policy_id, len([k for k in record if not k.startswith(("policy_", "insurer_", "_"))]),
+            len(chunk_payload), ddir,
+        )
+        return record
+    except Exception as e:  # noqa: BLE001 — convert to a loud typed failure
+        raise RuntimeError(
+            f"persist_upload failed for {policy_id}: {type(e).__name__}: {e}"
+        ) from e
+
+
+# ---------------------------------------------------------------------------
+# Read side — used by _load_curated_facts (cards) + startup re-ingest (chunks)
+# ---------------------------------------------------------------------------
+
+
+def load_persisted_records() -> dict[str, dict]:
+    """{policy_id: curated-facts-shaped record} for every persisted upload.
+
+    Consumed by backend.main._load_curated_facts so each uploaded doc
+    surfaces as a marketplace card via the EXISTING Pass-2 + build_scorecard
+    path. A single corrupt record is skipped (logged) — it must not take
+    down the whole catalogue.
+    """
+    out: dict[str, dict] = {}
+    root = settings.UPLOADED_DOCS_DIR
+    if not root.exists():
+        return out
+    for d in sorted(root.iterdir()):
+        if not d.is_dir():
+            continue
+        rec_path = d / "record.json"
+        if not rec_path.exists():
+            continue
+        try:
+            rec = json.loads(rec_path.read_text())
+            pid = rec.get("policy_id") or d.name
+            out[pid] = rec
+        except Exception as e:  # noqa: BLE001
+            _log.warning(
+                "skipping corrupt uploaded record %s: %s: %s",
+                rec_path, type(e).__name__, e,
+            )
+            continue
+    return out
+
+
+def iter_persisted_chunks():
+    """Yield (policy_id, policy_name, [chunk dicts]) for every persisted doc.
+
+    Used by the startup re-ingest to rebuild the uploaded docs' vectors in
+    the working Chroma `policies` collection after a Space restart wiped the
+    ephemeral rag/vectors snapshot.
+    """
+    root = settings.UPLOADED_DOCS_DIR
+    if not root.exists():
+        return
+    for d in sorted(root.iterdir()):
+        if not d.is_dir():
+            continue
+        ch_path = d / "chunks.json"
+        meta_path = d / "meta.json"
+        if not (ch_path.exists() and meta_path.exists()):
+            continue
+        try:
+            meta = json.loads(meta_path.read_text())
+            chunks = json.loads(ch_path.read_text())
+        except Exception as e:  # noqa: BLE001
+            _log.warning(
+                "skipping unreadable persisted chunks %s: %s: %s",
+                ch_path, type(e).__name__, e,
+            )
+            continue
+        yield (
+            meta.get("policy_id") or d.name,
+            meta.get("policy_name") or d.name,
+            chunks,
+        )
+
+
+async def reingest_persisted_into_policies() -> dict:
+    """Re-embed every persisted uploaded doc's chunks into the working
+    Chroma `policies` collection (idempotent: deletes the doc's prior
+    chunks first, keyed by policy_id).
+
+    Globally visible by design (#52: uploaded doc is added to THE
+    marketplace). Returns a small summary dict. Raises only if Chroma /
+    embedder are completely unavailable; a single bad doc is logged and
+    skipped so one corrupt upload can't block boot.
+    """
+    from rag.ingest import get_chroma_collection
+    from backend.providers.local_embeddings import LocalEmbeddings
+
+    docs = list(iter_persisted_chunks())
+    summary = {"docs": 0, "chunks": 0, "skipped": 0}
+    if not docs:
+        return summary
+
+    collection = get_chroma_collection()
+    embedder = LocalEmbeddings()
+
+    for policy_id, policy_name, chunks in docs:
+        if not chunks:
+            summary["skipped"] += 1
+            continue
+        try:
+            texts = [c["text"] for c in chunks]
+            vectors = await embedder.embed(texts, input_type="document")
+            ids = [f"{policy_id}::chunk{c['chunk_idx']}" for c in chunks]
+            metadatas = [
+                {
+                    "policy_id": policy_id,
+                    "insurer_slug": UPLOAD_INSURER_SLUG,
+                    "policy_name": policy_name,
+                    "doc_type": UPLOAD_DOC_TYPE,
+                    "source_url": "",
+                    "page_start": c["page_start"],
+                    "page_end": c["page_end"],
+                    "chunk_idx": c["chunk_idx"],
+                    # NOTE: no session_id — these are GLOBAL marketplace
+                    # chunks by design, not session-private quarantine.
+                }
+                for c in chunks
+            ]
+            try:
+                collection.delete(where={"policy_id": policy_id})
+            except Exception:  # noqa: BLE001 — first-ever ingest has nothing to delete
+                pass
+            collection.add(
+                ids=ids, documents=texts, embeddings=vectors, metadatas=metadatas,
+            )
+            summary["docs"] += 1
+            summary["chunks"] += len(chunks)
+            _log.info(
+                "re-ingested uploaded doc %s (%d chunks) into policies",
+                policy_id, len(chunks),
+            )
+        except Exception as e:  # noqa: BLE001 — one bad doc must not block boot
+            summary["skipped"] += 1
+            _log.warning(
+                "startup re-ingest skipped %s: %s: %s",
+                policy_id, type(e).__name__, e,
+            )
+    return summary

entrypoint.sh

@@ -40,6 +40,18 @@ if [ -d "/data" ] && [ -w "/data" ]; then
     fi
     rm -f /app/rag/policies.duckdb
     ln -sf /data/policies.duckdb /app/rag/policies.duckdb
+
+    # #52 — PERSIST user-uploaded policy docs across Space rebuilds.
+    #
+    # Unlike rag/vectors (intentionally ephemeral — KI-119), an uploaded
+    # policy that became a marketplace card MUST survive a restart. We point
+    # backend.config.settings.UPLOADED_DOCS_DIR at the persistent /data disk;
+    # the FastAPI startup handler (_startup_reingest_uploaded_docs) re-embeds
+    # the persisted chunks into the fresh Chroma snapshot on boot, and
+    # _load_curated_facts merges the persisted JSON records so the cards
+    # reappear. Locally (no /data) the same code uses 40-data/uploaded_docs.
+    export UPLOADED_DOCS_DIR="/data/uploaded_docs"
+    mkdir -p /data/uploaded_docs
     # Vectors stay at /app/rag/vectors — read from the fresh dataset
     # snapshot. The previous /data/vectors symlink is intentionally removed.
     if [ -L "/app/rag/vectors" ]; then

rag/retrieve.py

@@ -460,6 +460,43 @@ async def retrieve(
             # Review boost is additive; failure shouldn't kill the main result
             pass
 
+    # #52 — uploaded-doc global merge pass. A user-uploaded policy that was
+    # added to THE marketplace (doc_type='user_upload', NO session_id — it
+    # is globally visible by design) is usually a 1–few-chunk document, so
+    # it loses the raw-cosine race against the 140+ multi-chunk corpus
+    # policies and never enters the top-k even when it's the best answer to
+    # a question literally about that document. Mirror the regulatory /
+    # review boost passes: run a SECOND query restricted to
+    # doc_type='user_upload', score-boost the hits, and merge. Skipped when
+    # the caller already scoped to specific policies/insurers (then they're
+    # asking about a known corpus policy, not browsing uploaded docs). The
+    # session-scoped quarantine pass above is unaffected — that path serves
+    # the uploader's OWN still-private upload; this serves docs already
+    # promoted to the public marketplace.
+    if not policy_ids and not insurer_slugs:
+        try:
+            up_res = collection.query(
+                query_embeddings=[query_vec],
+                n_results=3,
+                where={"doc_type": "user_upload"},
+            )
+            if up_res["ids"] and up_res["ids"][0]:
+                seen = {c.chunk_id for c in out}
+                up_chunks: list[RetrievedChunk] = []
+                for cid, doc, meta, dist in zip(
+                    up_res["ids"][0], up_res["documents"][0],
+                    up_res["metadatas"][0], up_res["distances"][0],
+                ):
+                    if cid in seen:
+                        continue
+                    boosted = (1.0 - dist) * 1.1
+                    up_chunks.append(_build_chunk(cid, doc, meta, boosted))
+                merged = sorted(out + up_chunks, key=lambda c: c.score, reverse=True)
+                out = merged[:top_k]
+        except Exception:
+            # Uploaded-doc boost is additive; failure must not break retrieval
+            pass
+
     # KI-034 — populate cache with the FINAL merged result so subsequent
     # identical queries skip Voyage embed + Chroma query + boost passes.
     _cache_set(cache_key, out)

tests/test_pdf_upload_to_marketplace_e2e.py

@@ -0,0 +1,317 @@
+"""#52 — graded server-assignment for user-uploaded PDFs (E2E regression).
+
+WHAT #52 GUARANTEES (and this test pins, end-to-end, fully offline)
+------------------------------------------------------------------
+When a user uploads a policy PDF it must:
+  1. be accepted (POST /api/upload-policy, 8 security gates),
+  2. be chunked + bge-small-embedded into the SAME Chroma store the chat
+     retrieval reads (the global `policies` collection),
+  3. CREATE a real curated-facts-shaped JSON record (not the data-starved
+     "—"/0 sentinel) under the PERSISTENT uploaded-docs store,
+  4. appear in the marketplace (/api/policies/all) as a card whose grade
+     flows through backend.main.marketplace_grade — the #40 single source
+     of truth (NO re-implemented grading),
+  5. be retrievable when a user asks about it,
+  6. SURVIVE a restart: re-running the startup re-ingest handler must
+     restore its chunks AND keep the card graded.
+
+NETWORK / LLM
+-------------
+The whole upload+persist+index+grade+retrieve path is intrinsically
+offline: the security gates are byte/text scans, embeddings are LOCAL
+bge-small (no Voyage), grading is pure-Python build_scorecard, retrieval
+is Chroma cosine. So no stub is needed for steps 1-6. The final NL
+synthesis (Gemini) is the ONLY networked hop and is intentionally NOT
+exercised here — the repo's pattern is to assert the tool layer the brain
+calls returns the answer-bearing chunk (which IS the load-bearing proof).
+
+ISOLATION
+---------
+The test points settings.UPLOADED_DOCS_DIR at a tmp dir and uses a
+throwaway Chroma collection name so it never mutates the real persistent
+store or the 148-policy corpus index. The scorecard-parity (#40) guard is
+re-asserted in tests/test_scorecard_parity.py — this file additionally
+proves the uploaded card resolves through marketplace_grade identically.
+"""
+
+from __future__ import annotations
+
+import asyncio
+import importlib
+import sys
+from pathlib import Path
+
+import pytest
+
+_REPO = Path(__file__).resolve().parents[1]
+for _p in (str(_REPO), str(_REPO / "backend")):
+    if _p not in sys.path:
+        sys.path.insert(0, _p)
+
+
+def _run(coro):
+    """Run a coroutine on a FRESH event loop.
+
+    Other suites in the gate close / consume the process-default asyncio
+    loop, so `asyncio.get_event_loop()` raises `RuntimeError: There is no
+    current event loop` when this test runs after them. A dedicated
+    new_event_loop() per call is robust regardless of prior-test state.
+    """
+    loop = asyncio.new_event_loop()
+    try:
+        asyncio.set_event_loop(loop)
+        return loop.run_until_complete(coro)
+    finally:
+        loop.close()
+        asyncio.set_event_loop(None)
+
+
+# --- a structurally valid, multi-page, keyword-dense insurance PDF that
+#     pdfplumber actually extracts (insert_text, not insert_textbox) -------
+
+_PARAS = [
+    "ZENITH SECURE HEALTH INSURANCE POLICY. Unique Identification No: "
+    "ZENHLIP24077V012324. This health insurance policy covers in-patient "
+    "hospitalisation, sum insured, premium and claim.",
+    "The initial waiting period is 30 days from the first policy commencement "
+    "date; expenses within 30 days shall be excluded except accidents. "
+    "Pre-existing disease (PED) and its direct complications shall be excluded "
+    "until the expiry of 36 months of continuous coverage.",
+    "Specific disease waiting period of 24 months applies to cataract and "
+    "hernia. Maternity benefit has a waiting period of 24 months. "
+    "Pre-hospitalisation cover is 60 days and post-hospitalisation cover is "
+    "90 days.",
+    "A co-payment of 20% applies to each claim for insured persons above 60 "
+    "years. No-claim bonus (cumulative bonus) of 50% per claim-free year is "
+    "granted. Room rent: no sub-limit on room rent.",
+    "The insurer has 12,000 network hospitals for cashless treatment across "
+    "India. Maximum entry age is 65 years. AYUSH treatment is covered. "
+    "Ambulance cover is included. Day care procedures are covered.",
+    "The IRDAI claim settlement ratio for the insurer is 97.2%. Cashless "
+    "treatment is supported at all network hospitals. Pre-existing disease, "
+    "exclusions, renewal and free-look terms are defined herein. This is a "
+    "comprehensive indemnity health insurance plan regulated by IRDAI.",
+]
+
+
+def _real_policy_pdf_bytes() -> bytes:
+    import fitz  # PyMuPDF
+
+    doc = fitz.open()
+    for _ in range(2):  # 12 pages, dense extractable text per page
+        for para in _PARAS:
+            page = doc.new_page()
+            y = 72
+            for line in [para[i:i + 90] for i in range(0, len(para), 90)]:
+                page.insert_text((56, y), line, fontsize=11, fontname="helv")
+                y += 16
+    base = doc.tobytes()
+    doc.close()
+    data = base + b"\n" + (b"%" + b"X" * 118 + b"\n") * 40 + b"%%EOF\n"
+    assert data.startswith(b"%PDF") and 5_000 < len(data) < 25 * 1024 * 1024
+    return data
+
+
+@pytest.fixture()
+def isolated_store(tmp_path, monkeypatch):
+    """Point the persistent uploaded-docs dir + the working Chroma vectors
+    at throwaway dirs so the test never touches the real store/corpus."""
+    import backend.config as cfg
+
+    udir = tmp_path / "uploaded_docs"
+    vdir = tmp_path / "vectors"
+    udir.mkdir()
+    vdir.mkdir()
+    monkeypatch.setattr(cfg.settings, "UPLOADED_DOCS_DIR", udir, raising=False)
+    monkeypatch.setattr(cfg.settings, "VECTORS_DIR", vdir, raising=False)
+
+    # Reload the modules that captured settings paths at import time so they
+    # see the isolated dirs. main + brain_tools read settings live, so a
+    # monkeypatch is enough; rag.retrieve/ingest build the Chroma client
+    # path from settings.VECTORS_DIR per call, so no reload needed there.
+    import backend.main as M
+    import backend.uploaded_docs as U
+    import rag.retrieve as R
+    import rag.ingest as I
+
+    # Bust caches that may hold corpus-built state.
+    M._CORPUS_PDF_IDX = None
+    with M._MG_LOCK:
+        M._MG_CACHE["sig"] = None
+        M._MG_CACHE["index"] = None
+    R._RETRIEVAL_CACHE.clear()
+    yield {"M": M, "U": U, "R": R, "I": I, "udir": udir}
+
+
+def test_upload_pdf_becomes_a_graded_persistent_marketplace_card(isolated_store):
+    M = isolated_store["M"]
+    U = isolated_store["U"]
+    R = isolated_store["R"]
+    I = isolated_store["I"]
+
+    from fastapi.testclient import TestClient
+
+    client = TestClient(M.app, raise_server_exceptions=True)
+    pdf = _real_policy_pdf_bytes()
+
+    # ---- STEP 1+2+3: accept + chunk/embed + create JSON record -----------
+    resp = client.post(
+        "/api/upload-policy",
+        files={"file": ("zenith_secure_policy.pdf", pdf, "application/pdf")},
+        data={"session_id": "pytest-e2e-sid-001"},
+    )
+    assert resp.status_code == 200, resp.text
+    body = resp.json()
+    pid = body["policy_id"]
+    assert pid.startswith("user-upload__")
+    assert body["chunks_added"] >= 1
+    assert body["pages_indexed"] >= 1
+
+    # STEP 3 — the created JSON record exists, is curated-facts-shaped, and
+    # carries real sourced fields (NOT empty → it WILL grade, not sentinel).
+    recs = U.load_persisted_records()
+    assert pid in recs, f"no persisted record for {pid}: {list(recs)}"
+    rec = recs[pid]
+    assert rec["insurer_slug"] == "user-upload"
+    sourced = [
+        k for k, v in rec.items()
+        if isinstance(v, dict) and "value" in v and v.get("value") is not None
+    ]
+    # The scorecard's sentinel fires below ~2/23 scored fields; we sourced
+    # well above that from this document's own text.
+    assert len(sourced) >= 6, f"too few sourced fields ({sourced})"
+    # Every emitted field must be backed by a verbatim source quote (the
+    # #52 'no fabrication' contract — a field only exists if its evidence
+    # is literally in the document).
+    for k in sourced:
+        assert rec[k].get("source_quote"), f"{k} has no source_quote"
+    # Persisted artefacts (PDF + chunks + meta) for restart survival.
+    ddir = isolated_store["udir"] / pid
+    for fn in ("source.pdf", "record.json", "chunks.json", "meta.json"):
+        assert (ddir / fn).is_file(), f"missing persisted {fn}"
+
+    # ---- STEP 4: marketplace card + graded via the #40 SSOT -------------
+    feed = client.get("/api/policies/all").json()
+    cards = [p for p in feed["policies"] if p["policy_id"] == pid]
+    assert len(cards) == 1, "uploaded doc not in /api/policies/all exactly once"
+    card = cards[0]
+    assert card["grade"] in ("A", "B", "C", "D", "F"), card["grade"]
+    assert card["grade"] != "—" and card["overall_score"] > 0, card
+    assert card["data_completeness_pct"] >= 9.0  # above the sentinel floor
+
+    # The card grade MUST come from backend.main.marketplace_grade (the #40
+    # single source of truth) and the recommendation-path signal MUST agree
+    # — no parallel grading implementation.
+    mg = M.marketplace_grade(pid)
+    assert mg.get("_grade") == card["grade"], (mg, card["grade"])
+    assert mg.get("_overall_score") == card["overall_score"]
+    from backend.brain_tools import _scorecard_signal
+    sig = _scorecard_signal(pid)
+    assert sig.get("_grade") == card["grade"], (sig, card["grade"])
+
+    # ---- STEP 5+6: retrievable + answer-bearing for a Q&A ---------------
+    async def _ask():
+        from backend.brain_tools import retrieve_policies
+
+        R._RETRIEVAL_CACHE.clear()
+        # Anonymous GLOBAL query (no session): the doc was added to THE
+        # marketplace, so it must surface for anyone asking about it.
+        r = await retrieve_policies(
+            query="Zenith Secure policy pre-existing disease waiting period "
+                  "and co-payment percentage",
+            top_k=10,
+            intent="qa",
+        )
+        return r
+
+    r = _run(_ask())
+    up = [c for c in (r.get("chunks") or [])
+          if str(c.get("policy_id", "")).startswith("user-upload__")]
+    assert up, f"uploaded doc not retrievable for its own Q&A: {r}"
+    answer_text = " ".join(c["chunk_text"] for c in up)
+    # The retrieved chunk literally contains the answer to the question.
+    assert "36 months" in answer_text  # PED waiting period
+    assert "20%" in answer_text        # co-payment
+
+    # ---- RESTART PERSISTENCE: simulate a Space rebuild ------------------
+    # A rebuild wipes the ephemeral Chroma. Delete the uploaded doc's chunks
+    # from the working collection, prove they're GONE, then run the startup
+    # re-ingest handler and prove steps 4-6 hold again from the PERSISTED
+    # store alone.
+    coll = I.get_chroma_collection()
+    coll.delete(where={"policy_id": pid})
+    R._RETRIEVAL_CACHE.clear()
+    gone = coll.get(where={"policy_id": pid}, limit=5)
+    assert not gone.get("ids"), "chunks should be gone post simulated rebuild"
+
+    _run(M._startup_reingest_uploaded_docs())
+
+    # STEP 4 again — card still present + still graded (from persisted JSON).
+    feed2 = client.get("/api/policies/all").json()
+    cards2 = [p for p in feed2["policies"] if p["policy_id"] == pid]
+    assert len(cards2) == 1, "card lost after restart"
+    assert cards2[0]["grade"] == card["grade"], "grade changed after restart"
+
+    # STEP 5+6 again — chunks restored + retrievable from persisted payload.
+    async def _ask2():
+        from backend.brain_tools import retrieve_policies
+
+        R._RETRIEVAL_CACHE.clear()
+        return await retrieve_policies(
+            query="Zenith Secure policy co-payment percentage and "
+                  "pre-existing disease waiting period",
+            top_k=10,
+            intent="qa",
+        )
+
+    r2 = _run(_ask2())
+    up2 = [c for c in (r2.get("chunks") or [])
+           if str(c.get("policy_id", "")).startswith("user-upload__")]
+    assert up2, f"uploaded doc NOT retrievable after restart re-ingest: {r2}"
+    txt2 = " ".join(c["chunk_text"] for c in up2)
+    assert "36 months" in txt2 and "20%" in txt2, "answer lost after restart"
+
+
+def test_no_silent_failure_on_persist_error(isolated_store, monkeypatch):
+    """A persist failure MUST surface as an HTTP 500 (the #52 'no silent
+    failure' contract) — a 200 that didn't persist is forbidden."""
+    M = isolated_store["M"]
+    from fastapi.testclient import TestClient
+    import backend.uploaded_docs as U
+
+    def _boom(**_kw):
+        raise RuntimeError("persist_upload simulated disk failure")
+
+    monkeypatch.setattr(U, "persist_upload", _boom)
+    client = TestClient(M.app, raise_server_exceptions=False)
+    resp = client.post(
+        "/api/upload-policy",
+        files={"file": ("zenith_secure_policy.pdf",
+                        _real_policy_pdf_bytes(), "application/pdf")},
+        data={"session_id": "pytest-e2e-sid-fail"},
+    )
+    assert resp.status_code == 500, resp.text
+    assert "Indexing failed" in resp.text
+    # The PDF must NOT be left orphaned in rag/corpus/user-upload on failure.
+
+
+def test_extract_fields_emits_only_evidenced_facts():
+    """The heuristic extractor must NEVER fabricate: a doc with zero
+    structured terms yields {} (→ the honest sentinel), and every emitted
+    field carries its verbatim source quote."""
+    import backend.uploaded_docs as U
+
+    assert U.extract_fields_from_text("just some prose with no policy terms at all") == {}
+
+    fields = U.extract_fields_from_text(
+        "Pre-existing disease shall be excluded until the expiry of 48 "
+        "months. A co-payment of 15% applies. 30 days initial waiting period."
+    )
+    assert fields["pre_existing_disease_waiting_months"]["value"] == 48
+    assert fields["copayment_pct"]["value"] == 15
+    for cell in fields.values():
+        assert cell["source_quote"], "every fact must be source-quoted"
+
+
+if __name__ == "__main__":
+    sys.exit(pytest.main([__file__, "-q"]))