Resolve PAD model review findings

.env.example

@@ -58,7 +58,7 @@ PAD_MODEL_RELATIVE_PATH=liveness/pad.onnx
 PAD_MODEL_NAME=MiniFASNetV2
 PAD_MODEL_VERSION=2.7_80x80
 PAD_OUTPUT_KIND=logits
-PAD_LIVE_CLASS_INDEX=0
+PAD_LIVE_CLASS_INDEX=1
 PAD_CROP_SCALE=2.7
 PAD_MIN_SCORE=0.80
 

backend/ai-service/README.md

@@ -32,7 +32,8 @@ docker compose --profile model-bootstrap run --rm ai-model-bootstrap
 ```
 
 The initial candidate is MiniFASNetV2 2.7_80x80 with a three-logit output
-(`live`, `print attack`, `replay attack`). It is a baseline candidate,
+(`print attack`, `live`, `replay attack`), so `PAD_LIVE_CLASS_INDEX` defaults
+to `1`. It is a baseline candidate,
 not a production performance claim; it must pass our held-out Ghana/device PAD
 evaluation before pilot use.
 

backend/ai-service/app/bootstrap_models.py

@@ -5,6 +5,7 @@
 import yaml
 
 from app.pipeline import get_insightface_analyzer, get_paddle_ocr_engine
+from app.pad import validate_pad_model_contract
 from app.settings import get_settings
 
 
@@ -27,12 +28,18 @@ def _persist_downloaded_paddle_models(settings) -> None:
         if settings.paddle_model_is_complete(target):
             continue
         source = next(
-            (root / name for root in official_roots for name in PADDLE_MODEL_CANDIDATES[kind]
-             if settings.paddle_model_is_complete(root / name)),
+            (
+                root / name
+                for root in official_roots
+                for name in PADDLE_MODEL_CANDIDATES[kind]
+                if settings.paddle_model_is_complete(root / name)
+            ),
             None,
         )
         if source is None:
-            raise RuntimeError(f"Downloaded PaddleOCR {kind} model was not found under: {', '.join(map(str, official_roots))}.")
+            raise RuntimeError(
+                f"Downloaded PaddleOCR {kind} model was not found under: {', '.join(map(str, official_roots))}."
+            )
         shutil.copytree(source, target, dirs_exist_ok=True)
 
 
@@ -49,6 +56,7 @@ def main() -> None:
             "PAD model is missing. Place the approved ONNX asset at "
             f"{settings.pad_model_path} before bootstrapping production models."
         )
+    validate_pad_model_contract(settings.pad_model_path, settings.pad_live_class_index)
     paddle_directories = (
         settings.paddle_text_detection_model_dir,
         settings.paddle_text_recognition_model_dir,
@@ -61,32 +69,49 @@ def main() -> None:
 
     _persist_downloaded_paddle_models(settings)
 
-    incomplete_directories = [str(path) for path in paddle_directories if not settings.paddle_model_is_complete(path)]
+    incomplete_directories = [
+        str(path)
+        for path in paddle_directories
+        if not settings.paddle_model_is_complete(path)
+    ]
     if incomplete_directories:
         raise RuntimeError(
-            "PaddleOCR bootstrap did not populate complete models: " + ", ".join(incomplete_directories)
+            "PaddleOCR bootstrap did not populate complete models: "
+            + ", ".join(incomplete_directories)
         )
     files = []
     manifest_path = getattr(
         settings, "model_manifest_path", settings.ai_model_root / "manifest.json"
     )
     for path in sorted(settings.ai_model_root.rglob("*")):
         if path.is_file() and path != manifest_path:
-            files.append({
-                "path": str(path.relative_to(settings.ai_model_root)),
-                "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
-                "bytes": path.stat().st_size,
-            })
+            files.append(
+                {
+                    "path": str(path.relative_to(settings.ai_model_root)),
+                    "sha256": hashlib.sha256(path.read_bytes()).hexdigest(),
+                    "bytes": path.stat().st_size,
+                }
+            )
     manifest_path.parent.mkdir(parents=True, exist_ok=True)
-    manifest_path.write_text(json.dumps({
-        "insightface_model": settings.insightface_model_name,
-        "paddle_models": {
-            "detection": _model_name(settings.paddle_text_detection_model_dir),
-            "recognition": _model_name(settings.paddle_text_recognition_model_dir),
-        },
-        "files": files,
-    }, indent=2), encoding="utf-8")
-    print(f"Model bootstrap complete: {len(files)} artifacts recorded in {manifest_path}")
+    manifest_path.write_text(
+        json.dumps(
+            {
+                "insightface_model": settings.insightface_model_name,
+                "paddle_models": {
+                    "detection": _model_name(settings.paddle_text_detection_model_dir),
+                    "recognition": _model_name(
+                        settings.paddle_text_recognition_model_dir
+                    ),
+                },
+                "files": files,
+            },
+            indent=2,
+        ),
+        encoding="utf-8",
+    )
+    print(
+        f"Model bootstrap complete: {len(files)} artifacts recorded in {manifest_path}"
+    )
 
 
 if __name__ == "__main__":

backend/ai-service/app/pad.py

@@ -11,6 +11,8 @@
 """
 
 from functools import lru_cache
+import hashlib
+from pathlib import Path
 from typing import Any
 
 import cv2
@@ -30,13 +32,84 @@ def get_pad_session() -> ort.InferenceSession:
             f"PAD model is missing: {model_path}. Provide a verified model asset."
         )
     try:
-        return ort.InferenceSession(
+        session = ort.InferenceSession(
             str(model_path), providers=["CPUExecutionProvider"]
         )
     except Exception as exc:
         raise ProcessingConfigurationError(
             f"PAD model could not be loaded: {model_path}"
         ) from exc
+    validate_pad_session_contract(session, get_settings().pad_live_class_index)
+    return session
+
+
+def validate_pad_session_contract(
+    session: ort.InferenceSession, live_class_index: int
+) -> None:
+    inputs = session.get_inputs()
+    outputs = session.get_outputs()
+    if len(inputs) != 1 or not outputs:
+        raise ProcessingConfigurationError(
+            "PAD model must expose exactly one input and at least one output."
+        )
+    input_shape = inputs[0].shape
+    if inputs[0].type != "tensor(float)":
+        raise ProcessingConfigurationError(
+            "PAD model input must use float32 tensor elements."
+        )
+    if len(input_shape) != 4 or (
+        isinstance(input_shape[1], int) and input_shape[1] != 3
+    ):
+        raise ProcessingConfigurationError(
+            "PAD model input must be a four-dimensional NCHW RGB tensor."
+        )
+    for dimension in input_shape[2:]:
+        if isinstance(dimension, int) and dimension <= 0:
+            raise ProcessingConfigurationError(
+                "PAD model spatial dimensions must be positive."
+            )
+    output_shape = outputs[0].shape
+    if outputs[0].type != "tensor(float)":
+        raise ProcessingConfigurationError(
+            "PAD model output must use float32 tensor elements."
+        )
+    if len(output_shape) < 2:
+        raise ProcessingConfigurationError(
+            "PAD model output must include batch and class dimensions."
+        )
+    if output_shape and isinstance(output_shape[-1], int):
+        class_count = output_shape[-1]
+        if class_count > 1 and not 0 <= live_class_index < class_count:
+            raise ProcessingConfigurationError(
+                "PAD live-class index is outside the model output contract."
+            )
+
+
+@lru_cache(maxsize=8)
+def _validate_pad_model_contract(
+    model_path: Path, live_class_index: int, model_digest: str
+) -> None:
+    try:
+        session = ort.InferenceSession(
+            str(model_path), providers=["CPUExecutionProvider"]
+        )
+        validate_pad_session_contract(session, live_class_index)
+    except ProcessingConfigurationError:
+        raise
+    except Exception as exc:
+        raise ProcessingConfigurationError(
+            f"PAD model could not be loaded: {model_path}"
+        ) from exc
+
+
+def validate_pad_model_contract(model_path: Path, live_class_index: int) -> None:
+    try:
+        model_digest = hashlib.sha256(model_path.read_bytes()).hexdigest()
+    except OSError as exc:
+        raise ProcessingConfigurationError(
+            f"PAD model could not be read: {model_path}"
+        ) from exc
+    _validate_pad_model_contract(model_path, live_class_index, model_digest)
 
 
 def _softmax(values: np.ndarray) -> np.ndarray:
@@ -83,10 +156,22 @@ def run_pad_model(
         bbox = face_boxes[index] if face_boxes and index < len(face_boxes) else None
         cropped = _crop_face(frame, bbox, settings.pad_crop_scale)
         resized = cv2.resize(cropped, (width, height), interpolation=cv2.INTER_AREA)
-        tensors.append(np.transpose(resized.astype(np.float32) / 255.0, (2, 0, 1)))
-
-    raw = session.run(None, {input_meta.name: np.stack(tensors, axis=0)})[0]
-    values = np.asarray(raw, dtype=np.float32)
+        rgb = cv2.cvtColor(resized, cv2.COLOR_BGR2RGB)
+        tensors.append(np.transpose(rgb.astype(np.float32) / 255.0, (2, 0, 1)))
+
+    declared_batch = input_shape[0]
+    batch_size = declared_batch if isinstance(declared_batch, int) else len(tensors)
+    if batch_size <= 0:
+        raise ProcessingConfigurationError("PAD model batch dimension is invalid.")
+    output_batches = []
+    for start in range(0, len(tensors), batch_size):
+        chunk = tensors[start : start + batch_size]
+        actual_size = len(chunk)
+        if actual_size < batch_size:
+            chunk.extend([chunk[-1]] * (batch_size - actual_size))
+        raw = session.run(None, {input_meta.name: np.stack(chunk, axis=0)})[0]
+        output_batches.append(np.asarray(raw, dtype=np.float32)[:actual_size])
+    values = np.concatenate(output_batches, axis=0)
     if values.ndim == 1:
         values = values[:, None]
     values = values.reshape(values.shape[0], -1)
@@ -96,9 +181,7 @@ def run_pad_model(
             live_scores = 1.0 - live_scores
     else:
         probabilities = (
-            _softmax(values)
-            if settings.pad_output_kind == "logits"
-            else values
+            _softmax(values) if settings.pad_output_kind == "logits" else values
         )
         live_scores = probabilities[:, settings.pad_live_class_index]