fix(ai-service): enforce PII scrubbing on the humanitarian verification path

app/ai-service/README.md

@@ -32,6 +32,8 @@ The service starts at `http://localhost:8000`. Interactive API documentation is
 | `BACKEND_WEBHOOK_URL` | `http://localhost:3001/ai/webhook` | Backend notification endpoint |
 | `MAX_REQUEST_BODY_BYTES` | `10485760` (10 MiB) | Maximum HTTP request body size; oversized requests are rejected with HTTP 413 to prevent memory-exhaustion DoS. Set to `0` to disable (not recommended in production). |
 | `REQUEST_BODY_BYPASS_PATHS` | _(empty)_ | Comma-separated path entries that bypass body-size limiting. Entries without a trailing `'/'` must match the path exactly; entries with a trailing `'/'` (e.g. `/hooks/`) match any path with that prefix. The default bypass list (`/health`, `/`, `/ai/metrics`, `/docs`, `/redoc`, `/openapi.json`) is always merged in. |
+| `PII_SCRUBBING_ENABLED` | `true` | Master switch for the mandatory PII preprocessing stage on `/v1/ai/humanitarian/verify`. When disabled the service **fails closed**: verification requests are rejected rather than sending unredacted evidence to an external LLM provider. |
+| `PII_DECISIONS_ENABLED` | `false` | When truthy, every successful scrubbing event (the `/v1/ai/anonymize` endpoint and the humanitarian verification pipeline) writes aggregate audit metadata (counts + fingerprint only, never text) to the PII decisions store. |
 
 ## Core services
 
@@ -116,6 +118,38 @@ Analyzes claim metadata batches using Local Outlier Factor and flags anomalous p
 
 ---
 
+## PII scrubbing posture
+
+PII anonymization is enforced as a **mandatory preprocessing stage**, not an
+opt-in helper. Before any prompt is built for an external LLM provider, the
+pipeline scrubs `aid_claim`, every `supporting_evidence` entry, and
+string-valued `context_factors` using the same detector that backs
+`/v1/ai/anonymize`.
+
+| Endpoint | Scrubs PII before external processing | Notes |
+|---|---|---|
+| `POST /v1/ai/humanitarian/verify` | ✅ Yes | Fail-closed: if scrubbing is disabled (`PII_SCRUBBING_ENABLED=false`) or fails, the request is rejected with `success=false` and no provider call is made. The response includes a `pii_scrubbing` block (`applied`, `anonymized`, `pii_summary`); the raw text is never returned and remains only on the backend (human-review) side. |
+| `POST /v1/ai/anonymize` | ✅ Yes | Explicit scrubbing endpoint; the caller receives the anonymized text and aggregate summary. |
+| `POST /ai/ocr` | ❌ No | Local document extraction (Tesseract); no external LLM call is made from this path. |
+| `POST /v1/ai/fraud/detect` | ❌ No | Local statistical anomaly detection (LOF); no external LLM call is made from this path. |
+| `POST /ai/proof-of-life` | ❌ No | Local face/liveness analysis; no external LLM call is made from this path. |
+
+**Residual risk.** Detection relies on a regex set plus a spaCy `blank`
+entity ruler, so precision is imperfect: generic capitalized phrases can be
+over-redacted, and unusual identifier formats (e.g. non-Nigerian ID numbers,
+foreign phone formats) can pass through undetected. Treat scrubbing as a
+privacy-control layer, not a guarantee; recipients should be advised that
+names and locations embedded in evidence are material to verification and
+may affect the verdict. Do not rely on scrubbed output to fully de-identify
+free text.
+
+**Audit.** When `PII_DECISIONS_ENABLED=true`, each scrubbing event writes an
+aggregate record (entity counts, token counts, a non-reversible SHA-256
+fingerprint, model version) to the SQLite `pii_decisions` store. Raw or
+anonymized text is never persisted.
+
+---
+
 ## Versioned API
 
 All routes are available under versioned and legacy paths during the transition period.

app/ai-service/config.py

@@ -54,6 +54,11 @@ class Settings(BaseSettings):
             survives before the periodic sweeper removes it. Default: 30.
         PII_DECISIONS_SWEEP_INTERVAL_SECONDS: How often the in-process
             retention sweep runs. Default: 3600 (1 hour).
+        PII_SCRUBBING_ENABLED: Master switch for the mandatory PII
+            preprocessing stage on the humanitarian verification pipeline.
+            When disabled the service fails closed: verification requests
+            are rejected rather than sending unredacted evidence to an
+            external LLM provider. Default: true.
     """
 
     # API Keys
@@ -102,6 +107,11 @@ class Settings(BaseSettings):
     # the store is opt-in for tests; the runtime behaviour of
     # /v1/ai/anonymize is unchanged when disabled.
     pii_decisions_enabled: bool = False
+
+    # PII scrubbing (Issue #430): mandatory preprocessing stage for the
+    # humanitarian verification pipeline. Fails closed when disabled so raw
+    # recipient text is never sent to an external LLM provider.
+    pii_scrubbing_enabled: bool = True
     pii_decisions_db_path: str = "./data/pii_decisions.db"
     pii_decisions_retention_days: int = 30
     pii_decisions_sweep_interval_seconds: int = 3600

app/ai-service/schemas/humanitarian.py

@@ -17,4 +17,5 @@ class HumanitarianVerificationResponse(BaseModel):
     prompt_variant: Optional[str] = None
     verification: Optional[Dict[str, Any]] = None
     error: Optional[str] = None
-    model_version: Optional[str] = None
+    model_version: Optional[str] = None
+    pii_scrubbing: Optional[Dict[str, Any]] = None

app/ai-service/schemas/humanitarian_verification_v2.py

@@ -21,4 +21,5 @@ class HumanitarianVerificationResponseV2(BaseModel):
     error: Optional[str] = None
     model_version: Optional[str] = None
     stamp: Optional[Dict[str, str]] = None
+    pii_scrubbing: Optional[Dict[str, Any]] = None
 

app/ai-service/services/humanitarian_verification.py

@@ -1,8 +1,9 @@
 """Humanitarian claim verification service with model/provider fallbacks."""
 
+import hashlib
 import json
 import logging
-from typing import Any, Dict, List, Optional
+from typing import Any, Dict, List, Optional, Tuple
 import time
 import metrics
 
@@ -11,6 +12,7 @@
 from config import settings
 from services.humanitarian_prompt import HumanitarianPromptEngine
 from services.circuit_breaker import CircuitBreaker
+from services.pii_scrubber import PIIScrubberService
 from services.test_provider import TestProvider
 from exceptions import AIServiceError
 
@@ -20,9 +22,14 @@
 class HumanitarianVerificationService:
     """Runs humanitarian verification against configured LLM providers."""
 
+    _PII_SUMMARY_KEYS = ("names", "locations", "dates", "emails", "phones", "ids", "total")
+
     def __init__(self):
         self.prompt_engine = HumanitarianPromptEngine()
         self.test_provider = TestProvider()
+        # Issue #430: the scrubber is a mandatory preprocessing stage, not an
+        # optional endpoint. Raw recipient text must never reach a provider.
+        self.scrubber = PIIScrubberService()
         self.breakers = {
             "openai": CircuitBreaker(
                 name="openai",
@@ -49,15 +56,22 @@ def verify_claim(
             evidence = supporting_evidence or []
             context = context_factors or {}
 
+            # Issue #430: mandatory PII preprocessing before any provider
+            # call. Only the scrubbed inputs ever reach prompt construction,
+            # so raw recipient text cannot leave the trust boundary.
+            scrubbed_claim, scrubbed_evidence, scrubbed_context, pii_summary, pii_token_counts = (
+                self._scrub_inputs(aid_claim, evidence, context)
+            )
+
             primary_prompt = self.prompt_engine.build_primary_prompt(
-                aid_claim=aid_claim,
-                supporting_evidence=evidence,
-                context_factors=context,
+                aid_claim=scrubbed_claim,
+                supporting_evidence=scrubbed_evidence,
+                context_factors=scrubbed_context,
             )
             fallback_prompt = self.prompt_engine.build_fallback_prompt(
-                aid_claim=aid_claim,
-                supporting_evidence=evidence,
-                context_factors=context,
+                aid_claim=scrubbed_claim,
+                supporting_evidence=scrubbed_evidence,
+                context_factors=scrubbed_context,
             )
 
             providers = self._provider_attempt_order(provider_preference)
@@ -92,6 +106,16 @@ def verify_claim(
                         parsed = parse_verification_response(provider, raw_content)
                         if breaker:
                             breaker.record_success()
+                        # Aggregate-only audit record (counts + fingerprint,
+                        # never text) so the redaction decision is traceable.
+                        self._record_pii_decision(
+                            raw_claim=aid_claim,
+                            raw_evidence=evidence,
+                            raw_context=context,
+                            summary=pii_summary,
+                            token_counts=pii_token_counts,
+                            model=model,
+                        )
                         return {
                             "provider": provider,
                             "model": model,
@@ -102,7 +126,12 @@ def verify_claim(
                                 "provider": provider,
                                 "model": model,
                                 "prompt_variant": prompt_variant,
-                            }
+                            },
+                            "pii_scrubbing": {
+                                "applied": True,
+                                "anonymized": pii_summary["total"] > 0,
+                                "pii_summary": pii_summary,
+                            },
                         }
                     except Exception as exc:
                         if breaker:
@@ -116,6 +145,133 @@ def verify_claim(
             latency = time.time() - start_time
             metrics.PIPELINE_STEP_LATENCY.labels(step_name='verify').observe(latency)
 
+    def _scrub_inputs(
+        self,
+        aid_claim: str,
+        evidence: List[str],
+        context: Dict[str, Any],
+    ) -> Tuple[str, List[str], Dict[str, Any], Dict[str, int], Dict[str, int]]:
+        """Fail-closed PII preprocessing stage for the verification pipeline.
+
+        Masks names/locations/dates/emails/phones/IDs in the claim, each
+        evidence entry, and string-valued context factors *before* any prompt
+        is built, so raw recipient text can never reach an external LLM
+        provider. Raises when scrubbing is disabled or fails rather than
+        silently forwarding unredacted text (Issue #430).
+
+        Returns the scrubbed inputs plus aggregated ``pii_summary`` and
+        ``token_counts`` for the response envelope and the audit store.
+        """
+        if not settings.pii_scrubbing_enabled:
+            raise RuntimeError(
+                "PII scrubbing is disabled (PII_SCRUBBING_ENABLED=false); "
+                "refusing to send unredacted evidence to an external LLM provider"
+            )
+
+        summary: Dict[str, int] = {key: 0 for key in self._PII_SUMMARY_KEYS}
+        token_counts: Dict[str, int] = {}
+
+        try:
+            scrubbed_claim, field_summary, field_tokens = self._scrub_field(aid_claim)
+            self._merge_scrub_stats(summary, token_counts, field_summary, field_tokens)
+
+            scrubbed_evidence: List[str] = []
+            for entry in evidence:
+                scrubbed, field_summary, field_tokens = self._scrub_field(entry)
+                scrubbed_evidence.append(scrubbed)
+                self._merge_scrub_stats(summary, token_counts, field_summary, field_tokens)
+
+            scrubbed_context: Dict[str, Any] = {}
+            for key, value in context.items():
+                if isinstance(value, str):
+                    scrubbed, field_summary, field_tokens = self._scrub_field(value)
+                    scrubbed_context[key] = scrubbed
+                    self._merge_scrub_stats(summary, token_counts, field_summary, field_tokens)
+                else:
+                    scrubbed_context[key] = value
+        except Exception as exc:
+            raise RuntimeError(
+                f"PII scrubbing failed ({exc}); refusing to send unredacted "
+                "evidence to an external LLM provider"
+            ) from exc
+
+        return scrubbed_claim, scrubbed_evidence, scrubbed_context, summary, token_counts
+
+    def _scrub_field(self, text: str) -> Tuple[str, Dict[str, int], Dict[str, int]]:
+        """Scrub one free-text field; returns (masked, summary, token_counts)."""
+        result = self.scrubber.scrub_text(text)
+        return (
+            result["anonymized_text"],
+            result["pii_summary"],
+            result["token_counts"],
+        )
+
+    def _merge_scrub_stats(
+        self,
+        summary: Dict[str, int],
+        token_counts: Dict[str, int],
+        field_summary: Dict[str, int],
+        field_tokens: Dict[str, int],
+    ) -> None:
+        """Aggregate one field's scrub summary/token counts into the totals."""
+        for key in self._PII_SUMMARY_KEYS:
+            summary[key] += field_summary.get(key, 0)
+        for token, count in (field_tokens or {}).items():
+            token_counts[token] = token_counts.get(token, 0) + count
+
+    def _record_pii_decision(
+        self,
+        raw_claim: str,
+        raw_evidence: List[str],
+        raw_context: Dict[str, Any],
+        summary: Dict[str, int],
+        token_counts: Dict[str, int],
+        model: str,
+    ) -> None:
+        """Persist aggregate scrub metadata (never text) when enabled.
+
+        Mirrors the /v1/ai/anonymize audit path: aggregate counts plus a
+        non-reversible SHA-256 fingerprint of the concatenated inputs.
+        Failures are logged, never raised, so an audit hiccup can't break a
+        verification.
+        """
+        try:
+            if not settings.pii_decisions_enabled:
+                return
+            from persistence.pii_decisions import (
+                PIIDecisionRecord,
+                PIIDecisionStore,
+                new_record_id,
+            )
+
+            raw_text = (
+                raw_claim
+                + "\n"
+                + "\n".join(raw_evidence)
+                + "\n"
+                + json.dumps(raw_context, sort_keys=True)
+            )
+            record = PIIDecisionRecord(
+                id=new_record_id(),
+                created_at=time.time(),
+                original_length=len(raw_text),
+                pii_summary=summary,
+                token_counts=token_counts,
+                text_fingerprint=hashlib.sha256(raw_text.encode("utf-8")).hexdigest(),
+                model_version=model,
+            )
+            PIIDecisionStore(settings.pii_decisions_db_path).save_decision(
+                record,
+                settings.pii_decisions_retention_days,
+            )
+            logger.info(
+                "stored pii_decision id=%s total=%d",
+                record.id,
+                summary.get("total", 0),
+            )
+        except Exception as exc:  # pragma: no cover - defensive
+            logger.error("pii_decision persistence failed: %s", exc)
+
     def _provider_attempt_order(self, provider_preference: str) -> List[str]:
         available: List[str] = []
         if settings.test_provider_mode:

app/ai-service/services/pii_scrubber.py

@@ -78,10 +78,25 @@ def __init__(self):
         metrics.PII_MODEL_VERSION.labels(version=PII_MODEL_VERSION).set(1)
 
     def anonymize(self, text: str) -> Dict[str, object]:
-        """Return privacy-preserving anonymized text and summary metadata."""
+        """Return privacy-preserving anonymized text and summary metadata.
+
+        When ``test_provider_mode`` is enabled the result is served from
+        fixture files so staging/testnet deployments can exercise the
+        contract without running the detector; otherwise the local detector
+        runs via :meth:`scrub_text`.
+        """
         if settings.test_provider_mode:
             return self.test_provider.get_response("anonymize", {"text": text})
+        return self.scrub_text(text)
+
+    def scrub_text(self, text: str) -> Dict[str, object]:
+        """Run the local PII detector and return masked text + summary metadata.
 
+        Unlike :meth:`anonymize`, this always executes the local regex and
+        spaCy detector and never routes through the test provider, so
+        hot-path consumers (e.g. the humanitarian verification pipeline)
+        get deterministic masking even when ``test_provider_mode`` is set.
+        """
         start_time = time.time()
         try:
             if not text: