feat: add governed nationwide AOI orchestration and CUDA enforcement

backend/app/services/aoi_operation_executor.py

@@ -0,0 +1,104 @@
+from __future__ import annotations
+
+from uuid import UUID
+
+from geoalchemy2.shape import to_shape
+
+from app.core.errors import AppError
+from app.models import AoiOperation, AoiOperationPartition
+from app.schemas.grb import GrbAcquireRequest
+from app.schemas.dhmv import DhmvAcquireRequest
+from app.schemas.spw_terrain import SpwTerrainAcquireRequest
+from app.schemas.official_vector import OfficialVectorAcquireRequest
+from app.schemas.flood_hazard import FloodHazardAcquireRequest
+from app.schemas.thematic_raster import ThematicRasterAcquireRequest
+from app.schemas.bathymetry import BathymetryProfileAcquireRequest, MdkBathymetryAcquireRequest
+from app.schemas.job import JobCreate
+from app.schemas.operations import VectorSelectionBBox
+from app.schemas.orthophoto import OrthophotoAcquireRequest
+from app.services.aoi_operation_service import AoiOperationService
+from app.services.grb_acquisition_service import GrbAcquisitionService
+from app.services.dhmv_acquisition_service import DhmvAcquisitionService
+from app.services.spw_terrain_service import SpwTerrainService
+from app.services.official_vector_acquisition_service import OfficialVectorAcquisitionService
+from app.services.flood_hazard_acquisition_service import FloodHazardAcquisitionService
+from app.services.thematic_raster_acquisition_service import ThematicRasterAcquisitionService
+from app.services.walous_land_cover_service import WalousLandCoverService
+from app.services.bathymetry_profile_acquisition_service import BathymetryProfileAcquisitionService
+from app.services.mdk_bathymetry_acquisition_service import MdkBathymetryAcquisitionService
+from app.services.job_service import JobService
+from app.services.orthophoto_acquisition_service import OrthophotoAcquisitionService
+
+
+class AoiOperationExecutor:
+    """Execute one bounded partition through an existing governed provider."""
+
+    @staticmethod
+    def execute_next(db, project_id: UUID, operation_id: UUID) -> dict:
+        partition = AoiOperationService.claim_next(db, project_id, operation_id)
+        if partition is None:
+            AoiOperationService._refresh_parent(db, operation_id)
+            return AoiOperationService.read(db, project_id, operation_id)
+        operation = db.get(AoiOperation, operation_id)
+        child = JobService.create_job(db, JobCreate(
+            job_type=f"aoi.{operation.operation_type}.partition",
+            project_id=project_id,
+            parameters_json={
+                "aoi_operation_id": str(operation_id),
+                "partition_id": str(partition.id),
+                "partition_key": partition.partition_key,
+                "provider_key": partition.provider_key,
+                "product_key": partition.product_key,
+            },
+        ))
+        partition = db.get(AoiOperationPartition, partition.id)
+        partition.child_job_id = child.id
+        db.add(partition); db.commit()
+        JobService.mark_running(db, child.id)
+        try:
+            result = AoiOperationExecutor._dispatch(db, project_id, operation, partition)
+            output_id = result.get("output_dataset_id") if isinstance(result, dict) else None
+            JobService.mark_success(db, child.id, result=result, output_dataset_id=UUID(str(output_id)) if output_id else None)
+            return AoiOperationService.complete(db, project_id, operation_id, partition.id, result)
+        except AppError as exc:
+            JobService.mark_failed(db, child.id, exc.message, {"code": exc.code, "details": exc.details})
+            return AoiOperationService.fail(db, project_id, operation_id, partition.id, exc.message, AoiOperationExecutor._retryable(exc), {"code": exc.code, "details": exc.details})
+        except Exception:
+            try:
+                db.rollback()
+                JobService.mark_failed(db, child.id, "Unexpected partition execution error", {"code": "AOI_PARTITION_INTERNAL_ERROR"})
+            finally:
+                AoiOperationService.fail(db, project_id, operation_id, partition.id, "Unexpected partition execution error", True, {"code": "AOI_PARTITION_INTERNAL_ERROR"})
+            raise
+
+    @staticmethod
+    def _dispatch(db, project_id: UUID, operation: AoiOperation, partition: AoiOperationPartition) -> dict:
+        geometry = to_shape(partition.geometry)
+        min_x, min_y, max_x, max_y = geometry.bounds
+        bbox = VectorSelectionBBox(min_x=min_x, min_y=min_y, max_x=max_x, max_y=max_y, crs="EPSG:4326")
+        force_refresh = bool((operation.request_json or {}).get("parameters_json", {}).get("force_refresh", False))
+        if partition.provider_key == "grb":
+            return GrbAcquisitionService.acquire(db, project_id, GrbAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh))
+        if partition.provider_key == "orthophoto":
+            return OrthophotoAcquisitionService.acquire(db, project_id, OrthophotoAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh))
+        if partition.provider_key == "dhmv":
+            return DhmvAcquisitionService.acquire(db, project_id, DhmvAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh))
+        if partition.provider_key == "spw_terrain":
+            return SpwTerrainService.acquire(db, project_id, SpwTerrainAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh))
+        if partition.provider_key == "official_vector":
+            return OfficialVectorAcquisitionService.acquire(db, project_id, OfficialVectorAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh))
+        if partition.provider_key == "flood_hazard":
+            return FloodHazardAcquisitionService.acquire(db, project_id, FloodHazardAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh))
+        if partition.provider_key == "thematic_raster":
+            return ThematicRasterAcquisitionService.acquire(db, project_id, ThematicRasterAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh))
+        if partition.provider_key == "walous":
+            return WalousLandCoverService.acquire(db, project_id, ThematicRasterAcquireRequest(bbox=bbox, area_id=operation.area_id, product_key=partition.product_key, force_refresh=force_refresh))
+        if partition.provider_key == "bathymetry_profiles":
+            return BathymetryProfileAcquisitionService.acquire(db, project_id, BathymetryProfileAcquireRequest(bbox=bbox, area_id=operation.area_id, force_refresh=force_refresh))
+        if partition.provider_key == "mdk_bathymetry":
+            return MdkBathymetryAcquisitionService.acquire(db, project_id, MdkBathymetryAcquireRequest(bbox=bbox, area_id=operation.area_id, force_refresh=force_refresh))
+        raise AppError(code="AOI_PROVIDER_UNSUPPORTED", message="No governed AOI executor is registered for this provider", details={"provider_key": partition.provider_key}, status_code=422)
+
+    @staticmethod
+    def _retryable(error: AppError) -> bool:
+        return error.status_code >= 500 or error.code.endswith(("TIMEOUT", "UNAVAILABLE", "TLS_ERROR"))

backend/app/services/aoi_operation_service.py

@@ -0,0 +1,276 @@
+from __future__ import annotations
+
+from collections import Counter
+from datetime import datetime, timezone
+from hashlib import sha256
+import math
+from uuid import UUID, uuid4
+
+from geoalchemy2.shape import from_shape, to_shape
+from pyproj import Transformer
+from shapely.geometry import MultiPolygon, Polygon, box
+from shapely.ops import transform
+
+from app.core.errors import AppError
+from app.core.config import get_settings
+from app.models import AoiOperation, AoiOperationPartition, Area, Project
+from app.schemas.aoi_operation import AoiOperationCreate
+
+
+class AoiOperationService:
+    MAX_PARTITIONS = 4096
+    _to_metric = Transformer.from_crs("EPSG:4326", "EPSG:31370", always_xy=True)
+    _to_wgs84 = Transformer.from_crs("EPSG:31370", "EPSG:4326", always_xy=True)
+    SCOPE_AREA_NAMES = {
+        "belgium": "Belgium land", "flanders": "Flanders", "wallonia": "Wallonia",
+        "brussels": "Brussels-Capital Region", "belgian_north_sea": "Belgian part of the North Sea",
+        "territorial_sea": "Belgian territorial sea (0-12 nautical miles)",
+        "exclusive_economic_zone": "Belgian exclusive economic zone beyond territorial sea",
+        "continental_shelf": "Belgian continental shelf beyond territorial sea",
+    }
+
+    @staticmethod
+    def create(db, project_id: UUID, payload: AoiOperationCreate) -> dict:
+        if db.get(Project, project_id) is None:
+            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
+        geometry = AoiOperationService._resolve_geometry(db, project_id, payload)
+        if payload.coverage_zone:
+            geometry = AoiOperationService._clip_to_zone(db, project_id, geometry, payload.coverage_zone)
+        geometry = AoiOperationService._as_multipolygon(geometry)
+        metric_geometry = transform(AoiOperationService._to_metric.transform, geometry)
+        partition_side_m = AoiOperationService._partition_side(payload.provider_key, payload.max_partition_side_m)
+        cells = AoiOperationService._partition(metric_geometry, partition_side_m)
+        operation_id = uuid4()
+        now = datetime.now(timezone.utc)
+        operation = AoiOperation(
+            id=operation_id,
+            project_id=project_id,
+            area_id=payload.area_id,
+            operation_type=payload.operation_type,
+            status="queued",
+            geometry=from_shape(geometry, srid=4326),
+            request_json=payload.model_dump(mode="json", exclude_none=True),
+            plan_json={
+                "partition_strategy": "epsg31370_square_grid_intersection_v1",
+                "max_partition_side_m": partition_side_m,
+                "budget_source": "governed_provider_registry" if payload.max_partition_side_m is None else "stricter_operator_override",
+                "partition_count": len(cells),
+                "provider_key": payload.provider_key,
+                "product_key": payload.product_key,
+            },
+            created_at=now,
+        )
+        db.add(operation)
+        for ordinal, cell in enumerate(cells):
+            wgs84 = transform(AoiOperationService._to_wgs84.transform, cell)
+            wgs84 = AoiOperationService._as_multipolygon(wgs84)
+            digest = sha256(wgs84.wkb).hexdigest()[:20]
+            db.add(AoiOperationPartition(
+                id=uuid4(), operation_id=operation_id,
+                partition_key=f"{payload.provider_key}:{payload.product_key}:{ordinal:05d}:{digest}",
+                provider_key=payload.provider_key, product_key=payload.product_key,
+                ordinal=ordinal, status="queued", geometry=from_shape(wgs84, srid=4326),
+                attempt_count=0, max_attempts=payload.max_attempts, created_at=now,
+            ))
+        db.commit()
+        return AoiOperationService.read(db, project_id, operation_id)
+
+    @staticmethod
+    def _clip_to_zone(db, project_id: UUID, geometry, zone: str):
+        area_name = AoiOperationService.SCOPE_AREA_NAMES.get(zone)
+        if area_name is None:
+            raise AppError(code="AOI_COVERAGE_ZONE_UNSUPPORTED", message="Unknown governed coverage zone", details={"coverage_zone": zone}, status_code=422)
+        scope = db.query(Area).filter(Area.project_id == project_id, Area.name == area_name).first()
+        if scope is None:
+            raise AppError(code="AOI_COVERAGE_ZONE_NOT_MATERIALIZED", message="The governed coverage-zone geometry is not persisted in this project", details={"coverage_zone": zone}, status_code=409)
+        clipped = geometry.intersection(to_shape(scope.geometry))
+        if clipped.is_empty:
+            raise AppError(code="AOI_OUTSIDE_PROVIDER_ZONE", message="The AOI does not intersect the provider coverage zone", details={"coverage_zone": zone}, status_code=422)
+        return clipped
+
+    @staticmethod
+    def _as_multipolygon(geometry) -> MultiPolygon:
+        if isinstance(geometry, Polygon):
+            return MultiPolygon([geometry])
+        if isinstance(geometry, MultiPolygon):
+            return geometry
+        polygons = [part for part in getattr(geometry, "geoms", []) if isinstance(part, Polygon)]
+        if not polygons:
+            raise AppError(code="AOI_GEOMETRY_EMPTY", message="AOI contains no polygonal area after clipping", status_code=422)
+        return MultiPolygon(polygons)
+
+    @staticmethod
+    def _partition_side(provider_key: str, requested: float | None) -> float:
+        settings = get_settings()
+        def raster_side(max_side_m: float, max_pixels: int, resolution_m: float) -> float:
+            # Keep every square grid cell within both the provider's spatial
+            # extent limit and its decoded-pixel budget.  The small safety
+            # margin absorbs ceil/edge rounding in the acquisition services.
+            pixel_limited_side = math.sqrt(float(max_pixels)) * float(resolution_m) * 0.99
+            return min(float(max_side_m), pixel_limited_side)
+
+        budgets = {
+            "orthophoto": float(settings.orthophoto_max_side_m),
+            "grb": float(settings.grb_max_side_m),
+            "dhmv": raster_side(
+                settings.dhmv_max_side_m,
+                settings.dhmv_max_pixels,
+                settings.dhmv_resolution_m,
+            ),
+            "spw_terrain": raster_side(
+                settings.spw_terrain_max_side_m,
+                settings.spw_terrain_max_pixels,
+                settings.spw_terrain_analysis_resolution_m,
+            ),
+            "official_vector": 20_000.0,
+            "flood_hazard": raster_side(
+                settings.flood_hazard_max_side_m,
+                settings.flood_hazard_max_pixels,
+                settings.flood_hazard_resolution_m,
+            ),
+            "thematic_raster": raster_side(
+                settings.thematic_raster_max_side_m,
+                settings.thematic_raster_max_pixels,
+                10.0,
+            ),
+            "walous": raster_side(
+                settings.walous_max_side_m,
+                settings.walous_max_pixels,
+                settings.walous_analysis_resolution_m,
+            ),
+            "bathymetry_profiles": 20_000.0,
+            "mdk_bathymetry": 20_000.0,
+        }
+        if provider_key not in budgets:
+            raise AppError(code="AOI_PROVIDER_UNSUPPORTED", message="No governed partition budget is registered for this provider", details={"provider_key": provider_key}, status_code=422)
+        governed = budgets[provider_key]
+        return min(governed, float(requested)) if requested is not None else governed
+
+    @staticmethod
+    def _resolve_geometry(db, project_id: UUID, payload: AoiOperationCreate):
+        if (payload.area_id is None) == (payload.bbox is None):
+            raise AppError(code="AOI_SELECTION_REQUIRED", message="Provide exactly one area_id or bbox", status_code=422)
+        if payload.area_id is not None:
+            area = db.get(Area, payload.area_id)
+            if area is None or area.project_id != project_id:
+                raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
+            return to_shape(area.geometry)
+        bbox = payload.bbox
+        if bbox is None or bbox.crs != "EPSG:4326":
+            raise AppError(code="INVALID_AOI_CRS", message="AOI bbox must use EPSG:4326", status_code=422)
+        return box(bbox.min_x, bbox.min_y, bbox.max_x, bbox.max_y)
+
+    @staticmethod
+    def _partition(geometry, side_m: float) -> list:
+        min_x, min_y, max_x, max_y = geometry.bounds
+        columns = max(1, math.ceil((max_x - min_x) / side_m))
+        rows = max(1, math.ceil((max_y - min_y) / side_m))
+        if columns * rows > AoiOperationService.MAX_PARTITIONS:
+            raise AppError(code="AOI_PARTITION_LIMIT_EXCEEDED", message="AOI requires too many bounded partitions", details={"candidate_count": columns * rows, "max_partitions": AoiOperationService.MAX_PARTITIONS}, status_code=422)
+        partitions = []
+        for row in range(rows):
+            for column in range(columns):
+                clipped = geometry.intersection(box(min_x + column * side_m, min_y + row * side_m, min(min_x + (column + 1) * side_m, max_x), min(min_y + (row + 1) * side_m, max_y)))
+                if not clipped.is_empty and clipped.area > 0:
+                    partitions.append(clipped)
+        return partitions
+
+    @staticmethod
+    def read(db, project_id: UUID, operation_id: UUID) -> dict:
+        operation = db.get(AoiOperation, operation_id)
+        if operation is None or operation.project_id != project_id:
+            raise AppError(code="AOI_OPERATION_NOT_FOUND", message="AOI operation not found", status_code=404)
+        partitions = db.query(AoiOperationPartition).filter(AoiOperationPartition.operation_id == operation_id).order_by(AoiOperationPartition.ordinal).all()
+        counts = Counter(partition.status for partition in partitions)
+        complete = counts["success"] + counts["skipped"]
+        return {
+            "id": operation.id, "project_id": operation.project_id, "area_id": operation.area_id,
+            "parent_job_id": operation.parent_job_id, "operation_type": operation.operation_type,
+            "status": operation.status, "request_json": operation.request_json, "plan_json": operation.plan_json,
+            "result_json": operation.result_json, "error_message": operation.error_message,
+            "progress": round(complete / len(partitions), 6) if partitions else 0.0,
+            "partition_counts": dict(counts), "partitions": partitions,
+            "created_at": operation.created_at, "started_at": operation.started_at, "finished_at": operation.finished_at,
+        }
+
+    @staticmethod
+    def list(db, project_id: UUID, limit: int = 50) -> dict:
+        rows = db.query(AoiOperation).filter(AoiOperation.project_id == project_id).order_by(AoiOperation.created_at.desc()).limit(limit).all()
+        return {"items": [AoiOperationService.read(db, project_id, row.id) for row in rows], "total": len(rows)}
+
+    @staticmethod
+    def claim_next(db, project_id: UUID, operation_id: UUID):
+        operation = db.get(AoiOperation, operation_id)
+        if operation is None or operation.project_id != project_id:
+            raise AppError(code="AOI_OPERATION_NOT_FOUND", message="AOI operation not found", status_code=404)
+        partition = db.query(AoiOperationPartition).filter(AoiOperationPartition.operation_id == operation_id, AoiOperationPartition.status == "queued").order_by(AoiOperationPartition.ordinal).with_for_update(skip_locked=True).first()
+        if partition is None:
+            return None
+        now = datetime.now(timezone.utc)
+        partition.status = "running"; partition.started_at = now; partition.attempt_count += 1; partition.error_message = None
+        operation.status = "running"; operation.started_at = operation.started_at or now
+        db.add(partition); db.add(operation); db.commit(); db.refresh(partition)
+        return partition
+
+    @staticmethod
+    def checkpoint(db, project_id: UUID, operation_id: UUID, partition_id: UUID, checkpoint: dict):
+        partition = AoiOperationService._partition_row(db, project_id, operation_id, partition_id)
+        if partition.status != "running":
+            raise AppError(code="AOI_PARTITION_NOT_RUNNING", message="Only a running partition can be checkpointed", status_code=409)
+        partition.checkpoint_json = checkpoint; db.add(partition); db.commit(); db.refresh(partition)
+        return partition
+
+    @staticmethod
+    def complete(db, project_id: UUID, operation_id: UUID, partition_id: UUID, result: dict, skipped: bool = False):
+        partition = AoiOperationService._partition_row(db, project_id, operation_id, partition_id)
+        if partition.status == "success" or partition.status == "skipped":
+            return AoiOperationService.read(db, project_id, operation_id)
+        if partition.status != "running":
+            raise AppError(code="AOI_PARTITION_NOT_RUNNING", message="Only a running partition can complete", status_code=409)
+        partition.status = "skipped" if skipped else "success"; partition.result_json = result; partition.finished_at = datetime.now(timezone.utc)
+        db.add(partition); db.commit(); AoiOperationService._refresh_parent(db, operation_id)
+        return AoiOperationService.read(db, project_id, operation_id)
+
+    @staticmethod
+    def fail(db, project_id: UUID, operation_id: UUID, partition_id: UUID, message: str, retryable: bool, details: dict):
+        partition = AoiOperationService._partition_row(db, project_id, operation_id, partition_id)
+        partition.error_message = message; partition.result_json = {"details": details}
+        partition.status = "queued" if retryable and partition.attempt_count < partition.max_attempts else "failed"
+        partition.finished_at = None if partition.status == "queued" else datetime.now(timezone.utc)
+        db.add(partition); db.commit(); AoiOperationService._refresh_parent(db, operation_id)
+        return AoiOperationService.read(db, project_id, operation_id)
+
+    @staticmethod
+    def _partition_row(db, project_id, operation_id, partition_id):
+        operation = db.get(AoiOperation, operation_id); partition = db.get(AoiOperationPartition, partition_id)
+        if operation is None or operation.project_id != project_id or partition is None or partition.operation_id != operation_id:
+            raise AppError(code="AOI_PARTITION_NOT_FOUND", message="AOI partition not found", status_code=404)
+        return partition
+
+    @staticmethod
+    def _refresh_parent(db, operation_id):
+        operation = db.get(AoiOperation, operation_id)
+        partitions = db.query(AoiOperationPartition).filter(AoiOperationPartition.operation_id == operation_id).order_by(AoiOperationPartition.ordinal).all()
+        statuses = [partition.status for partition in partitions]
+        output_dataset_ids = []
+        for partition in partitions:
+            output_id = (partition.result_json or {}).get("output_dataset_id") if isinstance(partition.result_json, dict) else None
+            if output_id and str(output_id) not in output_dataset_ids:
+                output_dataset_ids.append(str(output_id))
+        operation.result_json = {
+            "partition_count": len(partitions),
+            "completed_partition_count": sum(status in {"success", "skipped"} for status in statuses),
+            "failed_partition_count": statuses.count("failed"),
+            "output_dataset_ids": output_dataset_ids,
+            "merge_contract": "source_aware_spatial_union",
+            "vector_deduplication": "source_feature_id_then_geometry",
+            "raster_deduplication": "governed_mosaic_grid",
+            "complete_coverage": bool(statuses) and all(status in {"success", "skipped"} for status in statuses),
+        }
+        now = datetime.now(timezone.utc)
+        if statuses and all(status in {"success", "skipped"} for status in statuses):
+            operation.status = "success"; operation.finished_at = now; operation.error_message = None
+        elif "failed" in statuses and not any(status in {"queued", "running"} for status in statuses):
+            operation.status = "partial" if any(status in {"success", "skipped"} for status in statuses) else "failed"; operation.finished_at = now
+            operation.error_message = "One or more bounded source partitions failed; inspect partition evidence."
+        db.add(operation); db.commit()

backend/app/services/aoi_operation_worker.py

@@ -0,0 +1,38 @@
+from __future__ import annotations
+
+import asyncio
+import logging
+
+from app.db.session import SessionLocal
+from app.models import AoiOperation
+from app.services.aoi_operation_executor import AoiOperationExecutor
+
+
+logger = logging.getLogger("geointel.aoi_worker")
+
+
+class AoiOperationWorker:
+    @staticmethod
+    def run_once() -> int:
+        db = SessionLocal()
+        try:
+            rows = db.query(AoiOperation).filter(AoiOperation.status.in_(("queued", "running"))).order_by(AoiOperation.created_at).limit(10).all()
+            for operation in rows:
+                try:
+                    AoiOperationExecutor.execute_next(db, operation.project_id, operation.id)
+                except Exception:
+                    db.rollback()
+                    logger.exception("AOI partition execution failed operation_id=%s", operation.id)
+            return len(rows)
+        finally:
+            db.close()
+
+    @staticmethod
+    async def run(stop_event: asyncio.Event, poll_seconds: float) -> None:
+        while not stop_event.is_set():
+            processed = await asyncio.to_thread(AoiOperationWorker.run_once)
+            if processed == 0:
+                try:
+                    await asyncio.wait_for(stop_event.wait(), timeout=poll_seconds)
+                except TimeoutError:
+                    pass

backend/app/services/coverage_registry_service.py

@@ -452,10 +452,12 @@ class CoverageRegistryService:
         theme: str,
         zone: str,
         selection: Any,
-    ) -> list[Dataset]:
+    ) -> tuple[list[Dataset], bool]:
         if definition.operational_themes and theme not in definition.operational_themes:
-            return []
+            return [], False
         matches: list[Dataset] = []
+        bounded_scopes: list[Any] = []
+        zone_scoped_materialization = False
         for dataset in datasets:
             if dataset.status != "ready" or dataset.source_name not in definition.materialized_source_names:
                 continue
@@ -484,13 +486,21 @@ class CoverageRegistryService:
                     acquired_scope = box(*(float(value) for value in acquired_bbox))
                 except (TypeError, ValueError):
                     continue
-                if not acquired_scope.is_valid or not acquired_scope.covers(selection):
+                if not acquired_scope.is_valid or not acquired_scope.intersects(selection):
                     continue
+                bounded_scopes.append(acquired_scope)
+            elif definition.contract.acquisition_mode == "bounded_api" and coverage_zones:
+                zone_scoped_materialization = zone in coverage_zones or "belgium" in coverage_zones
             layer_matches = not layer_names or dataset.reference_layer_name in layer_names
             zone_matches = not coverage_zones or zone in coverage_zones or "belgium" in coverage_zones
             if layer_matches and zone_matches:
                 matches.append(dataset)
-        return matches
+        if not matches:
+            return [], False
+        fully_covered = True
+        if definition.contract.acquisition_mode == "bounded_api":
+            fully_covered = zone_scoped_materialization or (bool(bounded_scopes) and unary_union(bounded_scopes).covers(selection))
+        return matches, fully_covered
 
     @staticmethod
     def _resolve_item(
@@ -516,10 +526,11 @@ class CoverageRegistryService:
             )
 
         materialized: list[Dataset] = []
+        evidence: list[dict[str, Any]] = []
         source_statuses: list[str] = []
         limitations: list[str] = []
         for definition in definitions:
-            matches = CoverageRegistryService._matching_datasets(
+            matches, fully_covered = CoverageRegistryService._matching_datasets(
                 datasets,
                 definition,
                 theme,
@@ -527,8 +538,28 @@ class CoverageRegistryService:
                 selection,
             )
             materialized.extend(matches)
-            if matches:
+            for dataset in matches:
+                metadata = dataset.source_metadata if isinstance(getattr(dataset, "source_metadata", None), dict) else {}
+                observed_at = getattr(dataset, "observed_at", None)
+                published_at = metadata.get("published_at") or metadata.get("publication_date") or metadata.get("published_on")
+                evidence.append({
+                    "dataset_id": dataset.id,
+                    "source_name": str(dataset.source_name or definition.contract.source_name),
+                    "authority_level": definition.contract.authority_level,
+                    "source_version": getattr(dataset, "source_version", None),
+                    "observed_at": observed_at.isoformat() if hasattr(observed_at, "isoformat") else (str(observed_at) if observed_at else None),
+                    "published_at": str(published_at) if published_at else None,
+                    "crs": getattr(dataset, "crs", None) or metadata.get("source_crs"),
+                    "resolution": getattr(dataset, "resolution_json", None),
+                    "coverage_bbox_epsg4326": metadata.get("bbox_epsg4326"),
+                    "attribution": metadata.get("attribution") or definition.contract.attribution,
+                    "license_note": metadata.get("license_note") or definition.contract.license_note,
+                    "checksum_sha256": getattr(dataset, "checksum_sha256", None),
+                })
+            if matches and fully_covered:
                 source_statuses.append("operational")
+            elif matches:
+                source_statuses.append("partial")
             elif (
                 definition.contract.integration_status == "operational"
                 and (not definition.operational_themes or theme in definition.operational_themes)
@@ -549,6 +580,7 @@ class CoverageRegistryService:
             status=best_status,
             source_names=[definition.contract.source_name for definition in definitions],
             materialized_dataset_ids=list(dict.fromkeys(dataset.id for dataset in materialized)),
+            evidence=list({str(item["dataset_id"]): item for item in evidence}.values()),
             limitation_message=" ".join(dict.fromkeys(limitations)),
         )
 

backend/app/services/detection_service.py

@@ -15,7 +15,7 @@ from sqlalchemy import func
 from app.core.config import Settings, get_settings
 from app.core.errors import AppError
 from app.core.request_context import get_request_id
-from app.models import AnalysisRun, Dataset, Detection, Job, Project, VectorFeature
+from app.models import AnalysisRun, Area, Dataset, Detection, Job, Project, VectorFeature
 from app.schemas.detection import DetectionListResponse, DetectionRead, DetectionRunListResponse, DetectionRunRead, DetectionRunResponse
 from app.services.detection_georeferencing import pixel_bbox_to_epsg4326_polygon
 from app.services.detection_qa_service import DetectionQaService
@@ -90,6 +90,12 @@ class DetectionService:
                 message="Configured YOLO inference requires an existing raster tile manifest path",
                 status_code=400,
             )
+        requested_classes = {DetectionService._canonical_class_name(value) for value in (class_filter or [])}
+        unsupported_classes = sorted(requested_classes - set(model.supported_classes))
+        if unsupported_classes:
+            raise AppError(code="DETECTION_CLASS_NOT_VALIDATED", message="The selected model is not validated for one or more requested classes", details={"unsupported_classes": unsupported_classes, "supported_classes": model.supported_classes}, status_code=422)
+        if model.model_id == resolved_settings.yolo_model_id and resolved_settings.yolo_enforce_validation_scope:
+            DetectionService._validate_model_area_scope(db, dataset, resolved_settings)
 
         run_parameters = {
             "model_id": model.model_id,
@@ -214,6 +220,18 @@ class DetectionService:
         raise AppError(code="DETECTION_MODEL_UNAVAILABLE", message="Detection model is unavailable", status_code=503)
 
     @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):
+            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},
+                status_code=422,
+            )
+    @staticmethod
     def _fail_run_after_exception(db, analysis_run: AnalysisRun, job: Job, exc: Exception, fallback_code: str) -> None:
         try:
             db.rollback()

backend/app/services/model_registry_service.py

@@ -10,6 +10,7 @@ from app.services.segmentation_adapter import (
     YoloSegmentationAdapter,
 )
 from app.services.yolo_adapter import YoloDetectionAdapter
+from app.core.errors import AppError
 
 
 class ModelRegistryService:
@@ -224,16 +225,24 @@ class ModelRegistryService:
             elif not model_path.exists() or not model_path.is_file():
                 limitation = "YOLO_MODEL_PATH does not point to an existing local model file. GeoIntel will not download model weights automatically."
             else:
-                configured = True
-                status = "configured"
-                limitation = "Configured for local YOLO inference over an existing raster tile manifest."
+                try:
+                    validate_runtime = getattr(yolo_adapter_class, "validate_runtime", None)
+                    if validate_runtime is not None:
+                        yolo_adapter_class(settings).validate_runtime()
+                except AppError as exc:
+                    status = "accelerator_unavailable"
+                    limitation = exc.message
+                else:
+                    configured = True
+                    status = "configured"
+                    limitation = "Configured for local YOLO inference over an existing raster tile manifest within its validated area scope."
 
         return DetectionModelCapability(
             model_id=settings.yolo_model_id,
             display_name=settings.yolo_model_display_name,
             framework="ultralytics/pytorch",
             task_type="object_detection",
-            supported_classes=["building", "road", "water", "landuse"],
+            supported_classes=[value.strip().lower() for value in settings.yolo_model_classes.split(",") if value.strip()],
             configured=configured,
             status=status,
             limitation_message=limitation,

backend/app/services/runtime_reconciliation_service.py

@@ -5,13 +5,15 @@ from datetime import datetime, timezone
 
 from sqlalchemy.orm import Session
 
-from app.models import AnalysisRun, Job
+from app.models import AoiOperation, AoiOperationPartition, AnalysisRun, Job
 
 
 @dataclass(frozen=True)
 class ReconciliationResult:
     interrupted_jobs: int
     interrupted_analysis_runs: int
+    resumed_aoi_partitions: int
+    exhausted_aoi_partitions: int
 
 
 class RuntimeReconciliationService:
@@ -51,8 +53,43 @@ class RuntimeReconciliationService:
                 synchronize_session=False,
             )
         )
+        resumed_aoi_partitions = (
+            db.query(AoiOperationPartition)
+            .filter(
+                AoiOperationPartition.status == "running",
+                AoiOperationPartition.attempt_count < AoiOperationPartition.max_attempts,
+            )
+            .update(
+                {
+                    AoiOperationPartition.status: "queued",
+                    AoiOperationPartition.error_message: RuntimeReconciliationService.ERROR_MESSAGE,
+                    AoiOperationPartition.started_at: None,
+                },
+                synchronize_session=False,
+            )
+        )
+        exhausted_aoi_partitions = (
+            db.query(AoiOperationPartition)
+            .filter(
+                AoiOperationPartition.status == "running",
+                AoiOperationPartition.attempt_count >= AoiOperationPartition.max_attempts,
+            )
+            .update(
+                {
+                    AoiOperationPartition.status: "failed",
+                    AoiOperationPartition.finished_at: resolved_finished_at,
+                    AoiOperationPartition.error_message: RuntimeReconciliationService.ERROR_MESSAGE,
+                },
+                synchronize_session=False,
+            )
+        )
+        db.query(AoiOperation).filter(AoiOperation.status == "running").update(
+            {AoiOperation.status: "queued"}, synchronize_session=False
+        )
         db.commit()
         return ReconciliationResult(
             interrupted_jobs=interrupted_jobs,
             interrupted_analysis_runs=interrupted_analysis_runs,
+            resumed_aoi_partitions=resumed_aoi_partitions,
+            exhausted_aoi_partitions=exhausted_aoi_partitions,
         )

backend/app/services/yolo_adapter.py

@@ -38,6 +38,8 @@ class YoloDetectionAdapter:
                 status_code=503,
             )
 
+        self.validate_runtime()
+
         try:
             from ultralytics import YOLO
         except ImportError as exc:
@@ -57,6 +59,32 @@ class YoloDetectionAdapter:
                 status_code=503,
             ) from exc
 
+    def validate_runtime(self) -> None:
+        if not self.settings.yolo_require_cuda:
+            return
+        try:
+            import torch
+        except Exception as exc:
+            raise AppError(
+                code="DETECTION_ACCELERATOR_UNAVAILABLE",
+                message="NVIDIA CUDA is required for configured YOLO inference, but PyTorch is not importable.",
+                status_code=503,
+            ) from exc
+        if not torch.cuda.is_available():
+            raise AppError(
+                code="DETECTION_ACCELERATOR_UNAVAILABLE",
+                message="NVIDIA CUDA is required for configured YOLO inference, but no CUDA device is available.",
+                details={"configured_device": self.settings.yolo_device},
+                status_code=503,
+            )
+        if not str(self.settings.yolo_device).lower().startswith(("cuda", "0", "1", "2", "3")):
+            raise AppError(
+                code="DETECTION_ACCELERATOR_MISCONFIGURED",
+                message="NVIDIA CUDA is required, but YOLO_DEVICE does not select a CUDA device.",
+                details={"configured_device": self.settings.yolo_device},
+                status_code=503,
+            )
+
     def predict_tile(self, model, tile_path: Path, confidence_threshold: float) -> list[dict[str, Any]]:
         if not tile_path.exists() or not tile_path.is_file():
             raise AppError(

backend/app/services/yolo_preflight_service.py

@@ -38,6 +38,7 @@ class YoloPreflightService:
             "checks": {
                 "enabled": resolved_settings.yolo_enabled,
                 "dependencies_available": None,
+                "accelerator_ready": None,
                 "model_path_set": None,
                 "model_file_exists": None,
                 "model_load_requested": check_model_load,
@@ -70,6 +71,21 @@ class YoloPreflightService:
             result["message"] = "YOLO dependencies are not installed. Install backend optional extras with geointel-backend[ai]."
             return result
 
+        if not assume_dependencies:
+            try:
+                adapter = yolo_adapter_class(resolved_settings)
+                validate_runtime = getattr(adapter, "validate_runtime", None)
+                if validate_runtime is not None:
+                    validate_runtime()
+            except AppError as exc:
+                result["checks"]["accelerator_ready"] = False
+                result["status"] = "accelerator_unavailable"
+                result["message"] = exc.message
+                result["error_code"] = exc.code
+                result["details"] = exc.details
+                return result
+        result["checks"]["accelerator_ready"] = True
+
         result["checks"]["model_path_set"] = bool(resolved_settings.yolo_model_path)
         if not resolved_settings.yolo_model_path:
             result["message"] = "YOLO_MODEL_PATH is not set. GeoIntel will not download model weights automatically."
@@ -143,6 +159,8 @@ class YoloPreflightService:
             "torch_version": YoloPreflightService._package_version("torch"),
             "ultralytics_version": YoloPreflightService._package_version("ultralytics"),
             "cuda_available": None,
+            "configured_device": settings.yolo_device,
+            "cuda_required": settings.yolo_require_cuda,
         }
 
     @staticmethod