Harden Servy V5 trained-model serving verification

.env.example

@@ -31,6 +31,7 @@ STARLLY_API_KEY_ID=2
 # Put the FULL best_model directory in the specified location; alternatively
 # set SERVY_RERANKER_MODEL_PATH=best_model if it is under prototype/ directly.
 # SERVY_RERANKER_ENABLED=1
+# SERVY_RERANKER_REQUIRED=1   # strict deployment: refuse startup if V5 cannot score
 # SERVY_RERANKER_MODEL_PATH=ml_models/checkpoints/servy_v5/best_model
 # SERVY_RERANKER_EXPECTED_SHA256=b1cf23ba0806f0057d73a5ab93fde88d634fc278f71b27f3a2a915afd8f8bcda
 # SERVY_RERANKER_DEVICE=cpu

V5_LOCAL_RERANKER_SETUP.md

@@ -46,6 +46,7 @@ not yet reconciled. Neither is an independent live-UI accuracy guarantee.
 
    ```ini
    SERVY_RERANKER_ENABLED=1
+   SERVY_RERANKER_REQUIRED=1
    SERVY_RERANKER_MODEL_PATH=ml_models/checkpoints/servy_v5/best_model
    SERVY_RERANKER_EXPECTED_SHA256=b1cf23ba0806f0057d73a5ab93fde88d634fc278f71b27f3a2a915afd8f8bcda
    SERVY_RERANKER_DEVICE=cpu
@@ -65,8 +66,12 @@ not yet reconciled. Neither is an independent live-UI accuracy guarantee.
    python manage.py verify_v5_reranker
    ```
 
-   Expected: `SERVY V5 RERANKER VERIFIED: ...`. This tests only local model
-   loading and one inference; it does not certify the full RAG application.
+   Expected: `SERVY V5 RERANKER VERIFIED: ...`. This executes the actual
+   serving-time reranker on one **synthetic knowledge passage**, including the
+   SHA-256-pinned offline checkpoint load, raw-logit scoring, and verified V5
+   model provenance. Missing weights, fallback, or invalid scores make the
+   command fail rather than produce a false success. It does **not** certify
+   tenant authorization, HTTP requests, benchmark accuracy, or deployment.
 
 5. Start your normal Django/MySQL and React processes. Log in and use
    **Customer AI Support** or **Engineer Copilot**. Authorized retrieved
@@ -86,6 +91,7 @@ not yet reconciled. Neither is an independent live-UI accuracy guarantee.
 ## Safety and rollout
 
 - If `SERVY_RERANKER_ENABLED=0` (the default), existing behavior is unchanged.
+- For deployment, set `SERVY_RERANKER_REQUIRED=1` **in addition to** the enabled flag, full local path, and approved digest. Django startup then runs one synthetic passage through the actual SHA-pinned V5 scorer and refuses to start if the checkpoint is absent, incompatible, or returns invalid scores. Subsequent scoring failures also propagate as request errors rather than silently returning baseline-ranked results. This opt-in strict mode is off by default. It is a model smoke check, not an independent benchmark or a live user-query test.
 - With the feature enabled, a nonexistent checkpoint or wrong weights SHA-256
   makes the reranker fail closed to **the original, already-authorized ranking**.
   It must never silently download or switch to a different remote model in

core/apps.py

@@ -9,6 +9,13 @@ class CoreConfig(AppConfig):
     def ready(self):
         from . import signals  # noqa: F401
 
+        # Production opt-in: do not report a healthy startup if the pinned
+        # V5 checkpoint cannot run the exact serving scorer.
+        from django.conf import settings
+        if getattr(settings, "SERVY_RERANKER_REQUIRED", False):
+            from core.ml.reranker import require_verified_v5_reranker
+            require_verified_v5_reranker()
+
         # Pre-warm the embedding model in background thread so the first RAG query has 0s cold start
         def _warmup():
             try:

core/management/commands/verify_v5_reranker.py

@@ -1,35 +1,65 @@
-"""Verify the configured local Servy V5 checkpoint (no DB or API calls)."""
+"""Verify that the pinned local V5 model actually runs through Servy's scorer.
+
+This is an offline, synthetic serving-path smoke check, not a production
+promotion decision, independent accuracy evaluation, or HTTP authorization test.
+"""
+import math
+from types import SimpleNamespace
+
 from django.conf import settings
 from django.core.management.base import BaseCommand, CommandError
 
-from core.ml.reranker import (SERVY_V5_WEIGHTS_SHA256, get_reranker, is_reranker_enabled, _local_checkpoint)
+from core.ml.reranker import (
+    SERVY_V5_WEIGHTS_SHA256,
+    _local_checkpoint,
+    is_reranker_enabled,
+    rerank_candidates,
+)
 
 
 class Command(BaseCommand):
-    help = "Load and score one benign query with the local V5 cross-encoder."
+    help = "Verify the pinned V5 checkpoint by executing the real reranker scoring path."
 
     def handle(self, *args, **options):
         if not is_reranker_enabled():
             raise CommandError("SERVY_RERANKER_ENABLED=1 must be set for the V5 smoke check.")
         checkpoint = _local_checkpoint()
         if checkpoint is None:
             raise CommandError("Set SERVY_RERANKER_MODEL_PATH to your full best_model folder.")
-        expected = str(getattr(settings, "SERVY_RERANKER_EXPECTED_SHA256", "") or "").strip().lower()
+        expected = str(
+            getattr(settings, "SERVY_RERANKER_EXPECTED_SHA256", "") or ""
+        ).strip().lower()
         if expected != SERVY_V5_WEIGHTS_SHA256:
             raise CommandError("SERVY_RERANKER_EXPECTED_SHA256 must equal the approved V5 weights digest.")
-        model = get_reranker()
-        if model is None:
-            raise CommandError("V5 reranker unavailable: inspect Django logs for the exact cause.")
+
+        # No private document, customer text, network API, or database is used.
+        # This calls the same scorer as a real RAG request, including raw-logit
+        # inference and its fail-closed behavior. A fallback must NOT pass.
+        probe = SimpleNamespace(
+            source_kind="knowledge_document",
+            text="Inspect the hydraulic pressure gauge before servicing the pump.",
+            score=0.5,
+        )
         try:
-            scores = model.predict([["machine fault", "machine fault procedure"]])
-            if len(scores) != 1:
-                raise ValueError("Model did not return exactly one score")
-            score = float(scores[0])
-            import math
-            if not math.isfinite(score):
-                raise ValueError("Non-finite score")
+            ranked = rerank_candidates(
+                "How should I inspect the hydraulic pump?",
+                [probe],
+                top_k=1,
+            )
+            if (
+                len(ranked) != 1
+                or ranked[0] is not probe
+                or getattr(probe, "reranker_model", None) != "servy_v5_bge_verified"
+                or not math.isfinite(float(getattr(probe, "rerank_logit")))
+                or not math.isfinite(float(getattr(probe, "reranker_score")))
+            ):
+                raise ValueError("V5 scorer did not produce a verified finite score")
         except Exception as exc:
-            raise CommandError(f"V5 model inference failed: {exc}") from exc
+            raise CommandError(
+                "V5 serving-path verification failed or fell back: "
+                f"{exc}. Inspect the Django reranker logs."
+            ) from exc
+
         self.stdout.write(self.style.SUCCESS(
             f"SERVY V5 RERANKER VERIFIED: {checkpoint} (device={settings.SERVY_RERANKER_DEVICE})"
         ))

core/ml/reranker.py

@@ -101,6 +101,40 @@ def get_reranker():
             return None
 
 
+def require_verified_v5_reranker() -> None:
+    """Fail startup unless the configured, SHA-pinned V5 runs real scoring.
+
+    Called only for explicit SERVY_RERANKER_REQUIRED deployments. Optional
+    development deployments retain their existing observable fallback.
+    """
+    if not is_reranker_enabled():
+        raise RuntimeError("SERVY_RERANKER_REQUIRED needs SERVY_RERANKER_ENABLED=1")
+    if _local_checkpoint() is None:
+        raise RuntimeError("SERVY_RERANKER_REQUIRED needs a local model path")
+    expected = str(getattr(settings, "SERVY_RERANKER_EXPECTED_SHA256", "") or "").strip().lower()
+    if expected != SERVY_V5_WEIGHTS_SHA256:
+        raise RuntimeError("SERVY_RERANKER_REQUIRED needs the pinned V5 weights SHA-256")
+
+    # Use the serving scorer, not a different default-activation prediction.
+    # No database, tenant data, network API, or production corpus is needed.
+    from types import SimpleNamespace
+
+    probe = SimpleNamespace(
+        source_kind="knowledge_document",
+        text="Check the hydraulic pressure gauge before servicing the pump.",
+        score=0.5,
+    )
+    ranked = rerank_candidates("How can I inspect pump pressure?", [probe], top_k=1)
+    if (
+        len(ranked) != 1
+        or ranked[0] is not probe
+        or getattr(probe, "reranker_model", None) != "servy_v5_bge_verified"
+        or not math.isfinite(float(getattr(probe, "rerank_logit", float("nan"))))
+        or not math.isfinite(float(getattr(probe, "reranker_score", float("nan"))))
+    ):
+        raise RuntimeError("Required V5 failed to load or score; refusing to start")
+
+
 # Exact digest of the exploratory V5 BGE reranker, verified on 2026-10-19.
 # This is public checkpoint identity metadata, not a secret or a promotion signal.
 SERVY_V5_WEIGHTS_SHA256 = "b1cf23ba0806f0057d73a5ab93fde88d634fc278f71b27f3a2a915afd8f8bcda"
@@ -129,10 +163,16 @@ def rerank_candidates(query: str, candidates: List, top_k: int = 8) -> List:
     No source discovery, authorization bypass, or changes to historical cases.
     Fail closed to the existing order if scoring fails.
     """
-    if not is_reranker_enabled() or not candidates:
+    if not candidates:
+        return candidates[:top_k]
+    if not is_reranker_enabled():
+        if getattr(settings, "SERVY_RERANKER_REQUIRED", False):
+            raise RuntimeError("Required V5 reranker is disabled")
         return candidates[:top_k]
     model = get_reranker()
     if model is None:
+        if getattr(settings, "SERVY_RERANKER_REQUIRED", False):
+            raise RuntimeError("Required V5 reranker failed to load")
         return candidates[:top_k]
     try:
         t0 = time.perf_counter()
@@ -170,5 +210,7 @@ def rerank_candidates(query: str, candidates: List, top_k: int = 8) -> List:
         )
         return (pool + docs[pool_size:] + others)[:top_k]
     except Exception as exc:
+        if getattr(settings, "SERVY_RERANKER_REQUIRED", False):
+            raise RuntimeError("Required V5 reranker inference failed") from exc
         logger.warning("CrossEncoder scoring failed; preserving original ranking: %s", exc)
         return candidates[:top_k]

core/services/retriever.py

@@ -1235,6 +1235,9 @@ def retrieve(
         from core.ml.reranker import rerank_candidates
         selected = rerank_candidates(evidence_text, selected, top_k=top_k)
     except Exception as exc:
+        # Required-model deployments must not silently serve baseline results.
+        if getattr(settings, "SERVY_RERANKER_REQUIRED", False):
+            raise
         logger.warning("Reranker step skipped due to error: %s", exc)
         selected = selected[:top_k]
 

core/tests_v5_activation.py

@@ -1,4 +1,8 @@
-"""The V5 verification command must never certify other checkpoint identities."""
+"""The V5 activation check must exercise the real scoring path, not a substitute.
+
+These tests are synthetic and mock the model weights. The optional real-weight
+HTTP and database test is in core.tests_v5_runtime.
+"""
 from io import StringIO
 from pathlib import Path
 from unittest import mock
@@ -16,10 +20,10 @@ class V5ActivationCommandTests(SimpleTestCase):
         SERVY_RERANKER_EXPECTED_SHA256="0" * 64,
     )
     def test_rejects_non_v5_digest_before_loading(self):
-        with mock.patch("core.management.commands.verify_v5_reranker._local_checkpoint",
-                        return_value=Path("/synthetic/best_model")), mock.patch(
-            "core.management.commands.verify_v5_reranker.get_reranker"
-        ) as load:
+        with mock.patch(
+            "core.management.commands.verify_v5_reranker._local_checkpoint",
+            return_value=Path("/synthetic/best_model"),
+        ), mock.patch("core.ml.reranker.get_reranker") as load:
             with self.assertRaisesRegex(CommandError, "approved V5"):
                 call_command("verify_v5_reranker")
         load.assert_not_called()
@@ -28,16 +32,125 @@ def test_rejects_non_v5_digest_before_loading(self):
         SERVY_RERANKER_ENABLED=True,
         SERVY_RERANKER_EXPECTED_SHA256=SERVY_V5_WEIGHTS_SHA256,
         SERVY_RERANKER_DEVICE="cpu",
+        SERVY_RERANKER_POOL_SIZE=5,
+        SERVY_RERANKER_MODEL_PATH="best_model",
     )
-    def test_v5_digest_and_finite_prediction_succeed(self):
+    def test_verified_v5_uses_serving_raw_logits(self):
         model = mock.Mock()
         model.predict.return_value = [1.5]
         output = StringIO()
-        with mock.patch("core.management.commands.verify_v5_reranker._local_checkpoint",
-                        return_value=Path("/synthetic/best_model")), mock.patch(
-            "core.management.commands.verify_v5_reranker.get_reranker",
+        with mock.patch(
+            "core.management.commands.verify_v5_reranker._local_checkpoint",
+            return_value=Path("/synthetic/best_model"),
+        ), mock.patch(
+            "core.ml.reranker.get_reranker",
             return_value=model,
         ):
             call_command("verify_v5_reranker", stdout=output)
+
         self.assertIn("SERVY V5 RERANKER VERIFIED", output.getvalue())
-        model.predict.assert_called_once_with([["machine fault", "machine fault procedure"]])
+        model.predict.assert_called_once()
+        args, kwargs = model.predict.call_args
+        self.assertEqual(
+            args[0],
+            [[
+                "How should I inspect the hydraulic pump?",
+                "Inspect the hydraulic pressure gauge before servicing the pump.",
+            ]],
+        )
+        import torch
+        self.assertIsInstance(kwargs["activation_fn"], torch.nn.Identity)
+        self.assertIs(kwargs["show_progress_bar"], False)
+
+    @override_settings(
+        SERVY_RERANKER_ENABLED=True,
+        SERVY_RERANKER_EXPECTED_SHA256=SERVY_V5_WEIGHTS_SHA256,
+        SERVY_RERANKER_POOL_SIZE=5,
+        SERVY_RERANKER_MODEL_PATH="best_model",
+    )
+    def test_model_unavailable_is_not_misreported_as_verified(self):
+        with mock.patch(
+            "core.management.commands.verify_v5_reranker._local_checkpoint",
+            return_value=Path("/synthetic/best_model"),
+        ), mock.patch("core.ml.reranker.get_reranker", return_value=None):
+            with self.assertRaisesRegex(CommandError, "serving-path verification failed"):
+                call_command("verify_v5_reranker")
+
+    @override_settings(
+        SERVY_RERANKER_ENABLED=True,
+        SERVY_RERANKER_EXPECTED_SHA256=SERVY_V5_WEIGHTS_SHA256,
+        SERVY_RERANKER_POOL_SIZE=5,
+        SERVY_RERANKER_MODEL_PATH="best_model",
+    )
+    def test_nonfinite_prediction_is_not_misreported_as_verified(self):
+        model = mock.Mock()
+        model.predict.return_value = [float("nan")]
+        with mock.patch(
+            "core.management.commands.verify_v5_reranker._local_checkpoint",
+            return_value=Path("/synthetic/best_model"),
+        ), mock.patch("core.ml.reranker.get_reranker", return_value=model):
+            with self.assertRaisesRegex(CommandError, "serving-path verification failed"):
+                call_command("verify_v5_reranker")
+
+    @override_settings(SERVY_RERANKER_ENABLED=False)
+    def test_disabled_model_is_not_misreported_as_verified(self):
+        with mock.patch("core.ml.reranker.get_reranker") as load:
+            with self.assertRaisesRegex(CommandError, "SERVY_RERANKER_ENABLED=1"):
+                call_command("verify_v5_reranker")
+        load.assert_not_called()
+
+    @override_settings(SERVY_RERANKER_REQUIRED=True, SERVY_RERANKER_ENABLED=False)
+    def test_required_v5_blocks_disabled_configuration(self):
+        from core.ml.reranker import require_verified_v5_reranker
+        with self.assertRaisesRegex(RuntimeError, "SERVY_RERANKER_ENABLED"):
+            require_verified_v5_reranker()
+
+    @override_settings(
+        SERVY_RERANKER_REQUIRED=True, SERVY_RERANKER_ENABLED=True,
+        SERVY_RERANKER_MODEL_PATH="best_model",
+        SERVY_RERANKER_EXPECTED_SHA256="0" * 64,
+    )
+    def test_required_v5_blocks_unapproved_digest_without_loading(self):
+        from core.ml.reranker import require_verified_v5_reranker
+        with mock.patch("core.ml.reranker.get_reranker") as loader:
+            with self.assertRaisesRegex(RuntimeError, "pinned V5"):
+                require_verified_v5_reranker()
+        loader.assert_not_called()
+
+    @override_settings(
+        SERVY_RERANKER_REQUIRED=True, SERVY_RERANKER_ENABLED=True,
+        SERVY_RERANKER_MODEL_PATH="best_model",
+        SERVY_RERANKER_EXPECTED_SHA256=SERVY_V5_WEIGHTS_SHA256,
+        SERVY_RERANKER_POOL_SIZE=5,
+    )
+    def test_required_v5_blocks_unavailable_checkpoint_fallback(self):
+        from core.ml.reranker import require_verified_v5_reranker
+        with mock.patch("core.ml.reranker.get_reranker", return_value=None):
+            with self.assertRaisesRegex(RuntimeError, "Required V5 reranker failed to load"):
+                require_verified_v5_reranker()
+
+    @override_settings(
+        SERVY_RERANKER_REQUIRED=True, SERVY_RERANKER_ENABLED=True,
+        SERVY_RERANKER_MODEL_PATH="best_model",
+        SERVY_RERANKER_EXPECTED_SHA256=SERVY_V5_WEIGHTS_SHA256,
+        SERVY_RERANKER_POOL_SIZE=5,
+    )
+    def test_required_v5_executes_actual_serving_score(self):
+        from core.ml.reranker import require_verified_v5_reranker
+        model = mock.Mock()
+        model.predict.return_value = [1.25]
+        with mock.patch("core.ml.reranker.get_reranker", return_value=model):
+            require_verified_v5_reranker()
+        model.predict.assert_called_once()
+        import torch
+        self.assertIsInstance(model.predict.call_args.kwargs["activation_fn"], torch.nn.Identity)
+
+    @override_settings(SERVY_RERANKER_REQUIRED=True)
+    def test_django_ready_refuses_startup_on_required_model_failure(self):
+        from django.apps import apps
+        with mock.patch(
+            "core.ml.reranker.require_verified_v5_reranker",
+            side_effect=RuntimeError("required model unavailable"),
+        ):
+            with self.assertRaisesRegex(RuntimeError, "required model unavailable"):
+                apps.get_app_config("core").ready()

core/tests_v5_reranker.py

@@ -225,3 +225,25 @@ def test_stale_minilm_vector_cannot_supply_a_v5_candidate(self):
         self.assertFalse(_matches_current_minilm_vector(
             {**metadata, "chunk_id": 999}, source, profile,
         ))
+
+    @override_settings(SERVY_RERANKER_ENABLED=True, SERVY_RERANKER_REQUIRED=True)
+    def test_required_mode_propagates_missing_model(self):
+        docs = [chunk("approved service instruction")]
+        with mock.patch.object(reranker, "get_reranker", return_value=None):
+            with self.assertRaisesRegex(RuntimeError, "failed to load"):
+                reranker.rerank_candidates("service instruction", docs, top_k=1)
+
+    @override_settings(SERVY_RERANKER_ENABLED=True, SERVY_RERANKER_REQUIRED=True)
+    def test_required_mode_propagates_inference_failure(self):
+        docs = [chunk("approved service instruction")]
+        model = mock.Mock()
+        model.predict.side_effect = ValueError("synthetic scoring failure")
+        with mock.patch.object(reranker, "get_reranker", return_value=model):
+            with self.assertRaisesRegex(RuntimeError, "inference failed"):
+                reranker.rerank_candidates("service instruction", docs, top_k=1)
+
+    @override_settings(SERVY_RERANKER_ENABLED=True, SERVY_RERANKER_REQUIRED=True)
+    def test_required_mode_still_allows_empty_authorized_results(self):
+        with mock.patch.object(reranker, "get_reranker") as loader:
+            self.assertEqual(reranker.rerank_candidates("service instruction", [], top_k=1), [])
+        loader.assert_not_called()