feat(upload): backfill LLM extraction for old uploads + admin re-extract endpoint

Old uploads (persisted before ADR-044's LLM-extraction pipeline was
wired, OR persisted between dfaa4d6 and 245310d when the pipeline
was crashing on the missing asyncio import) never had a
rag/extracted/<policy_id>.json written. Their cards therefore stayed
at the ~13-17% heuristic-only completeness baseline forever — looked
identical pre- and post- fix because the new pipeline never touched
them.

CHANGES
─────────────────────────────────────────────────────────────────────
backend/uploaded_docs.py
- New backfill_extractions(force=False) — iterates every persisted
upload (UPLOADED_DOCS_DIR), skips ones that already have a
rag/extracted/<id>.json (unless force=True), calls
extract_one_for_upload sequentially. Returns
{processed, skipped, failed, policies}.

backend/main.py
- New @app .on_event("startup") _startup_upload_extraction_backfill —
fires backfill as a fire-and-forget asyncio task on every
container boot. Old uploads upgrade automatically to the new
pipeline's data depth.
- New POST /api/admin/upload/reextract — admin-gated trigger for
on-demand backfill, with `?force=true` to re-extract everything.

VERIFY
─────────────────────────────────────────────────────────────────────
- py_compile + import both clean
- Live audit on next deploy

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

backend/main.py

@@ -555,6 +555,20 @@ async def _startup_quarantine_ttl_purge():
     asyncio.create_task(_quarantine_purge_loop())
 
 
+@app.on_event("startup")
+async def _startup_upload_extraction_backfill():
+    """ADR-044 (2026-05-27) — on every container boot, run LLM-assisted
+    extraction on any persisted upload that doesn't yet have a
+    `rag/extracted/<policy_id>.json` file. This upgrades old uploads
+    (persisted before the LLM-extraction pipeline was wired, OR before
+    a fix to the pipeline was deployed) to the same data depth a fresh
+    upload now gets. Idempotent: extracts that already exist are skipped.
+    Fire-and-forget so it doesn't delay app readiness.
+    """
+    from backend import uploaded_docs as _udocs
+    asyncio.create_task(_udocs.backfill_extractions(force=False))
+
+
 @app.on_event("startup")
 async def _startup_single_brain_warmup():
     """Pre-warm the Gemini single-brain connection so the FIRST /api/chat turn
@@ -2067,6 +2081,26 @@ class ExtractionStatusResponse(BaseModel):
     error: Optional[str] = None
 
 
+@app.post("/api/admin/upload/reextract")
+async def admin_reextract_uploads(
+    request: Request,
+    x_admin_password: Optional[str] = Header(default=None, alias="X-Admin-Password"),
+    force: bool = False,
+):
+    """Run LLM-assisted extraction on every persisted upload that doesn't
+    yet have a `rag/extracted/<id>.json` (or `force=true` to re-extract all).
+    Admin-gated; fires synchronously so the response carries the summary.
+
+    Use when an upload was persisted before the LLM-extraction pipeline was
+    wired and needs to be upgraded without re-uploading.
+    """
+    from backend.admin import _check_admin
+    _check_admin(request, x_admin_password)
+    from backend import uploaded_docs as _udocs
+    summary = await _udocs.backfill_extractions(force=force)
+    return summary
+
+
 @app.get(
     "/api/upload/extraction-status/{policy_id}",
     response_model=ExtractionStatusResponse,

backend/uploaded_docs.py

@@ -957,8 +957,63 @@ async def extract_one_for_upload(
             )
         except Exception:
             pass
+        # fall through to the existing _log.warning that follows
         _log.warning(
             "[upload-extract] unexpected failure for %s: %s: %s",
             policy_id, type(e).__name__, str(e)[:400],
         )
         return False
+
+
+async def backfill_extractions(*, force: bool = False) -> dict:
+    """Run LLM-assisted extraction for every persisted upload that doesn't
+    yet have a corresponding `rag/extracted/<policy_id>.json` (or force=True
+    to re-extract every upload). Fires sequentially so we don't fan-out the
+    LLM chain. Returns a {processed, skipped, failed} summary.
+
+    Designed to be called once at server startup (to upgrade old uploads
+    that were persisted before the LLM-extraction pipeline was wired) AND
+    as the backing for POST /api/admin/upload/reextract.
+    """
+    from backend.config import settings as _settings
+    summary: dict = {"processed": 0, "skipped": 0, "failed": 0, "policies": []}
+    records = load_persisted_records()
+    for policy_id, record in records.items():
+        try:
+            out_json = _settings.EXTRACTED_DIR / f"{policy_id}.json"
+            if out_json.exists() and not force:
+                summary["skipped"] += 1
+                continue
+            pdf_path = _doc_dir(policy_id) / "source.pdf"
+            if not pdf_path.exists():
+                _log.warning(
+                    "[backfill] missing source.pdf for %s — skipping", policy_id,
+                )
+                summary["skipped"] += 1
+                continue
+            policy_name = record.get("policy_name") or policy_id
+            insurer_slug = record.get("insurer_slug") or UPLOAD_INSURER_SLUG
+            insurer_name = record.get("insurer_name") or detected_insurer_name(insurer_slug)
+            ok = await extract_one_for_upload(
+                policy_id=policy_id,
+                pdf_path=pdf_path,
+                policy_name=policy_name,
+                insurer_slug=insurer_slug,
+                insurer_name=insurer_name,
+            )
+            if ok:
+                summary["processed"] += 1
+                summary["policies"].append(policy_id)
+            else:
+                summary["failed"] += 1
+        except Exception as e:  # noqa: BLE001
+            _log.warning(
+                "[backfill] failed for %s: %s: %s",
+                policy_id, type(e).__name__, str(e)[:200],
+            )
+            summary["failed"] += 1
+    _log.info(
+        "[backfill] done: processed=%d skipped=%d failed=%d",
+        summary["processed"], summary["skipped"], summary["failed"],
+    )
+    return summary