Presidio fail-loud: surface NER degradation instead of silent regex fallback

bin/ghost

@@ -71,6 +71,8 @@ else
 fi
 [ -n "$PASS_PATHS" ] && STATUS="$STATUS · 🗂️ path-aware (real paths visible to hosted model)"
 [ -f "$NOSCRUB_MARK" ] && STATUS="$STATUS · 🔓 PII redaction OFF (secrets still scrubbed)"
+# NER scrubber expected but failed to load -> regex fallback (names may not be scrubbed). Loud.
+[ -f "__GHOST_HOME__/privacy/.presidio_failed" ] && STATUS="$STATUS · ⚠️ NER scrubber OFF (regex fallback -- reinstall or check Presidio)"
 echo "$STATUS" >&2
 
 HB="__ENG__/venv/bin/hermes"

privacy/scrubbing_proxy.py

@@ -49,6 +49,9 @@
 # .presidio marker; falls back hard to the legacy regex scrubber on any import/runtime
 # error so the bridge never goes down over a scrubber problem.
 PRESIDIO_MARKER = os.path.expanduser("~/.ghost/privacy/.presidio")
+# Written when NER is EXPECTED (.presidio set) but unavailable/failed, so the degradation to the
+# weaker regex scrubber is visible (bin/ghost surfaces it) instead of silent. Cleared on success.
+PRESIDIO_FAILED_MARKER = os.path.expanduser("~/.ghost/privacy/.presidio_failed")
 try:
     import presidio_scrub
     _PRESIDIO_OK = True
@@ -517,16 +520,44 @@ def _relay_stream(self, obj):
                 raise
 
 
+def _mark_presidio_failed(reason):
+    """Record that NER was expected but isn't working, so bin/ghost can surface it loudly."""
+    log(f"!! NER scrubber UNAVAILABLE -- falling back to regex (names may NOT be scrubbed): {reason}")
+    try:
+        with open(PRESIDIO_FAILED_MARKER, "w") as f:
+            f.write(reason + "\n")
+    except Exception:
+        pass
+
+
+def _clear_presidio_failed():
+    try:
+        os.remove(PRESIDIO_FAILED_MARKER)
+    except OSError:
+        pass
+
+
 if __name__ == "__main__":
-    # Warm spaCy now (if Presidio is enabled) so the first real request isn't slow.
-    if _PRESIDIO_OK and os.path.exists(PRESIDIO_MARKER):
-        try:
-            presidio_scrub.anonymize("warmup")
-            log("presidio warm (NER scrubbing active)")
-        except Exception as e:
-            log(f"presidio warmup failed ({e}); using legacy scrub")
+    # If NER is expected (.presidio set), prove it actually works at startup and FAIL LOUD if not
+    # -- a silent fall-through to the regex scrubber (empty denylist => no names scrubbed) is the
+    # dangerous failure mode for a privacy tool. The marker is surfaced in `ghost`'s status line.
+    expected = os.path.exists(PRESIDIO_MARKER)
+    active = False
+    if expected:
+        if not _PRESIDIO_OK:
+            _mark_presidio_failed("presidio/spacy import failed")
+        else:
+            try:
+                presidio_scrub.anonymize("warmup")
+                active = True
+                _clear_presidio_failed()
+                log("presidio warm (NER scrubbing active)")
+            except Exception as e:
+                _mark_presidio_failed(f"warmup error: {e}")
+    else:
+        _clear_presidio_failed()  # NER not expected; not a failure
     log(
         f"scrubbing bridge up on {LISTEN[0]}:{LISTEN[1]} -> og-veil {VEIL_URL}; "
-        f"{len(DENY)} denylist terms; presidio={'on' if (_PRESIDIO_OK and os.path.exists(PRESIDIO_MARKER)) else 'off'}"
+        f"{len(DENY)} denylist terms; presidio={'on' if active else ('FAILED' if expected else 'off')}"
     )
     ThreadingHTTPServer(LISTEN, Handler).serve_forever()

tests/test_scrub.py

@@ -192,6 +192,16 @@ def test_expanded_secret_coverage(secret):
     assert n >= 1
 
 
+def test_presidio_failed_marker_roundtrip(tmp_path, monkeypatch):
+    marker = tmp_path / ".presidio_failed"
+    monkeypatch.setattr(sp, "PRESIDIO_FAILED_MARKER", str(marker))
+    sp._mark_presidio_failed("import failed")
+    assert marker.exists() and "import failed" in marker.read_text()
+    sp._clear_presidio_failed()
+    assert not marker.exists()
+    sp._clear_presidio_failed()  # idempotent: clearing an absent marker must not raise
+
+
 def test_transient_detection():
     assert sp._is_transient(502, "Selected TEE is not active in the registry")
     assert sp._is_transient(500, "Stream setup failed")