feat(api): drift z-score in /predict/bbb response

- WORKER_CONFIDENCE_DEQUE: collections.deque(maxlen=100), per-worker
rolling window of confidences; drift_z computed against train-time
median when ≥10 samples buffered AND model has _neurobridge_train_stats.
- BBBPredictResponse gains drift_z (float | None) and rolling_n (int).
- 2 new tests: drift_z/rolling_n always present in body; deque rolls
at 100 after 105 predictions.

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

src/api/routes.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import os
 import time
+from collections import deque
 from pathlib import Path
 from typing import Callable
 
@@ -130,6 +131,35 @@ def _bbb_model_path() -> Path:
     return Path(os.environ.get("BBB_MODEL_PATH", str(_DEFAULT_BBB_MODEL_PATH)))
 
 
+# Per-worker rolling window of recent prediction confidences.
+# Cleared on worker restart; multi-worker setups have independent windows.
+WORKER_CONFIDENCE_DEQUE: deque[float] = deque(maxlen=100)
+_DRIFT_MIN_SAMPLES = 10
+
+
+def _compute_drift_z(model, confidence: float) -> tuple[float | None, int]:
+    """Append `confidence` to the worker deque and compute the drift z-score.
+
+    Returns (drift_z, rolling_n). drift_z is None until both:
+      (1) the deque has at least `_DRIFT_MIN_SAMPLES` samples, AND
+      (2) the model has `_neurobridge_train_stats` attached.
+
+    z = (rolling_median - train_median) / max(train_std, 1e-9)
+    """
+    import statistics
+
+    WORKER_CONFIDENCE_DEQUE.append(float(confidence))
+    rolling_n = len(WORKER_CONFIDENCE_DEQUE)
+    stats = getattr(model, "_neurobridge_train_stats", None)
+    if rolling_n < _DRIFT_MIN_SAMPLES or stats is None:
+        return None, rolling_n
+    rolling_median = statistics.median(WORKER_CONFIDENCE_DEQUE)
+    train_median = float(stats["median"])
+    train_std = max(float(stats["std"]), 1e-9)
+    drift_z = (rolling_median - train_median) / train_std
+    return float(drift_z), rolling_n
+
+
 def _matching_calibration_bin(model, confidence: float) -> CalibrationContext | None:
     """Pick the highest-threshold bin whose threshold <= confidence. None if no match or no metadata."""
     bins = getattr(model, "_neurobridge_calibration", None)
@@ -180,12 +210,15 @@ def predict_bbb(req: BBBPredictRequest) -> BBBPredictResponse:
 
     label_text = "permeable" if pred["label"] == 1 else "non-permeable"
     calibration = _matching_calibration_bin(model, pred["confidence"])
+    drift_z, rolling_n = _compute_drift_z(model, pred["confidence"])
     return BBBPredictResponse(
         label=pred["label"],
         label_text=label_text,
         confidence=pred["confidence"],
         top_features=[FeatureAttribution(**a) for a in attributions],
         calibration=calibration,
+        drift_z=drift_z,
+        rolling_n=rolling_n,
     )
 
 

src/api/schemas.py

@@ -71,7 +71,7 @@ class CalibrationContext(BaseModel):
 
 
 class BBBPredictResponse(BaseModel):
-    """Decision-system payload: prediction + uncertainty + explanation."""
+    """Decision-system payload: prediction + uncertainty + explanation + drift."""
     label: int
     label_text: str = Field(..., description="'permeable' or 'non-permeable'")
     confidence: float
@@ -80,6 +80,21 @@ class BBBPredictResponse(BaseModel):
         None,
         description="Statistical context: how often the model is right when this confident on held-out data.",
     )
+    drift_z: float | None = Field(
+        None,
+        description=(
+            "Z-score of the trailing-100 confidence median against the "
+            "train-time median; None when warming up (<10 samples) or "
+            "when the model lacks _neurobridge_train_stats."
+        ),
+    )
+    rolling_n: int = Field(
+        0,
+        description=(
+            "Number of confidence samples currently buffered in the worker's "
+            "rolling window (max 100). Zero on a fresh worker."
+        ),
+    )
 
 
 class MRIDiagnosticsRequest(BaseModel):

tests/api/test_routes.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 
 from pathlib import Path
 
+import pytest
 from fastapi.testclient import TestClient
 
 from src.api.main import app
@@ -89,8 +90,14 @@ class TestBBBPredictRoute:
         bbb_model.save(model, artifact)
         return artifact
 
+    @pytest.fixture
+    def _set_bbb_model_path(self, tmp_path: Path, monkeypatch):
+        """Build a model artifact and point BBB_MODEL_PATH at it for the test."""
+        artifact = self._setup_model_artifact(tmp_path)
+        monkeypatch.setenv("BBB_MODEL_PATH", str(artifact))
+        return artifact
+
     def test_returns_200_with_prediction_and_attributions(self, tmp_path: Path, monkeypatch):
-        import pytest
         artifact = self._setup_model_artifact(tmp_path)
         monkeypatch.setenv("BBB_MODEL_PATH", str(artifact))
 
@@ -120,6 +127,39 @@ class TestBBBPredictRoute:
         assert isinstance(cal["support"], int)
         assert cal["support"] >= 0
 
+    def test_predict_response_includes_drift_z_and_rolling_n(
+        self, _set_bbb_model_path,
+    ):
+        """T1B: drift_z and rolling_n keys must always appear in the body."""
+        # Reset deque before this test so rolling_n starts deterministic.
+        from src.api import routes
+        routes.WORKER_CONFIDENCE_DEQUE.clear()
+
+        resp = client.post("/predict/bbb", json={"smiles": "CCO", "top_k": 5})
+        assert resp.status_code == 200, resp.text
+        body = resp.json()
+        assert "drift_z" in body
+        assert "rolling_n" in body
+        # First request: buffer has 1 sample (just appended), so warming up.
+        assert body["rolling_n"] == 1
+        assert body["drift_z"] is None  # <10 samples = warming up
+
+    def test_predict_deque_rolls_at_100(self, _set_bbb_model_path):
+        """T1B: after 100 predictions, deque caps at maxlen=100 (rolls)."""
+        from src.api import routes
+        routes.WORKER_CONFIDENCE_DEQUE.clear()
+        # Fire 105 calls; final rolling_n must be 100, not 105.
+        last_body = None
+        for _ in range(105):
+            resp = client.post(
+                "/predict/bbb", json={"smiles": "CCO", "top_k": 3},
+            )
+            assert resp.status_code == 200
+            last_body = resp.json()
+        assert last_body["rolling_n"] == 100
+        # By call 105, drift_z is computable (≥10 samples) — assert numeric.
+        assert isinstance(last_body["drift_z"], float)
+
     def test_returns_400_on_invalid_smiles(self, tmp_path: Path, monkeypatch):
         artifact = self._setup_model_artifact(tmp_path)
         monkeypatch.setenv("BBB_MODEL_PATH", str(artifact))