port: migrate from mp.solutions to mp.tasks.vision (FaceLandmarker)

.env.example

@@ -175,3 +175,15 @@ FEATURE_FACE_VERIFICATION=True
 FEATURE_LIVENESS_DETECTION=True
 FEATURE_BATCH_PROCESSING=True
 FEATURE_QUALITY_ASSESSMENT=True
+
+# ============================================================================
+# MediaPipe Face Landmarker (Tasks API — ported 2026-05-12)
+# ----------------------------------------------------------------------------
+# The runtime container bakes face_landmarker.task at /app/models/. Override
+# the path in dev to point at a local copy, or in prod to point at an
+# operator-rotated asset. SHA256 is enforced when set.
+# ============================================================================
+FACE_LANDMARKER_MODEL_PATH=/app/models/face_landmarker.task
+# float16/latest as of 2026-05-12; verify with:
+#   sha256sum /app/models/face_landmarker.task
+FACE_LANDMARKER_MODEL_SHA256=64184e229b263107bc2b804c6625db1341ff2bb731874b0bcc2fe6544e0bc9ff

Dockerfile

@@ -58,6 +58,23 @@ RUN python -c "import cv2; print('OpenCV version:', cv2.__version__)" && \
     python -c "import numpy; print('NumPy version:', numpy.__version__)" && \
     python -c "import tensorflow; print('TensorFlow version:', tensorflow.__version__)"
 
+# Bake MediaPipe face_landmarker.task into the image (2026-05-12).
+# The new Tasks API (mp.tasks.vision.FaceLandmarker) requires a .task model
+# asset; without it the gaze tracker, active-liveness detector, quality
+# assessor and landmark detector all fail-soft with "model missing".
+# Pinning + SHA-verifying at build time gives us a reproducible image and
+# closes the supply-chain check the loader expects at runtime.
+# Companion env vars in .env.example:
+#   FACE_LANDMARKER_MODEL_PATH=/app/models/face_landmarker.task
+#   FACE_LANDMARKER_MODEL_SHA256=64184e229b263107bc2b804c6625db1341ff2bb731874b0bcc2fe6544e0bc9ff
+ARG FACE_LANDMARKER_SHA256=64184e229b263107bc2b804c6625db1341ff2bb731874b0bcc2fe6544e0bc9ff
+RUN set -eux; \
+    mkdir -p /app/models; \
+    curl -fsSL -o /app/models/face_landmarker.task \
+        "https://storage.googleapis.com/mediapipe-models/face_landmarker/face_landmarker/float16/latest/face_landmarker.task"; \
+    echo "${FACE_LANDMARKER_SHA256}  /app/models/face_landmarker.task" | sha256sum -c -; \
+    chmod 0644 /app/models/face_landmarker.task
+
 # Copy application code
 COPY . .
 

app/core/config.py

@@ -693,11 +693,24 @@ def get_api_key_config(self) -> dict:
             ".env.prod when shipping; empty = skip verification (dev only)."
         ),
     )
+    FACE_LANDMARKER_MODEL_PATH: str = Field(
+        default=str(_REPO_ROOT / "models" / "face_landmarker.task"),
+        description=(
+            "Filesystem path to the MediaPipe face_landmarker.task asset. "
+            "Used by server-side facial-landmark consumers (gaze tracker, "
+            "active liveness, quality assessor) after the 2026-05-12 port "
+            "from mp.solutions to mp.tasks.vision. The runtime container "
+            "bakes this file at /app/models/face_landmarker.task; override "
+            "via the env var to point at an operator-rotated asset."
+        ),
+    )
     FACE_LANDMARKER_MODEL_SHA256: str = Field(
         default="",
         description=(
             "Expected SHA256 hex digest for face_landmarker.task. Set in "
-            ".env.prod; empty = skip verification with a log warning."
+            ".env.prod; empty = skip verification with a log warning. "
+            "Live asset SHA256 (float16/latest, 2026-05-12): "
+            "64184e229b263107bc2b804c6625db1341ff2bb731874b0bcc2fe6544e0bc9ff"
         ),
     )
 

app/infrastructure/ml/landmarks/face_landmarker_loader.py

@@ -0,0 +1,165 @@
+"""Shared MediaPipe Face Landmarker (Tasks API) loader.
+
+Centralises the `.task` model resolution + SHA256 integrity check used by
+every server-side consumer of facial landmarks. Replaces the deprecated
+`mediapipe.solutions.face_mesh` API (removed in mediapipe 0.10.35).
+
+The loader honours:
+- ``FACE_LANDMARKER_MODEL_PATH`` env var (per-env override).
+- A repo-relative ``./models/face_landmarker.task`` default.
+- ``FACE_LANDMARKER_MODEL_SHA256`` env var for integrity verification
+  (warn-and-disable on mismatch; production MUST set this).
+
+Consumers should treat the returned object as opaque and call ``.detect()``
+or ``.detect_for_video()`` against an ``mp.Image`` instance.
+
+This module DOES NOT cache the loader globally; each consumer keeps a
+process-local cache appropriate to its lifecycle (singleton vs. per-request).
+"""
+
+from __future__ import annotations
+
+import hashlib
+import logging
+import os
+from pathlib import Path
+from typing import Any, Optional
+
+logger = logging.getLogger(__name__)
+
+# Default location: <repo>/models/face_landmarker.task. The file is gitignored
+# but is baked into the runtime container by the Dockerfile (see PR
+# `port: migrate from mp.solutions to mp.tasks.vision`).
+_DEFAULT_MODEL_PATH = (
+    Path(__file__).resolve().parent.parent.parent.parent / "models" / "face_landmarker.task"
+)
+
+
+def resolve_face_landmarker_model_path() -> Optional[Path]:
+    """Return the on-disk path to face_landmarker.task or None if absent.
+
+    Resolution order:
+      1. ``FACE_LANDMARKER_MODEL_PATH`` env var (if set, must exist).
+      2. ``<repo>/models/face_landmarker.task``.
+      3. ``./models/face_landmarker.task`` relative to cwd (dev fallback).
+    """
+    env_path = os.getenv("FACE_LANDMARKER_MODEL_PATH", "").strip()
+    candidates = []
+    if env_path:
+        candidates.append(Path(env_path))
+    candidates.append(_DEFAULT_MODEL_PATH)
+    candidates.append(Path("models/face_landmarker.task"))
+
+    for candidate in candidates:
+        if candidate.exists():
+            return candidate
+    return None
+
+
+def verify_model_sha256(model_path: Path) -> bool:
+    """Verify the SHA256 of the model file against ``FACE_LANDMARKER_MODEL_SHA256``.
+
+    Returns True when:
+      - The env var is unset (verification skipped — dev only).
+      - The env var matches the on-disk file.
+
+    Returns False when the env var is set and the digest mismatches.
+    Logs a warning either way.
+    """
+    expected = os.getenv("FACE_LANDMARKER_MODEL_SHA256", "").strip()
+    if not expected:
+        logger.warning(
+            "FACE_LANDMARKER_MODEL_SHA256 not set — model integrity NOT verified. "
+            "Set this in production to prevent supply-chain tampering."
+        )
+        return True
+    try:
+        actual = hashlib.sha256(model_path.read_bytes()).hexdigest()
+    except OSError as exc:
+        logger.error("Could not read face_landmarker.task for SHA256 check: %s", exc)
+        return False
+    if actual.lower() != expected.lower():
+        logger.error(
+            "face_landmarker.task SHA256 mismatch: expected=%s actual=%s path=%s",
+            expected,
+            actual,
+            model_path,
+        )
+        return False
+    return True
+
+
+def create_face_landmarker(
+    *,
+    static_image_mode: bool = True,
+    num_faces: int = 1,
+    output_face_blendshapes: bool = False,
+    min_face_detection_confidence: float = 0.5,
+    min_face_presence_confidence: float = 0.5,
+    min_tracking_confidence: float = 0.5,
+) -> Optional[Any]:
+    """Create a MediaPipe FaceLandmarker (Tasks API) or return None on failure.
+
+    Args mirror the relevant subset of the legacy ``mp.solutions.face_mesh``
+    constructor so call sites can be ported in-place. ``static_image_mode``
+    maps to ``RunningMode.IMAGE`` (True) or ``RunningMode.VIDEO`` (False).
+
+    Returns:
+        FaceLandmarker instance ready for ``.detect(mp_image)`` (IMAGE mode) or
+        ``.detect_for_video(mp_image, timestamp_ms)`` (VIDEO mode). Returns
+        ``None`` if MediaPipe is missing, the model file is absent, or the
+        SHA256 check fails — callers should fail-soft.
+    """
+    try:
+        from mediapipe.tasks import python as mp_python
+        from mediapipe.tasks.python import vision as mp_vision
+    except ImportError as exc:
+        logger.error("MediaPipe Tasks API not importable: %s", exc)
+        return None
+
+    model_path = resolve_face_landmarker_model_path()
+    if model_path is None:
+        logger.warning(
+            "face_landmarker.task not found. Set FACE_LANDMARKER_MODEL_PATH or "
+            "place the asset under <repo>/models/."
+        )
+        return None
+    if not verify_model_sha256(model_path):
+        return None
+
+    try:
+        running_mode = mp_vision.RunningMode.IMAGE if static_image_mode else mp_vision.RunningMode.VIDEO
+        base_options = mp_python.BaseOptions(model_asset_path=str(model_path))
+        options = mp_vision.FaceLandmarkerOptions(
+            base_options=base_options,
+            running_mode=running_mode,
+            num_faces=num_faces,
+            min_face_detection_confidence=min_face_detection_confidence,
+            min_face_presence_confidence=min_face_presence_confidence,
+            min_tracking_confidence=min_tracking_confidence,
+            output_face_blendshapes=output_face_blendshapes,
+        )
+        landmarker = mp_vision.FaceLandmarker.create_from_options(options)
+        logger.info(
+            "MediaPipe FaceLandmarker (Tasks API) initialised "
+            "(mode=%s, num_faces=%s, blendshapes=%s, model=%s)",
+            running_mode.name,
+            num_faces,
+            output_face_blendshapes,
+            model_path,
+        )
+        return landmarker
+    except Exception:  # noqa: BLE001
+        logger.exception("Failed to create FaceLandmarker from Tasks API")
+        return None
+
+
+def to_mp_image(image_rgb):
+    """Wrap an RGB numpy array in an ``mp.Image`` (SRGB format).
+
+    Helper so call sites don't need ``import mediapipe`` just for the format
+    enum. The caller is responsible for BGR→RGB conversion.
+    """
+    import mediapipe as mp
+
+    return mp.Image(image_format=mp.ImageFormat.SRGB, data=image_rgb)

app/infrastructure/ml/landmarks/mediapipe_landmarks.py

@@ -1,4 +1,10 @@
-"""MediaPipe-based facial landmark detector implementation."""
+"""MediaPipe-based facial landmark detector implementation.
+
+Ported 2026-05-12 from the legacy ``mediapipe.solutions.face_mesh`` API to
+``mediapipe.tasks.vision.FaceLandmarker``. The ``mp.solutions`` namespace was
+removed in mediapipe 0.10.35; the new Tasks API requires a ``.task`` model
+asset and exposes landmarks as ``result.face_landmarks[0][i].(x|y|z)``.
+"""
 
 import logging
 from typing import List, Optional
@@ -7,17 +13,23 @@
 
 from app.domain.entities.face_landmarks import HeadPose, Landmark, LandmarkResult
 from app.domain.exceptions.feature_errors import LandmarkError
+from app.infrastructure.ml.landmarks.face_landmarker_loader import (
+    create_face_landmarker,
+    to_mp_image,
+)
 
 logger = logging.getLogger(__name__)
 
 
 class MediaPipeLandmarkDetector:
-    """Facial landmark detector using MediaPipe Face Mesh.
+    """Facial landmark detector using MediaPipe Face Landmarker (Tasks API).
 
     Detects 468 facial landmarks with optional 3D coordinates.
     """
 
-    # Facial region indices for MediaPipe Face Mesh
+    # Facial region indices for MediaPipe Face Mesh (canonical 468-pt topology;
+    # indices are stable between the legacy face_mesh and the Tasks-API
+    # face_landmarker outputs).
     REGIONS = {
         "left_eye": [33, 133, 160, 159, 158, 144, 145, 153],
         "right_eye": [362, 263, 387, 386, 385, 373, 374, 380],
@@ -34,21 +46,23 @@ class MediaPipeLandmarkDetector:
 
     def __init__(self) -> None:
         """Initialize MediaPipe landmark detector."""
-        self._face_mesh = None
-        logger.info("MediaPipeLandmarkDetector initialized")
-
-    def _get_face_mesh(self):
-        """Lazy load MediaPipe Face Mesh."""
-        if self._face_mesh is None:
-            import mediapipe as mp
+        self._face_landmarker = None
+        logger.info("MediaPipeLandmarkDetector initialized (Tasks API)")
 
-            self._face_mesh = mp.solutions.face_mesh.FaceMesh(
+    def _get_face_landmarker(self):
+        """Lazy load MediaPipe Face Landmarker (Tasks API)."""
+        if self._face_landmarker is None:
+            self._face_landmarker = create_face_landmarker(
                 static_image_mode=True,
-                max_num_faces=1,
-                refine_landmarks=True,
-                min_detection_confidence=0.5,
+                num_faces=1,
+                min_face_detection_confidence=0.5,
             )
-        return self._face_mesh
+            if self._face_landmarker is None:
+                raise LandmarkError(
+                    "MediaPipe FaceLandmarker unavailable — model asset missing or "
+                    "Tasks API not importable. See logs for details."
+                )
+        return self._face_landmarker
 
     def detect(
         self, image: np.ndarray, include_3d: bool = False
@@ -68,19 +82,25 @@ def detect(
         logger.debug(f"Starting landmark detection (include_3d={include_3d})")
 
         try:
-            face_mesh = self._get_face_mesh()
-            results = face_mesh.process(image)
-
-            if not results.multi_face_landmarks:
+            face_landmarker = self._get_face_landmarker()
+            # Tasks API expects an mp.Image wrapping an RGB ndarray. Callers
+            # of this method already pass RGB (see the docstring), so no
+            # additional colour-space conversion is needed.
+            mp_image = to_mp_image(image)
+            result = face_landmarker.detect(mp_image)
+
+            face_landmarks_list = result.face_landmarks or []
+            if not face_landmarks_list:
                 raise LandmarkError("No face landmarks detected")
 
-            # Get first face landmarks
-            face_landmarks = results.multi_face_landmarks[0]
+            # Get first face landmarks. In the Tasks API each element is itself
+            # a flat list of NormalizedLandmark (no `.landmark` attribute).
+            face_landmarks = face_landmarks_list[0]
             height, width = image.shape[:2]
 
             # Extract landmarks
             landmarks = []
-            for idx, lm in enumerate(face_landmarks.landmark):
+            for idx, lm in enumerate(face_landmarks):
                 x = int(lm.x * width)
                 y = int(lm.y * height)
                 z = lm.z if include_3d else None