fix(ai): bind model scope to immutable geometry

backend/app/services/detection_service.py

@@ -22,6 +22,7 @@ from app.services.detection_qa_service import DetectionQaService
 from app.services.dataset_consumption_gate_service import DatasetConsumptionGate
 from app.services.model_asset_catalog_service import ModelAssetCatalogService
 from app.services.model_registry_service import ModelRegistryService
+from app.services.model_validation_scope_service import ModelValidationScopeService
 from app.services.qa_service import QaService
 from app.services.quality_service import QualityService
 from app.services.runtime_model_provenance_service import RuntimeModelProvenance, RuntimeModelProvenanceService
@@ -235,16 +236,30 @@ class DetectionService:
 
     @staticmethod
     def _validate_model_area_scope(db, dataset: Dataset, settings: Settings) -> None:
-        allowed_names = [value.strip().casefold() for value in settings.yolo_validated_area_names.split(",") if value.strip()]
         area = db.get(Area, dataset.area_id) if dataset.area_id else None
-        area_name = area.name.strip() if area is not None else ""
-        if not area_name or not any(token in area_name.casefold() for token in allowed_names):
+        if area is None or area.geometry is None:
             raise AppError(
                 code="DETECTION_VALIDATION_SCOPE_UNAVAILABLE",
-                message="Configured YOLO inference is not validated for this Dataset area.",
-                details={"dataset_id": str(dataset.id), "dataset_area": area_name or None, "validated_area_names": allowed_names},
+                message="Configured YOLO inference requires a persisted Dataset area geometry.",
+                details={"dataset_id": str(dataset.id)},
                 status_code=422,
             )
+        try:
+            area_geometry = to_shape(area.geometry)
+        except Exception as exc:
+            raise AppError(
+                code="DETECTION_VALIDATION_SCOPE_UNAVAILABLE",
+                message="The persisted Dataset area geometry cannot be validated for model inference.",
+                details={"dataset_id": str(dataset.id), "error_type": type(exc).__name__},
+                status_code=422,
+            ) from exc
+        ModelValidationScopeService.assert_area_covered(
+            area_geometry=area_geometry,
+            manifest_path=settings.yolo_validation_scope_manifest_path,
+            expected_manifest_sha256=settings.yolo_validation_scope_manifest_sha256,
+            model_id=settings.yolo_model_id,
+            model_path=settings.yolo_model_path,
+        )
     @staticmethod
     def _fail_run_after_exception(db, analysis_run: AnalysisRun, job: Job, exc: Exception, fallback_code: str) -> None:
         try:

backend/app/services/model_validation_scope_service.py

@@ -0,0 +1,142 @@
+"""Checksum-bound geographic validation scope for production model inference."""
+
+from __future__ import annotations
+
+from hashlib import sha256
+import json
+from pathlib import Path
+import re
+from typing import Any
+
+from shapely.geometry import shape
+from shapely.geometry.base import BaseGeometry
+
+from app.core.errors import AppError
+
+
+_SHA256 = re.compile(r"^[0-9a-f]{64}$")
+
+
+class ModelValidationScopeService:
+    """Load an immutable, model-bound AOI and prove that an input is covered."""
+
+    SCHEMA_VERSION = "geointel.model-validation-scope/v1"
+
+    @classmethod
+    def assert_area_covered(
+        cls,
+        *,
+        area_geometry: BaseGeometry,
+        manifest_path: str | None,
+        expected_manifest_sha256: str | None,
+        model_id: str,
+        model_path: str | None,
+    ) -> dict[str, str]:
+        path = Path(manifest_path).expanduser() if manifest_path else None
+        expected_checksum = (expected_manifest_sha256 or "").strip().lower()
+        if path is None or not expected_checksum:
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_NOT_CONFIGURED",
+                "Configured YOLO inference requires a checksum-bound geographic validation-scope manifest.",
+            )
+        if not _SHA256.fullmatch(expected_checksum):
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_INVALID",
+                "The configured validation-scope checksum must be a lowercase SHA-256 digest.",
+                manifest_path=str(path),
+            )
+        try:
+            raw_manifest = path.read_bytes()
+            payload = json.loads(raw_manifest.decode("utf-8"))
+        except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_INVALID",
+                "The configured validation-scope manifest is missing or unreadable.",
+                manifest_path=str(path),
+                error_type=type(exc).__name__,
+            )
+        observed_manifest_sha256 = sha256(raw_manifest).hexdigest()
+        if observed_manifest_sha256 != expected_checksum:
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_CHECKSUM_MISMATCH",
+                "The validation-scope manifest does not match its configured checksum.",
+                manifest_path=str(path),
+                expected=expected_checksum,
+                observed=observed_manifest_sha256,
+            )
+        if not isinstance(payload, dict) or payload.get("schema_version") != cls.SCHEMA_VERSION:
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_INVALID",
+                "The validation-scope manifest has an unsupported schema.",
+                manifest_path=str(path),
+            )
+        if payload.get("model_id") != model_id:
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_MODEL_MISMATCH",
+                "The validation scope is not bound to the selected model identity.",
+                expected_model_id=model_id,
+                observed_model_id=payload.get("model_id"),
+            )
+        configured_model_path = Path(model_path).expanduser() if model_path else None
+        if configured_model_path is None or not configured_model_path.is_file():
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_MODEL_UNAVAILABLE",
+                "The model bytes bound by the validation scope are unavailable.",
+            )
+        declared_model_sha256 = str(payload.get("model_sha256") or "").strip().lower()
+        observed_model_sha256 = cls._file_sha256(configured_model_path)
+        if not _SHA256.fullmatch(declared_model_sha256) or declared_model_sha256 != observed_model_sha256:
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_MODEL_MISMATCH",
+                "The validation scope is not bound to the exact selected model bytes.",
+                expected=declared_model_sha256 or None,
+                observed=observed_model_sha256,
+            )
+        if payload.get("crs") != "EPSG:4326":
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_INVALID",
+                "The validation-scope geometry must explicitly use EPSG:4326.",
+                observed_crs=payload.get("crs"),
+            )
+        try:
+            scope_geometry = shape(payload.get("geometry"))
+        except Exception as exc:
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_INVALID",
+                "The validation-scope geometry is not valid GeoJSON.",
+                error_type=type(exc).__name__,
+            )
+        if (
+            scope_geometry.is_empty
+            or not scope_geometry.is_valid
+            or scope_geometry.geom_type not in {"Polygon", "MultiPolygon"}
+        ):
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_INVALID",
+                "The validation scope must be a non-empty valid Polygon or MultiPolygon.",
+                geometry_type=scope_geometry.geom_type,
+            )
+        if area_geometry.is_empty or not area_geometry.is_valid or not scope_geometry.covers(area_geometry):
+            cls._raise(
+                "DETECTION_VALIDATION_SCOPE_UNAVAILABLE",
+                "Configured YOLO inference is not validated for the complete Dataset area.",
+                scope_key=payload.get("scope_key"),
+            )
+        return {
+            "scope_key": str(payload.get("scope_key") or "unspecified"),
+            "manifest_path": str(path.resolve()),
+            "manifest_sha256": observed_manifest_sha256,
+            "model_sha256": observed_model_sha256,
+        }
+
+    @staticmethod
+    def _file_sha256(path: Path) -> str:
+        digest = sha256()
+        with path.open("rb") as handle:
+            for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+                digest.update(chunk)
+        return digest.hexdigest()
+
+    @staticmethod
+    def _raise(code: str, message: str, **details: Any) -> None:
+        raise AppError(code=code, message=message, details=details, status_code=422)