geointel/backend/app/services/dataset_service.py

from __future__ import annotations

import json
import pathlib
import re
from dataclasses import dataclass
from hashlib import sha256
from datetime import datetime, timezone
from math import isfinite
from pathlib import Path
from typing import Any
from uuid import UUID
import uuid

from fastapi import UploadFile
from shapely.geometry import MultiPoint, shape
from sqlalchemy.orm import Session

from app.core.errors import AppError
from app.models import Area, Dataset, DatasetVersion, Project
from app.services.data_contract_validation import (
    ContractKind,
    DataAssetValidationInput,
    GeometryRecord,
    LineageEvidence,
    LineageStatus,
    ProvenanceStatus,
    QuarantineStatus,
    RASTER_GEOTIFF_CONTRACT_KEY,
    RASTER_GEOTIFF_CONTRACT_VERSION,
    TransformationEvidence,
    VECTOR_GEOJSON_CONTRACT_KEY,
    VECTOR_GEOJSON_CONTRACT_VERSION,
    ValidationIssue,
    ValidationReport,
    ValidationStatus,
    build_raster_ingest_input,
    build_vector_ingest_input,
    validate_registered_asset,
)
from app.services.data_quarantine_service import DataQuarantineService
from app.schemas.dataset import (
    DatasetCreateResponse,
    DatasetStorageResponse,
    DatasetTemporalUpdate,
    DatasetVectorSummary,
    DatasetVersionRead,
)
from app.services.geojson_service import parse_geojson_payload, load_dataset_text
from app.services.raster_service import extract_raster_metadata
from app.services.source_registry_service import SourceRegistryService
from app.services.storage_service import StorageService
from app.services.vector_feature_service import VectorFeatureService


_CHECKSUM_SHA256 = re.compile(r"^[0-9a-f]{64}$")


@dataclass(frozen=True)
class _SourceVectorSchema:
    """Server-owned vector schema expectations attached to a source registry row."""

    source_key: str
    expected_geometry_types: frozenset[str]
    required_attributes: tuple[str, ...]

    @classmethod
    def from_source(cls, source: Any) -> "_SourceVectorSchema":
        geometry_values = getattr(source, "expected_geometry_types_json", ())
        expected_geometry_types = (
            frozenset(str(value).strip() for value in geometry_values if str(value).strip())
            if isinstance(geometry_values, (list, tuple, set))
            else frozenset()
        )
        attributes = getattr(source, "expected_attributes_json", {})
        required_values = attributes.get("required") if isinstance(attributes, dict) else ()
        if isinstance(required_values, str):
            required_values = (required_values,)
        required_attributes = (
            tuple(sorted({str(value).strip() for value in required_values if str(value).strip()}))
            if isinstance(required_values, (list, tuple, set))
            else ()
        )
        return cls(
            source_key=str(getattr(source, "source_key", "") or "").strip().lower(),
            expected_geometry_types=expected_geometry_types,
            required_attributes=required_attributes,
        )

    def to_metadata(self, *, checked_feature_count: int) -> dict[str, Any]:
        return {
            "status": "passed",
            "source_key": self.source_key,
            "expected_geometry_types": sorted(self.expected_geometry_types),
            "required_attributes": list(self.required_attributes),
            "checked_feature_count": checked_feature_count,
        }


def _feature_source_identifier(feature: dict[str, Any], properties: dict[str, Any]) -> Any:
    """Return the source identity under the GeoJSON and registry conventions."""

    return feature.get("id") or properties.get("id") or properties.get("source_feature_id")


def _validate_vector_feature_source_schema(
    *,
    feature: dict[str, Any],
    properties: dict[str, Any],
    geometry: Any,
    schema: _SourceVectorSchema,
    feature_context: str,
) -> None:
    """Fail closed when a source-specific vector expectation is violated."""

    if schema.expected_geometry_types and geometry.geom_type not in schema.expected_geometry_types:
        raise AppError(
            code="SOURCE_SCHEMA_GEOMETRY_TYPE_NOT_ALLOWED",
            message=(
                f"{feature_context} has geometry type {geometry.geom_type}, which is not "
                f"allowed by source registry {schema.source_key or 'unknown'}"
            ),
            details={
                "source_key": schema.source_key,
                "expected_geometry_types": sorted(schema.expected_geometry_types),
                "observed_geometry_type": geometry.geom_type,
            },
            status_code=400,
        )
    for attribute in schema.required_attributes:
        value = _feature_source_identifier(feature, properties) if attribute == "id" else properties.get(attribute)
        if value is None or (isinstance(value, str) and not value.strip()):
            raise AppError(
                code="SOURCE_SCHEMA_REQUIRED_ATTRIBUTE_MISSING",
                message=(
                    f"{feature_context} is missing required source attribute {attribute!r} "
                    f"for registry {schema.source_key or 'unknown'}"
                ),
                details={"source_key": schema.source_key, "required_attribute": attribute},
                status_code=400,
            )


@dataclass(frozen=True)
class _PartitionedVectorAudit:
    """Aggregate evidence from a full per-feature partition audit.

    The importer materializes one GeoJSON partition at a time because the
    current parser is ``json.loads`` based.  It never materializes every
    regional partition or every regional Shapely geometry at once.
    """

    feature_count: int
    geometry_types: tuple[str, ...]
    bounds_json: dict[str, float]
    partition_checksums_sha256: dict[str, str]
    source_schema_validation: dict[str, Any]
    representative_record: GeometryRecord

    def to_metadata(self) -> dict[str, Any]:
        return {
            "feature_count": self.feature_count,
            "geometry_types": list(self.geometry_types),
            "bounds_json": dict(self.bounds_json),
            "partition_checksums_sha256": dict(self.partition_checksums_sha256),
            "source_schema_validation": dict(self.source_schema_validation),
            "validation_mode": "partition_bounded_per_feature_with_aggregate_contract_record",
        }


class _PartitionedGeoJsonRecords:
    """Perform a full, partition-bounded feature audit across GeoJSON partitions.

    The generic vector contract materializes its supplied geometry records. A
    regional artifact can contain hundreds of thousands of features, so this
    class validates one materialized partition at a time and emits a compact
    aggregate record for the generic source/checksum/CRS/bounds contract.
    Memory is bounded to the largest single partition, not to one feature.
    """

    def __init__(
        self,
        partition_paths: list[str | Path],
        *,
        expected_feature_count: int,
        declared_partition_checksums: dict[str, Any] | None,
        source_schema: _SourceVectorSchema | None = None,
    ) -> None:
        self._partition_paths = tuple(Path(path) for path in partition_paths)
        self._expected_feature_count = expected_feature_count
        self._declared_checksums = declared_partition_checksums
        self._source_schema = source_schema or _SourceVectorSchema(
            source_key="",
            expected_geometry_types=frozenset(),
            required_attributes=(),
        )

    def audit(self) -> _PartitionedVectorAudit:
        declared_checksums = self._validated_declared_checksums()
        observed_checksums: dict[str, str] = {}
        feature_count = 0
        source_feature_ids: set[str] = set()
        geometry_types: set[str] = set()
        min_x: float | None = None
        min_y: float | None = None
        max_x: float | None = None
        max_y: float | None = None
        for partition_path in self._partition_paths:
            try:
                raw = partition_path.read_bytes()
                payload = json.loads(raw.decode("utf-8"))
            except (OSError, UnicodeDecodeError, json.JSONDecodeError) as exc:
                raise AppError(
                    code="INVALID_GEOJSON_PARTITION",
                    message=f"Could not read GeoJSON partition {partition_path.name}",
                    status_code=400,
                ) from exc
            features = payload.get("features") if isinstance(payload, dict) else None
            if not isinstance(payload, dict) or payload.get("type") != "FeatureCollection" or not isinstance(features, list):
                raise AppError(
                    code="INVALID_GEOJSON_PARTITION",
                    message=f"GeoJSON partition {partition_path.name} must be a FeatureCollection",
                    status_code=400,
                )
            observed_checksum = sha256(raw).hexdigest()
            expected_checksum = declared_checksums[partition_path.name]
            if observed_checksum != expected_checksum:
                raise AppError(
                    code="PARTITION_CHECKSUM_MISMATCH",
                    message=(
                        f"Checksum for partition {partition_path.name} does not match "
                        "the governed acquisition manifest."
                    ),
                    status_code=400,
                )
            observed_checksums[partition_path.name] = observed_checksum
            for index, feature in enumerate(features):
                if not isinstance(feature, dict):
                    raise AppError(
                        code="INVALID_GEOJSON_PARTITION",
                        message=f"Feature {index} in {partition_path.name} must be an object",
                        status_code=400,
                    )
                properties = feature.get("properties") if isinstance(feature.get("properties"), dict) else {}
                source_feature_id = _feature_source_identifier(feature, properties)
                if source_feature_id is not None:
                    normalized_id = str(source_feature_id).strip()
                    if normalized_id:
                        if normalized_id in source_feature_ids:
                            raise AppError(
                                code="DUPLICATE_SOURCE_FEATURE",
                                message=(
                                    f"Duplicate source feature {normalized_id} across regional partitions"
                                ),
                                status_code=400,
                            )
                        source_feature_ids.add(normalized_id)
                try:
                    geometry = shape(feature.get("geometry"))
                except Exception as exc:
                    raise AppError(
                        code="GEOMETRY_PARSE_FAILED",
                        message=f"Feature {index} in {partition_path.name} has invalid GeoJSON geometry",
                        status_code=400,
                    ) from exc
                if geometry.is_empty:
                    raise AppError(
                        code="GEOMETRY_EMPTY",
                        message=f"Feature {index} in {partition_path.name} has an empty geometry",
                        status_code=400,
                    )
                if not geometry.is_valid:
                    raise AppError(
                        code="GEOMETRY_INVALID",
                        message=(
                            f"Feature {index} in {partition_path.name} is invalid; "
                            "partitioned ingestion never silently repairs geometry"
                        ),
                        status_code=400,
                    )
                _validate_vector_feature_source_schema(
                    feature=feature,
                    properties=properties,
                    geometry=geometry,
                    schema=self._source_schema,
                    feature_context=f"Feature {index} in {partition_path.name}",
                )
                feature_bounds = geometry.bounds
                if not all(isfinite(value) for value in feature_bounds):
                    raise AppError(
                        code="GEOMETRY_BOUNDS_INVALID",
                        message=f"Feature {index} in {partition_path.name} has non-finite bounds",
                        status_code=400,
                    )
                geometry_types.add(geometry.geom_type)
                min_x = feature_bounds[0] if min_x is None else min(min_x, feature_bounds[0])
                min_y = feature_bounds[1] if min_y is None else min(min_y, feature_bounds[1])
                max_x = feature_bounds[2] if max_x is None else max(max_x, feature_bounds[2])
                max_y = feature_bounds[3] if max_y is None else max(max_y, feature_bounds[3])
                feature_count += 1

        if feature_count != self._expected_feature_count:
            raise AppError(
                code="PARTITION_FEATURE_COUNT_MISMATCH",
                message=(
                    f"Regional artifact declares {self._expected_feature_count} features but "
                    f"partitions contain {feature_count} features"
                ),
                status_code=400,
            )
        if None in {min_x, min_y, max_x, max_y}:  # pragma: no cover - feature-count invariant above
            raise AppError(
                code="VECTOR_FEATURES_REQUIRED",
                message="Partitioned vector artifact has no geometry records.",
                status_code=400,
            )
        bounds_json = {
            "min_x": float(min_x),
            "min_y": float(min_y),
            "max_x": float(max_x),
            "max_y": float(max_y),
        }
        # A MultiPoint envelope is validation evidence only, not a replacement
        # for persisted source features. It gives the generic contract the
        # audited aggregate bounds without retaining all Shapely objects.
        representative_geometry = MultiPoint(
            [
                (bounds_json["min_x"], bounds_json["min_y"]),
                (bounds_json["max_x"], bounds_json["min_y"]),
                (bounds_json["max_x"], bounds_json["max_y"]),
                (bounds_json["min_x"], bounds_json["max_y"]),
            ]
        )
        return _PartitionedVectorAudit(
            feature_count=feature_count,
            geometry_types=tuple(sorted(geometry_types)),
            bounds_json=bounds_json,
            partition_checksums_sha256=dict(sorted(observed_checksums.items())),
            source_schema_validation=self._source_schema.to_metadata(checked_feature_count=feature_count),
            representative_record=GeometryRecord(
                geometry=representative_geometry,
                properties={"partitioned_geometry_audit": True},
                identifier="partitioned-geometry-audit",
            ),
        )

    def _validated_declared_checksums(self) -> dict[str, str]:
        """Require an exact filename-to-SHA256 manifest for every partition.

        A list of checksum values is insufficient: it cannot establish which
        municipality/source partition produced which persisted feature set.
        The explicit map is also retained with the aggregate audit evidence.
        """

        if not isinstance(self._declared_checksums, dict) or not self._declared_checksums:
            raise AppError(
                code="PARTITION_CHECKSUM_MANIFEST_REQUIRED",
                message="Partitioned ingestion requires a non-empty filename-to-checksum manifest.",
                status_code=400,
            )
        partition_names = [path.name for path in self._partition_paths]
        if len(set(partition_names)) != len(partition_names):
            raise AppError(
                code="DUPLICATE_PARTITION_IDENTITY",
                message="Partitioned ingestion requires unique partition filenames.",
                status_code=400,
            )
        declared = {
            str(key): str(value).strip().lower()
            for key, value in self._declared_checksums.items()
        }
        if len(declared) != len(partition_names) or set(declared) != set(partition_names):
            raise AppError(
                code="PARTITION_CHECKSUM_MANIFEST_MISMATCH",
                message="Partition checksum manifest must contain exactly one entry for each partition filename.",
                details={
                    "expected_partition_filenames": sorted(partition_names),
                    "declared_partition_filenames": sorted(declared),
                },
                status_code=400,
            )
        invalid = sorted(name for name, checksum in declared.items() if not _CHECKSUM_SHA256.fullmatch(checksum))
        if invalid:
            raise AppError(
                code="PARTITION_CHECKSUM_INVALID",
                message="Partition checksum manifest contains a non-SHA256 value.",
                details={"partition_filenames": invalid},
                status_code=400,
            )
        return declared


class DatasetService:
    VECTOR_EXTENSIONS = {".geojson", ".json"}
    RASTER_EXTENSIONS = {".tif", ".tiff", ".geotiff"}
    VECTOR_TYPES = {"vector", "geojson"}
    RASTER_TYPES = {"raster", "tif", "tiff", "geotiff"}
    VALID_DATASET_ROLES = {"source", "derived", "reference"}
    VALID_TEMPORAL_GRANULARITIES = {"snapshot", "day", "month", "year", "period"}
    CANONICAL_VECTOR_CRS = "EPSG:4326"

    @staticmethod
    def _registry_persistence_available(db: Session) -> bool:
        """Return true only for real ORM-backed ingestion transactions.

        Production request handling always supplies a SQLAlchemy Session. The
        narrow fallback keeps historical lightweight unit fakes (which predate
        the registry tables) isolated; it cannot bypass the database-backed
        production import path.
        """
        return callable(getattr(db, "query", None))

    @staticmethod
    def _stable_hash(payload: Any) -> str:
        return sha256(
            json.dumps(payload, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8")
        ).hexdigest()

    @classmethod
    def _canonical_vector_storage_bytes(cls, payload: dict[str, Any]) -> bytes:
        """Serialize the consumable GeoJSON representation deterministically.

        ``VectorFeatureService.canonicalize_geojson_payload`` is the one
        place that transforms source coordinates to EPSG:4326.  This helper
        makes the exact result of that transform the persisted, checksummed
        dataset artifact too; it must never remain merely an in-memory view.
        """

        return json.dumps(
            payload,
            ensure_ascii=False,
            sort_keys=True,
            separators=(",", ":"),
        ).encode("utf-8")

    @classmethod
    def _vector_storage_requires_canonicalization(cls, source_crs: str | None) -> bool:
        """Return whether the source file cannot itself be the canonical view.

        A missing CRS is intentionally treated as the GeoJSON/RFC-7946
        default EPSG:4326.  Other aliases (for example ``CRS:84``) are
        rewritten so every transformed consumption artifact explicitly says
        ``EPSG:4326``.
        """

        return str(source_crs or cls.CANONICAL_VECTOR_CRS).strip().upper() != cls.CANONICAL_VECTOR_CRS

    @classmethod
    def _persist_vector_source_evidence(
        cls,
        *,
        project_id: UUID,
        dataset_id: UUID,
        original_filename: str,
        content: bytes,
        content_type: str | None,
    ) -> dict[str, Any]:
        """Retain a non-canonical source file outside the consumption path.

        The Dataset's normal ``storage_path`` always points at the canonical
        artifact.  The source bytes are retained only below ``provenance/``
        and are referenced through structured provenance metadata; consumers
        must never treat this location as a dataset input.
        """

        safe_filename = StorageService._safe_filename(original_filename)
        evidence_path = (
            StorageService.dataset_root(str(project_id), str(dataset_id), "vector")
            / "provenance"
            / f"{dataset_id}_source_{safe_filename}"
        )
        return StorageService.persist_file(
            str(evidence_path),
            content,
            original_filename=safe_filename,
            content_type=content_type,
        )

    @classmethod
    def _record_vector_source_evidence(
        cls,
        *,
        source_metadata: dict[str, Any],
        provenance_metadata: dict[str, Any],
        source_crs: str,
        evidence: dict[str, Any],
        canonical_checksum_sha256: str,
    ) -> None:
        """Bind original source bytes to their canonical consumption artifact."""

        source_artifact = {
            "storage_path": evidence["storage_path"],
            "checksum_sha256": evidence["checksum_sha256"],
            "size_bytes": evidence["size_bytes"],
            "content_type": evidence["content_type"],
            "source_crs": source_crs,
            "retention": "provenance_evidence_only",
        }
        transformation = {
            "name": "vector_crs_normalization",
            "version": "1.0.0",
            "source_crs": source_crs,
            "storage_crs": cls.CANONICAL_VECTOR_CRS,
            "source_checksum_sha256": evidence["checksum_sha256"],
            "canonical_checksum_sha256": canonical_checksum_sha256,
        }
        source_metadata["source_artifact"] = source_artifact
        provenance_metadata["source_artifact"] = source_artifact
        provenance_metadata["canonical_consumption_artifact"] = {
            "checksum_sha256": canonical_checksum_sha256,
            "crs": cls.CANONICAL_VECTOR_CRS,
            "storage_role": "dataset_consumption",
        }
        provenance_metadata["transformations"] = [
            *(
                provenance_metadata.get("transformations")
                if isinstance(provenance_metadata.get("transformations"), list)
                else []
            ),
            transformation,
        ]

    @staticmethod
    def _calculate_file_checksum_sha256(path: str | Path) -> str:
        """Stream an operator artifact before storage for an idempotent ingest key."""

        artifact = Path(path)
        if not artifact.is_file():
            raise AppError(
                code="DATASET_FILE_MISSING",
                message="Partitioned vector artifact is missing",
                details={"artifact_path": str(artifact)},
                status_code=404,
            )
        digest = sha256()
        with artifact.open("rb") as stream:
            for chunk in iter(lambda: stream.read(8 * 1024 * 1024), b""):
                digest.update(chunk)
        return digest.hexdigest()

    @classmethod
    def _ingest_key(
        cls,
        *,
        project_id: UUID,
        source_key: str,
        checksum_sha256: str,
        dataset_type: str,
        dataset_role: str,
        area_id: UUID | None,
        reference_layer_name: str | None,
        source_version: str | None,
    ) -> str:
        return cls._stable_hash(
            {
                "project_id": str(project_id),
                "source_key": source_key,
                "checksum_sha256": checksum_sha256.lower(),
                "dataset_type": dataset_type,
                "dataset_role": dataset_role,
                "area_id": str(area_id) if area_id else None,
                "reference_layer_name": reference_layer_name or None,
                "source_version": source_version or None,
                "ingest_contract": "phase2-source-provenance-v1",
            }
        )

    @staticmethod
    def _contract_metadata(
        *,
        metadata: dict[str, Any],
        source_metadata: dict[str, Any] | None,
        provenance_metadata: dict[str, Any] | None,
        source: Any,
    ) -> dict[str, Any]:
        result = dict(metadata)
        source_values = source_metadata if isinstance(source_metadata, dict) else {}
        provenance_values = provenance_metadata if isinstance(provenance_metadata, dict) else {}
        result["license"] = (
            result.get("license")
            or source_values.get("license")
            or source_values.get("license_note")
            or provenance_values.get("license")
            or getattr(source, "license_name", None)
            or "unknown"
        )
        result.setdefault(
            "usage_restrictions",
            source_values.get("usage_restrictions")
            or getattr(source, "usage_restrictions", None)
            or "unknown",
        )
        return result

    @staticmethod
    def _validate_vector_source_schema(source: Any, feature_collection: dict[str, Any]) -> dict[str, Any]:
        """Validate the server-owned source schema after canonicalization.

        Generic GeoJSON validation proves that a feature collection is
        structurally valid.  This additional pass proves that it also matches
        the geometry and required-attribute expectations recorded for the
        selected source registry entry.  It deliberately uses the
        canonical-storage payload so the evidence describes exactly what will
        be persisted in ``vector_features``.
        """

        schema = _SourceVectorSchema.from_source(source)
        features = feature_collection.get("features") if isinstance(feature_collection, dict) else None
        if not isinstance(features, list):
            raise AppError(
                code="SOURCE_SCHEMA_FEATURE_COLLECTION_INVALID",
                message="Source-schema validation requires a GeoJSON FeatureCollection.",
                status_code=400,
            )
        for index, feature in enumerate(features):
            if not isinstance(feature, dict):
                raise AppError(
                    code="SOURCE_SCHEMA_FEATURE_INVALID",
                    message=f"Feature {index} is not an object during source-schema validation.",
                    status_code=400,
                )
            properties = feature.get("properties") if isinstance(feature.get("properties"), dict) else {}
            try:
                geometry = shape(feature.get("geometry"))
            except Exception as exc:
                raise AppError(
                    code="SOURCE_SCHEMA_GEOMETRY_INVALID",
                    message=f"Feature {index} has no parseable geometry during source-schema validation.",
                    status_code=400,
                ) from exc
            _validate_vector_feature_source_schema(
                feature=feature,
                properties=properties,
                geometry=geometry,
                schema=schema,
                feature_context=f"Feature {index}",
            )
        return schema.to_metadata(checked_feature_count=len(features))

    @staticmethod
    def _snapshot_freshness_status(
        source_key: str,
        source_metadata: dict[str, Any] | None,
        *,
        observed_at: datetime | None,
        source_version: str | None,
    ) -> str:
        metadata = source_metadata if isinstance(source_metadata, dict) else {}
        supplied = str(metadata.get("freshness_status") or "").strip().lower()
        allowed = {"unknown", "current", "due", "stale", "not_applicable", "review_required"}
        if supplied in allowed:
            return supplied
        if source_key in {"manual", "fixture", "map_selection", "derived", "experimental"}:
            return "not_applicable"
        return "current" if observed_at is not None or bool((source_version or "").strip()) else "review_required"

    @staticmethod
    def _resolution_unit_for_crs(crs: str | None) -> str:
        normalized = str(crs or "").strip().upper()
        return "degree" if normalized in {"EPSG:4326", "CRS:84", "OGC:CRS84"} else "m"

    @staticmethod
    def _failed_validation_report(
        *,
        asset_id: str,
        dataset_type: str,
        code: str,
        message: str,
        now: datetime,
        category: str = "parser",
    ) -> ValidationReport:
        if dataset_type == "vector":
            contract_key, contract_version = VECTOR_GEOJSON_CONTRACT_KEY, VECTOR_GEOJSON_CONTRACT_VERSION
        else:
            contract_key, contract_version = RASTER_GEOTIFF_CONTRACT_KEY, RASTER_GEOTIFF_CONTRACT_VERSION
        return ValidationReport(
            asset_id=asset_id,
            data_contract_key=contract_key,
            data_contract_version=contract_version,
            contract_fingerprint_sha256=None,
            validation_status=ValidationStatus.FAILED,
            provenance_status=ProvenanceStatus.INCOMPLETE,
            lineage_status=LineageStatus.INCOMPLETE,
            quarantine_status=QuarantineStatus.QUARANTINED,
            validation_scope=("ingest", dataset_type),
            checked_at=now,
            issues=(
                ValidationIssue(
                    code=code,
                    category=category,
                    field="artifact",
                    message=message,
                ),
            ),
        )

    @staticmethod
    def _normalize_datetime(value: datetime | None) -> datetime | None:
        if value is None:
            return None
        if value.tzinfo is None:
            return value.replace(tzinfo=timezone.utc)
        return value.astimezone(timezone.utc)

    @staticmethod
    def _validate_temporal_metadata(
        *,
        temporal_series_key: str | None,
        observed_at: datetime | None,
        valid_from: datetime | None,
        valid_to: datetime | None,
        temporal_granularity: str | None,
        source_version: str | None,
    ) -> dict[str, Any]:
        normalized_key = (temporal_series_key or "").strip() or None
        normalized_observed_at = DatasetService._normalize_datetime(observed_at)
        normalized_valid_from = DatasetService._normalize_datetime(valid_from)
        normalized_valid_to = DatasetService._normalize_datetime(valid_to)
        normalized_granularity = (temporal_granularity or "").strip().lower() or None
        normalized_source_version = (source_version or "").strip() or None

        if normalized_key and len(normalized_key) > 255:
            raise AppError(code="INVALID_TEMPORAL_METADATA", message="temporal_series_key is too long", status_code=400)
        if normalized_granularity and normalized_granularity not in DatasetService.VALID_TEMPORAL_GRANULARITIES:
            raise AppError(
                code="INVALID_TEMPORAL_METADATA",
                message="temporal_granularity must be snapshot, day, month, year or period",
                status_code=400,
            )
        if normalized_valid_from and normalized_valid_to and normalized_valid_to < normalized_valid_from:
            raise AppError(
                code="INVALID_TEMPORAL_METADATA",
                message="valid_to must be on or after valid_from",
                status_code=400,
            )
        if normalized_key and normalized_observed_at is None:
            raise AppError(
                code="INVALID_TEMPORAL_METADATA",
                message="observed_at is required when temporal_series_key is provided",
                status_code=400,
            )
        if normalized_observed_at and normalized_key is None:
            raise AppError(
                code="INVALID_TEMPORAL_METADATA",
                message="temporal_series_key is required when observed_at is provided",
                status_code=400,
            )
        return {
            "temporal_series_key": normalized_key,
            "observed_at": normalized_observed_at,
            "valid_from": normalized_valid_from,
            "valid_to": normalized_valid_to,
            "temporal_granularity": normalized_granularity,
            "source_version": normalized_source_version,
        }

    @staticmethod
    def _to_response(dataset: Dataset) -> DatasetCreateResponse:
        metadata_json = dataset.metadata_json if isinstance(dataset.metadata_json, dict) else {}
        return DatasetCreateResponse(
            id=dataset.id,
            name=dataset.name,
            dataset_type=dataset.dataset_type,
            source=dataset.source,
            dataset_role=dataset.dataset_role,
            source_name=dataset.source_name,
            reference_layer_name=dataset.reference_layer_name,
            source_metadata=dataset.source_metadata,
            provenance_metadata=dataset.provenance_metadata,
            ingest_key=dataset.ingest_key,
            source_registry_id=dataset.source_registry_id,
            source_snapshot_id=dataset.source_snapshot_id,
            data_contract_key=dataset.data_contract_key,
            data_contract_version=dataset.data_contract_version,
            validation_status=dataset.validation_status,
            validation_report_json=dataset.validation_report_json,
            provenance_status=dataset.provenance_status,
            lineage_status=dataset.lineage_status,
            quarantine_status=dataset.quarantine_status,
            imported_at=dataset.imported_at,
            temporal_series_key=dataset.temporal_series_key,
            observed_at=dataset.observed_at,
            valid_from=dataset.valid_from,
            valid_to=dataset.valid_to,
            temporal_granularity=dataset.temporal_granularity,
            source_version=dataset.source_version,
            project_id=dataset.project_id,
            area_id=dataset.area_id,
            storage_path=dataset.storage_path,
            original_filename=dataset.original_filename,
            stored_filename=dataset.stored_filename,
            content_type=dataset.content_type,
            size_bytes=dataset.size_bytes,
            checksum_sha256=dataset.checksum_sha256,
            crs=dataset.crs,
            bounds_json=dataset.bounds_json,
            metadata_json=dataset.metadata_json,
            vector_summary=DatasetService._extract_vector_summary(dataset.dataset_type, metadata_json),
            status=dataset.status,
            derived_from_dataset_id=dataset.derived_from_dataset_id,
            created_at=dataset.created_at,
            feature_count=metadata_json.get("feature_count"),
        )

    @staticmethod
    def _canonical_dataset_type(dataset_type: str) -> str:
        normalized = (dataset_type or "").strip().lower()
        if normalized in DatasetService.VECTOR_TYPES:
            return "vector"
        if normalized in DatasetService.RASTER_TYPES:
            return "raster"
        raise AppError(
            code="INVALID_DATASET_TYPE",
            message="dataset_type must be 'vector' or 'raster' (or legacy 'geojson')",
            status_code=400,
        )

    @staticmethod
    def _normalize_stored_dataset_type(dataset_type: str) -> str:
        normalized = (dataset_type or "").strip().lower()
        if normalized in DatasetService.VECTOR_TYPES:
            return "vector"
        if normalized in DatasetService.RASTER_TYPES:
            return "raster"
        return normalized

    @staticmethod
    def _is_vector_type(dataset_type: str) -> bool:
        return DatasetService._normalize_stored_dataset_type(dataset_type) == "vector"

    @staticmethod
    def _is_raster_type(dataset_type: str) -> bool:
        return DatasetService._normalize_stored_dataset_type(dataset_type) == "raster"

    @staticmethod
    def _normalize_dataset_role(dataset_role: str | None) -> str:
        normalized = (dataset_role or "").strip().lower() or "source"
        if normalized not in DatasetService.VALID_DATASET_ROLES:
            raise AppError(
                code="INVALID_DATASET_ROLE",
                message="dataset_role must be one of: source, derived, reference",
                status_code=400,
            )
        return normalized

    @staticmethod
    def _extension_for_path(filename: str) -> str:
        return Path(filename).suffix.lower()

    @staticmethod
    def _validate_upload_filename(filename: str | None) -> str:
        if not filename:
            raise AppError(code="INVALID_UPLOAD", message="Missing file name", status_code=400)
        return filename

    @staticmethod
    def list_datasets(db: Session, project_id: UUID, limit: int = 50, offset: int = 0) -> tuple[list[DatasetCreateResponse], int]:
        total = db.query(Dataset).filter(Dataset.project_id == project_id).count()
        rows = (
            db.query(Dataset)
            .filter(Dataset.project_id == project_id)
            .order_by(Dataset.created_at.desc())
            .offset(offset)
            .limit(limit)
            .all()
        )
        return [DatasetService._to_response(row) for row in rows], total

    @staticmethod
    def _extract_vector_summary(dataset_type: str, metadata_json: dict) -> DatasetVectorSummary | None:
        if not DatasetService._is_vector_type(dataset_type):
            return None
        if not isinstance(metadata_json, dict):
            return None
        return DatasetVectorSummary(
            feature_count=metadata_json.get("feature_count"),
            geometry_types=metadata_json.get("geometry_types"),
            bounds_json=metadata_json.get("bounds_json"),
            approximate_area_m2=metadata_json.get("approximate_area_m2"),
            crs=metadata_json.get("crs"),
            feature_geometry_count=metadata_json.get("feature_geometry_count"),
            invalid_features=metadata_json.get("invalid_features"),
            crs_assumed=metadata_json.get("crs_assumed"),
        )

    @staticmethod
    def _extract_raster_bounds_json(metadata_json: dict[str, Any]) -> dict[str, float] | None:
        existing = metadata_json.get("bounds_json")
        if isinstance(existing, dict):
            return existing
        bounds = metadata_json.get("bounds")
        if isinstance(bounds, (list, tuple)) and len(bounds) == 4:
            return {
                "minx": float(bounds[0]),
                "miny": float(bounds[1]),
                "maxx": float(bounds[2]),
                "maxy": float(bounds[3]),
            }
        return None

    @staticmethod
    def _extract_raster_resolution_json(metadata_json: dict[str, Any]) -> dict[str, float] | None:
        existing = metadata_json.get("resolution_json")
        if isinstance(existing, dict):
            return existing
        resolution = metadata_json.get("resolution")
        if isinstance(resolution, (list, tuple)) and len(resolution) >= 2:
            return {"x": float(resolution[0]), "y": float(resolution[1])}
        return None

    @staticmethod
    def _extract_raster_bands_json(metadata_json: dict[str, Any]) -> dict[str, Any] | None:
        existing = metadata_json.get("bands_json")
        if isinstance(existing, dict):
            return existing
        bands_json: dict[str, Any] = {}
        if metadata_json.get("band_count") is not None:
            bands_json["band_count"] = int(metadata_json["band_count"])
        if metadata_json.get("dtype") is not None:
            bands_json["dtype"] = metadata_json["dtype"]
        return bands_json or None

    @classmethod
    def _find_existing_ingest(cls, db: Session, project_id: UUID, ingest_key: str) -> Dataset | None:
        if not cls._registry_persistence_available(db):
            return None
        return SourceRegistryService.find_dataset_by_ingest_key(db, project_id, ingest_key)

    @classmethod
    def _record_snapshot(
        cls,
        db: Session,
        *,
        source_key: str,
        checksum_sha256: str,
        source_version: str | None,
        observed_at: datetime | None,
        valid_from: datetime | None,
        valid_to: datetime | None,
        source_crs: str | None,
        source_metadata: dict[str, Any] | None,
        metadata: dict[str, Any],
    ) -> tuple[Any | None, Any | None]:
        if not cls._registry_persistence_available(db):
            return None, None
        source = SourceRegistryService.ensure_server_owned_source(db, source_key)
        source_values = source_metadata if isinstance(source_metadata, dict) else {}
        resolution = metadata.get("resolution_json") or metadata.get("resolution") or {}
        if isinstance(resolution, (list, tuple)) and len(resolution) >= 2:
            resolution = {"x": resolution[0], "y": resolution[1], "unit": cls._resolution_unit_for_crs(source_crs)}
        if not isinstance(resolution, dict):
            resolution = {"status": "unknown"}
        snapshot_key = f"{source_key}:{source_version or 'unversioned'}:{checksum_sha256.lower()}"
        snapshot = SourceRegistryService.record_snapshot(
            db,
            source_key=source_key,
            snapshot_key=snapshot_key,
            checksum_sha256=checksum_sha256,
            source_version=source_version,
            snapshot_at=observed_at,
            fetched_at=datetime.now(timezone.utc),
            reuse_existing_snapshot=True,
            source_url=(
                source_values.get("source_url")
                or source_values.get("catalogue_url")
                or source_values.get("service_url")
            ),
            crs=source_crs,
            units=source_values.get("units") or source.default_units,
            spatial_resolution=resolution,
            temporal_coverage={
                "observed_at": observed_at.isoformat() if observed_at else None,
                "valid_from": valid_from.isoformat() if valid_from else None,
                "valid_to": valid_to.isoformat() if valid_to else None,
            },
            geographic_coverage={
                "bbox": metadata.get("source_bounds_json") or metadata.get("bounds_json") or metadata.get("bounds"),
                "coverage_zones": source_values.get("coverage_zones") or source_values.get("coverage_zone"),
            },
            observed_schema={
                "dataset_type": metadata.get("dataset_type"),
                "geometry_types": metadata.get("geometry_types"),
                "bands": metadata.get("band_count"),
                "attributes": source_values.get("expected_attributes"),
            },
            freshness_status=cls._snapshot_freshness_status(
                source_key,
                source_metadata,
                observed_at=observed_at,
                source_version=source_version,
            ),
            ingest_status="ingested",
            known_limitations=list(source_values.get("known_limitations") or []),
            snapshot_metadata={
                "source_metadata": source_values,
                "source_checksum_sha256": checksum_sha256.lower(),
            },
        )
        return source, snapshot

    @classmethod
    def _apply_validation_report(
        cls,
        db: Session,
        *,
        dataset: Dataset,
        dataset_version: DatasetVersion,
        report: ValidationReport,
        source: Any | None,
        snapshot: Any | None,
        artifact_path: str | None,
    ) -> None:
        fields = report.persistence_fields()
        dataset.validation_report_json = fields["validation_report_json"]
        dataset.quarantine_status = fields["quarantine_status"]
        dataset_version.validation_report_json = fields["validation_report_json"]
        if source is not None and snapshot is not None:
            SourceRegistryService.bind_dataset_provenance(
                dataset,
                source=source,
                snapshot=snapshot,
                data_contract_key=fields["data_contract_key"],
                data_contract_version=fields["data_contract_version"],
                validation_status=fields["validation_status"],
                provenance_status=fields["provenance_status"],
                lineage_status=fields["lineage_status"],
            )
            SourceRegistryService.bind_dataset_version_provenance(
                dataset_version,
                source=source,
                snapshot=snapshot,
                data_contract_key=fields["data_contract_key"],
                data_contract_version=fields["data_contract_version"],
                validation_status=fields["validation_status"],
                provenance_status=fields["provenance_status"],
                lineage_status=fields["lineage_status"],
            )
        else:
            for target in (dataset, dataset_version):
                target.data_contract_key = fields["data_contract_key"]
                target.data_contract_version = fields["data_contract_version"]
                target.validation_status = fields["validation_status"]
                target.provenance_status = fields["provenance_status"]
                target.lineage_status = fields["lineage_status"]

        decision = DataQuarantineService.decide(report)
        if decision.eligible_for_use:
            dataset.status = "ready"
            dataset.quarantine_status = "not_quarantined"
            return
        dataset.status = "quarantined"
        dataset.quarantine_status = "quarantined"
        if source is not None and snapshot is not None:
            SourceRegistryService.quarantine_dataset(
                db,
                dataset=dataset,
                dataset_version=dataset_version,
                source_snapshot=snapshot,
                stage="ingest_validation",
                reason_code=(decision.reason_codes[0] if decision.reason_codes else "DATA_CONTRACT_FAILED"),
                details={"validation_report": report.to_dict(), "quarantine_decision": decision.to_dict()},
                artifact_path=artifact_path,
                artifact_checksum_sha256=dataset.checksum_sha256,
            )

    @classmethod
    def _new_dataset_version(
        cls,
        dataset: Dataset,
        *,
        ingest_key: str | None,
    ) -> DatasetVersion:
        return DatasetVersion(
            id=uuid.uuid4(),
            dataset_id=dataset.id,
            version=1,
            storage_path=dataset.storage_path,
            source_version=dataset.source_version,
            observed_at=dataset.observed_at,
            valid_from=dataset.valid_from,
            valid_to=dataset.valid_to,
            checksum_sha256=dataset.checksum_sha256,
            ingest_key=f"{ingest_key}:v1" if ingest_key else None,
            source_metadata=dataset.source_metadata,
            provenance_metadata=dataset.provenance_metadata,
        )

    @staticmethod
    async def _upload_dataset_legacy(
        db: Session,
        project_id: UUID,
        file: UploadFile,
        dataset_type: str,
        source: str,
        dataset_role: str = "source",
        source_name: str | None = None,
        reference_layer_name: str | None = None,
        source_metadata: dict | None = None,
        provenance_metadata: dict | None = None,
        area_id: UUID | None = None,
        temporal_series_key: str | None = None,
        observed_at: datetime | None = None,
        valid_from: datetime | None = None,
        valid_to: datetime | None = None,
        temporal_granularity: str | None = None,
        source_version: str | None = None,
    ) -> DatasetCreateResponse:
        if not db.get(Project, project_id):
            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)

        filename = DatasetService._validate_upload_filename(file.filename)
        canonical_type = DatasetService._canonical_dataset_type(dataset_type)
        normalized_role = DatasetService._normalize_dataset_role(dataset_role)
        temporal = DatasetService._validate_temporal_metadata(
            temporal_series_key=temporal_series_key,
            observed_at=observed_at,
            valid_from=valid_from,
            valid_to=valid_to,
            temporal_granularity=temporal_granularity,
            source_version=source_version,
        )
        normalized_source_name = source_name
        if normalized_role == "reference" and not normalized_source_name:
            normalized_source_name = "manual"
        if normalized_role == "reference" and canonical_type == "raster":
            raise AppError(code="INVALID_DATASET_ROLE", message="reference role requires vector dataset type", status_code=400)
        extension = DatasetService._extension_for_path(filename)

        if canonical_type == "vector" and extension not in DatasetService.VECTOR_EXTENSIONS:
            raise AppError(code="INVALID_UPLOAD", message="Vector uploads require .geojson or .json files", status_code=415)
        if canonical_type == "raster" and extension not in DatasetService.RASTER_EXTENSIONS:
            raise AppError(
                code="INVALID_UPLOAD",
                message="Raster uploads require .tif, .tiff or .geotiff files",
                status_code=415,
            )

        raw = await file.read()
        storage_info = StorageService.persist_dataset_file(
            project_id=str(project_id),
            dataset_id=str(dataset_id := uuid.uuid4()),
            dataset_type=canonical_type,
            original_filename=filename,
            content=raw,
            content_type=file.content_type,
        )

        metadata: dict[str, Any] = {}
        vector_payload: dict[str, Any] | None = None
        status = "uploaded"
        try:
            status = "validating"
            if canonical_type == "vector":
                try:
                    text = raw.decode("utf-8")
                except UnicodeDecodeError as exc:
                    raise AppError(code="INVALID_UPLOAD", message="Upload must be UTF-8 encoded", status_code=400) from exc
                metadata = parse_geojson_payload(text)
                vector_payload = json.loads(text)
                status = "ready"
            else:
                metadata = extract_raster_metadata(storage_info["storage_path"])
                status = "ready"
        except ValueError as exc:
            status = "failed"
            StorageService.remove_dataset_file(storage_info["storage_path"])
            raise AppError(code="INVALID_GEOJSON", message=str(exc), status_code=400) from exc
        except AppError as exc:
            if canonical_type == "raster" and exc.code == "RASTER_PROCESSING_UNAVAILABLE":
                status = "failed"
                metadata = {
                    "processing_error": exc.message,
                    "processing_code": exc.code,
                }
            else:
                StorageService.remove_dataset_file(storage_info["storage_path"])
                raise

        bounds_json = metadata.get("bounds_json") if isinstance(metadata, dict) else None
        resolution_json = metadata.get("resolution_json") if isinstance(metadata, dict) else None
        bands_json = metadata.get("bands_json") if isinstance(metadata, dict) else None
        if canonical_type == "raster" and isinstance(metadata, dict):
            bounds_json = DatasetService._extract_raster_bounds_json(metadata)
            resolution_json = DatasetService._extract_raster_resolution_json(metadata)
            bands_json = DatasetService._extract_raster_bands_json(metadata)

        dataset = Dataset(
            id=dataset_id,
            project_id=project_id,
            area_id=area_id,
            name=filename,
            dataset_type=canonical_type,
            source=source,
            dataset_role=normalized_role,
            source_name=normalized_source_name,
            reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
            source_metadata=source_metadata,
            provenance_metadata=provenance_metadata,
            imported_at=datetime.now(timezone.utc),
            **temporal,
            storage_path=storage_info["storage_path"],
            original_filename=storage_info["original_filename"],
            stored_filename=storage_info["stored_filename"],
            content_type=storage_info["content_type"],
            size_bytes=storage_info["size_bytes"],
            checksum_sha256=storage_info["checksum_sha256"],
            crs=metadata.get("crs") if isinstance(metadata, dict) else None,
            bounds_json=bounds_json,
            resolution_json=resolution_json,
            bands_json=bands_json,
            metadata_json=metadata,
            status=status,
        )
        try:
            db.add(dataset)
            db.add(
                DatasetVersion(
                    dataset_id=dataset.id,
                    version=1,
                    storage_path=dataset.storage_path,
                    source_version=dataset.source_version,
                    observed_at=dataset.observed_at,
                    valid_from=dataset.valid_from,
                    valid_to=dataset.valid_to,
                    checksum_sha256=dataset.checksum_sha256,
                    source_metadata=dataset.source_metadata,
                    provenance_metadata=dataset.provenance_metadata,
                )
            )
            if canonical_type == "vector" and vector_payload is not None and status == "ready":
                feature_class = reference_layer_name if normalized_role == "reference" else None
                VectorFeatureService.persist_geojson_features(
                    db=db,
                    dataset_id=dataset.id,
                    payload=vector_payload,
                    feature_class=feature_class,
                    commit=False,
                )
            db.commit()
            db.refresh(dataset)
        except Exception:
            db.rollback()
            StorageService.remove_dataset_file(storage_info["storage_path"])
            raise

        return DatasetService._to_response(dataset)

    @staticmethod
    async def upload_dataset(
        db: Session,
        project_id: UUID,
        file: UploadFile,
        dataset_type: str,
        source: str,
        dataset_role: str = "source",
        source_name: str | None = None,
        reference_layer_name: str | None = None,
        source_metadata: dict | None = None,
        provenance_metadata: dict | None = None,
        area_id: UUID | None = None,
        temporal_series_key: str | None = None,
        observed_at: datetime | None = None,
        valid_from: datetime | None = None,
        valid_to: datetime | None = None,
        temporal_granularity: str | None = None,
        source_version: str | None = None,
    ) -> DatasetCreateResponse:
        """Stage a user upload as an explicitly manual, non-authoritative source.

        Client text such as ``source_name=grb`` is retained only as a claim in
        provenance. It cannot select an authoritative registry entry; only a
        server-owned acquisition adapter reaches those entries.
        """
        if not DatasetService._registry_persistence_available(db):
            return await DatasetService._upload_dataset_legacy(
                db=db,
                project_id=project_id,
                file=file,
                dataset_type=dataset_type,
                source=source,
                dataset_role=dataset_role,
                source_name=source_name,
                reference_layer_name=reference_layer_name,
                source_metadata=source_metadata,
                provenance_metadata=provenance_metadata,
                area_id=area_id,
                temporal_series_key=temporal_series_key,
                observed_at=observed_at,
                valid_from=valid_from,
                valid_to=valid_to,
                temporal_granularity=temporal_granularity,
                source_version=source_version,
            )
        if not db.get(Project, project_id):
            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
        if area_id is not None:
            area = db.get(Area, area_id)
            if not area:
                raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
            if area.project_id != project_id:
                raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400)

        filename = DatasetService._validate_upload_filename(file.filename)
        canonical_type = DatasetService._canonical_dataset_type(dataset_type)
        normalized_role = DatasetService._normalize_dataset_role(dataset_role)
        if normalized_role == "reference" and canonical_type == "raster":
            raise AppError(code="INVALID_DATASET_ROLE", message="reference role requires vector dataset type", status_code=400)
        extension = DatasetService._extension_for_path(filename)
        if canonical_type == "vector" and extension not in DatasetService.VECTOR_EXTENSIONS:
            raise AppError(code="INVALID_UPLOAD", message="Vector uploads require .geojson or .json files", status_code=415)
        if canonical_type == "raster" and extension not in DatasetService.RASTER_EXTENSIONS:
            raise AppError(code="INVALID_UPLOAD", message="Raster uploads require .tif, .tiff or .geotiff files", status_code=415)

        temporal = DatasetService._validate_temporal_metadata(
            temporal_series_key=temporal_series_key,
            observed_at=observed_at,
            valid_from=valid_from,
            valid_to=valid_to,
            temporal_granularity=temporal_granularity,
            source_version=source_version,
        )
        raw = await file.read()
        checksum_sha256 = StorageService.calculate_checksum_sha256(raw)
        ingest_key = DatasetService._ingest_key(
            project_id=project_id,
            source_key="manual",
            checksum_sha256=checksum_sha256,
            dataset_type=canonical_type,
            dataset_role=normalized_role,
            area_id=area_id,
            reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
            source_version=temporal["source_version"],
        )
        existing = DatasetService._find_existing_ingest(db, project_id, ingest_key)
        if existing is not None:
            return DatasetService._to_response(existing)

        raw_source_metadata = dict(source_metadata or {})
        raw_provenance_metadata = dict(provenance_metadata or {})
        raw_source_metadata.update(
            {
                "ingest_origin": "manual_upload",
                "claimed_source": source,
                "claimed_source_name": source_name,
                "authority_claim_accepted": False,
            }
        )
        raw_source_metadata.setdefault(
            "temporal_unknown_reason",
            "The manual upload does not assert a precise source observation timestamp.",
        )
        raw_source_metadata.setdefault(
            "source_version_unknown_reason",
            "The manual upload has no server-attested source edition or snapshot version.",
        )
        raw_provenance_metadata.update(
            {
                "ingest_origin": "manual_upload",
                "ingest_key": ingest_key,
                "claimed_source": {"source": source, "source_name": source_name},
            }
        )

        dataset_id = uuid.uuid4()
        storage_info: dict[str, Any] | None = None
        storage_content = raw
        source_evidence: dict[str, Any] | None = None
        imported_at = datetime.now(timezone.utc)
        metadata: dict[str, Any] = {"dataset_type": canonical_type}
        source_crs: str | None = None
        canonical_vector_payload: dict[str, Any] | None = None
        parser_error: tuple[str, str] | None = None
        try:
            if canonical_type == "vector":
                try:
                    payload = json.loads(raw.decode("utf-8"))
                except UnicodeDecodeError as exc:
                    raise AppError(code="INVALID_UPLOAD", message="Upload must be UTF-8 encoded", status_code=400) from exc
                raw_metadata = parse_geojson_payload(payload)
                source_crs = str(raw_metadata.get("crs") or "").strip() or None
                canonical_vector_payload = VectorFeatureService.canonicalize_geojson_payload(
                    payload,
                    source_crs=source_crs or DatasetService.CANONICAL_VECTOR_CRS,
                )
                metadata = parse_geojson_payload(canonical_vector_payload)
                metadata.update(
                    {
                        "dataset_type": "vector",
                        "source_crs": source_crs,
                        "source_bounds_json": raw_metadata.get("bounds_json"),
                        "source_crs_assumed": raw_metadata.get("crs_assumed", False),
                        "canonical_storage_crs": DatasetService.CANONICAL_VECTOR_CRS,
                    }
                )
                if DatasetService._vector_storage_requires_canonicalization(source_crs):
                    storage_content = DatasetService._canonical_vector_storage_bytes(canonical_vector_payload)
                    source_evidence = DatasetService._persist_vector_source_evidence(
                        project_id=project_id,
                        dataset_id=dataset_id,
                        original_filename=filename,
                        content=raw,
                        content_type=file.content_type,
                    )
            else:
                storage_info = StorageService.persist_dataset_file(
                    project_id=str(project_id),
                    dataset_id=str(dataset_id),
                    dataset_type=canonical_type,
                    original_filename=filename,
                    content=raw,
                    content_type=file.content_type,
                )
                metadata = extract_raster_metadata(storage_info["storage_path"])
                metadata["dataset_type"] = "raster"
                source_crs = metadata.get("crs")
        except (ValueError, json.JSONDecodeError, AppError) as exc:
            code = exc.code if isinstance(exc, AppError) else "INVALID_GEOJSON"
            parser_error = (code, str(exc))
            metadata = {
                "dataset_type": canonical_type,
                "processing_error": str(exc),
                "processing_code": code,
            }

        if storage_info is None:
            storage_info = StorageService.persist_dataset_file(
                project_id=str(project_id),
                dataset_id=str(dataset_id),
                dataset_type=canonical_type,
                original_filename=filename,
                content=storage_content,
                content_type=file.content_type,
            )
        computed_storage_checksum_sha256 = StorageService.calculate_checksum_sha256(storage_content)
        if source_evidence is not None:
            resolved_source_crs = source_crs or DatasetService.CANONICAL_VECTOR_CRS
            DatasetService._record_vector_source_evidence(
                source_metadata=raw_source_metadata,
                provenance_metadata=raw_provenance_metadata,
                source_crs=resolved_source_crs,
                evidence=source_evidence,
                canonical_checksum_sha256=computed_storage_checksum_sha256,
            )
            metadata.update(
                {
                    "source_artifact_checksum_sha256": source_evidence["checksum_sha256"],
                    "canonical_artifact_checksum_sha256": computed_storage_checksum_sha256,
                }
            )

        source_registry, source_snapshot = DatasetService._record_snapshot(
            db,
            source_key="manual",
            checksum_sha256=storage_info["checksum_sha256"],
            source_version=temporal["source_version"],
            observed_at=temporal["observed_at"],
            valid_from=temporal["valid_from"],
            valid_to=temporal["valid_to"],
            source_crs=source_crs,
            source_metadata=raw_source_metadata,
            metadata=metadata,
        )
        contract_metadata = DatasetService._contract_metadata(
            metadata=metadata,
            source_metadata=raw_source_metadata,
            provenance_metadata=raw_provenance_metadata,
            source=source_registry,
        )
        if parser_error is not None:
            report = DatasetService._failed_validation_report(
                asset_id=ingest_key,
                dataset_type=canonical_type,
                code=parser_error[0],
                message=parser_error[1],
                now=imported_at,
            )
        elif canonical_type == "vector":
            source_crs = source_crs or DatasetService.CANONICAL_VECTOR_CRS
            lineage = LineageEvidence()
            if source_crs.upper() != DatasetService.CANONICAL_VECTOR_CRS:
                lineage = LineageEvidence(
                    transformations=(
                        TransformationEvidence(
                            name="vector_crs_normalization",
                            version="1.0.0",
                            checksum_sha256=DatasetService._stable_hash(
                                {"source_crs": source_crs, "storage_crs": DatasetService.CANONICAL_VECTOR_CRS}
                            ),
                        ),
                    )
                )
            try:
                contract_metadata["source_schema_validation"] = DatasetService._validate_vector_source_schema(
                    source_registry,
                    canonical_vector_payload or {"type": "FeatureCollection", "features": []},
                )
                report = validate_registered_asset(
                    build_vector_ingest_input(
                        asset_id=ingest_key,
                        source_crs=source_crs,
                        storage_crs=DatasetService.CANONICAL_VECTOR_CRS,
                        feature_collection=canonical_vector_payload or {"type": "FeatureCollection", "features": []},
                        checksum_sha256=storage_info["checksum_sha256"],
                        computed_checksum_sha256=computed_storage_checksum_sha256,
                        content=storage_content,
                        source_registry_id=str(source_registry.id),
                        source_snapshot_id=str(source_snapshot.id),
                        imported_at=imported_at,
                        metadata=contract_metadata,
                        observed_at=temporal["observed_at"],
                        valid_from=temporal["valid_from"],
                        valid_to=temporal["valid_to"],
                        temporal_unknown_reason=raw_source_metadata["temporal_unknown_reason"],
                        source_version=temporal["source_version"],
                        source_version_unknown_reason=raw_source_metadata["source_version_unknown_reason"],
                        lineage=lineage,
                    )
                )
            except AppError as exc:
                report = DatasetService._failed_validation_report(
                    asset_id=ingest_key,
                    dataset_type="vector",
                    code=exc.code,
                    message=exc.message,
                    now=imported_at,
                    category="source_schema",
                )
        else:
            resolution_json = DatasetService._extract_raster_resolution_json(metadata)
            resolution = (
                {"x": resolution_json["x"], "y": resolution_json["y"], "unit": DatasetService._resolution_unit_for_crs(source_crs)}
                if resolution_json
                else None
            )
            report = validate_registered_asset(
                build_raster_ingest_input(
                    asset_id=ingest_key,
                    source_crs=source_crs,
                    storage_crs=source_crs,
                    raster_profile=metadata,
                    bounds=DatasetService._extract_raster_bounds_json(metadata),
                    resolution=resolution,
                    checksum_sha256=storage_info["checksum_sha256"],
                    computed_checksum_sha256=checksum_sha256,
                    content=raw,
                    source_registry_id=str(source_registry.id),
                    source_snapshot_id=str(source_snapshot.id),
                    imported_at=imported_at,
                    metadata=contract_metadata,
                    observed_at=temporal["observed_at"],
                    valid_from=temporal["valid_from"],
                    valid_to=temporal["valid_to"],
                    temporal_unknown_reason=raw_source_metadata["temporal_unknown_reason"],
                    source_version=temporal["source_version"],
                    source_version_unknown_reason=raw_source_metadata["source_version_unknown_reason"],
                )
            )

        bounds_json = metadata.get("bounds_json")
        resolution_json = metadata.get("resolution_json")
        bands_json = metadata.get("bands_json")
        if canonical_type == "raster":
            bounds_json = DatasetService._extract_raster_bounds_json(metadata)
            resolution_json = DatasetService._extract_raster_resolution_json(metadata)
            bands_json = DatasetService._extract_raster_bands_json(metadata)
        dataset = Dataset(
            id=dataset_id,
            project_id=project_id,
            area_id=area_id,
            name=filename,
            dataset_type=canonical_type,
            source="manual_upload",
            dataset_role=normalized_role,
            source_name="manual",
            reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
            source_metadata=raw_source_metadata,
            provenance_metadata=raw_provenance_metadata,
            imported_at=imported_at,
            ingest_key=ingest_key,
            **temporal,
            storage_path=storage_info["storage_path"],
            original_filename=storage_info["original_filename"],
            stored_filename=storage_info["stored_filename"],
            content_type=storage_info["content_type"],
            size_bytes=storage_info["size_bytes"],
            checksum_sha256=storage_info["checksum_sha256"],
            crs=(DatasetService.CANONICAL_VECTOR_CRS if canonical_type == "vector" else source_crs),
            bounds_json=bounds_json,
            resolution_json=resolution_json,
            bands_json=bands_json,
            metadata_json=contract_metadata,
            status="validating",
        )
        dataset_version = DatasetService._new_dataset_version(dataset, ingest_key=ingest_key)
        try:
            db.add(dataset)
            db.add(dataset_version)
            db.flush()
            DatasetService._apply_validation_report(
                db,
                dataset=dataset,
                dataset_version=dataset_version,
                report=report,
                source=source_registry,
                snapshot=source_snapshot,
                artifact_path=storage_info["storage_path"],
            )
            if report.validation_status == ValidationStatus.PASSED and canonical_vector_payload is not None:
                VectorFeatureService.persist_geojson_features(
                    db=db,
                    dataset_id=dataset.id,
                    payload=canonical_vector_payload,
                    feature_class=reference_layer_name if normalized_role == "reference" else None,
                    source_crs=DatasetService.CANONICAL_VECTOR_CRS,
                    commit=False,
                )
            db.commit()
            db.refresh(dataset)
        except Exception:
            db.rollback()
            # Keep the staged bytes. A transport/database failure must remain
            # inspectable instead of silently deleting the only evidence.
            raise
        return DatasetService._to_response(dataset)

    @staticmethod
    def _import_vector_bytes_legacy(
        db: Session,
        *,
        project_id: UUID,
        filename: str,
        content: bytes,
        source: str,
        source_name: str,
        dataset_role: str,
        reference_layer_name: str | None,
        source_metadata: dict[str, Any],
        provenance_metadata: dict[str, Any],
        area_id: UUID | None = None,
        temporal_series_key: str | None = None,
        observed_at: datetime | None = None,
        valid_from: datetime | None = None,
        valid_to: datetime | None = None,
        temporal_granularity: str | None = None,
        source_version: str | None = None,
        content_type: str = "application/geo+json",
    ) -> DatasetCreateResponse:
        if not db.get(Project, project_id):
            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
        if area_id is not None:
            area = db.get(Area, area_id)
            if not area:
                raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
            if area.project_id != project_id:
                raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400)
        if not content:
            raise AppError(code="INVALID_UPLOAD", message="Vector artifact is empty", status_code=400)
        safe_filename = DatasetService._validate_upload_filename(filename)
        if DatasetService._extension_for_path(safe_filename) not in DatasetService.VECTOR_EXTENSIONS:
            raise AppError(code="INVALID_UPLOAD", message="Vector artifacts require .geojson or .json files", status_code=415)
        normalized_role = DatasetService._normalize_dataset_role(dataset_role)
        normalized_source_name = (source_name or "").strip() or ("manual" if normalized_role == "reference" else None)
        temporal = DatasetService._validate_temporal_metadata(
            temporal_series_key=temporal_series_key,
            observed_at=observed_at,
            valid_from=valid_from,
            valid_to=valid_to,
            temporal_granularity=temporal_granularity,
            source_version=source_version,
        )
        try:
            text = content.decode("utf-8")
        except UnicodeDecodeError as exc:
            raise AppError(code="INVALID_UPLOAD", message="Vector artifact must be UTF-8 encoded", status_code=400) from exc
        try:
            metadata = parse_geojson_payload(text)
            vector_payload = json.loads(text)
        except (ValueError, json.JSONDecodeError) as exc:
            raise AppError(code="INVALID_GEOJSON", message=str(exc), status_code=400) from exc

        dataset_id = uuid.uuid4()
        storage_info = StorageService.persist_dataset_file(
            project_id=str(project_id),
            dataset_id=str(dataset_id),
            dataset_type="vector",
            original_filename=safe_filename,
            content=content,
            content_type=content_type,
        )
        dataset = Dataset(
            id=dataset_id,
            project_id=project_id,
            area_id=area_id,
            name=safe_filename,
            dataset_type="vector",
            source=source,
            dataset_role=normalized_role,
            source_name=normalized_source_name,
            reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
            source_metadata=source_metadata,
            provenance_metadata=provenance_metadata,
            imported_at=datetime.now(timezone.utc),
            **temporal,
            storage_path=storage_info["storage_path"],
            original_filename=storage_info["original_filename"],
            stored_filename=storage_info["stored_filename"],
            content_type=storage_info["content_type"],
            size_bytes=storage_info["size_bytes"],
            checksum_sha256=storage_info["checksum_sha256"],
            crs=metadata.get("crs"),
            bounds_json=metadata.get("bounds_json"),
            metadata_json=metadata,
            status="ready",
        )
        try:
            db.add(dataset)
            db.add(
                DatasetVersion(
                    dataset_id=dataset.id,
                    version=1,
                    storage_path=dataset.storage_path,
                    source_version=dataset.source_version,
                    observed_at=dataset.observed_at,
                    valid_from=dataset.valid_from,
                    valid_to=dataset.valid_to,
                    checksum_sha256=dataset.checksum_sha256,
                    source_metadata=dataset.source_metadata,
                    provenance_metadata=dataset.provenance_metadata,
                )
            )
            VectorFeatureService.persist_geojson_features(
                db=db,
                dataset_id=dataset.id,
                payload=vector_payload,
                feature_class=reference_layer_name if normalized_role == "reference" else None,
                commit=False,
            )
            db.commit()
            db.refresh(dataset)
        except Exception:
            db.rollback()
            StorageService.remove_dataset_file(storage_info["storage_path"])
            raise
        return DatasetService._to_response(dataset)

    @staticmethod
    def _import_raster_bytes_legacy(
        db: Session,
        *,
        project_id: UUID,
        filename: str,
        content: bytes,
        source: str,
        source_name: str,
        source_metadata: dict[str, Any],
        provenance_metadata: dict[str, Any],
        area_id: UUID | None = None,
        temporal_series_key: str | None = None,
        observed_at: datetime | None = None,
        valid_from: datetime | None = None,
        valid_to: datetime | None = None,
        temporal_granularity: str | None = None,
        source_version: str | None = None,
        content_type: str = "image/tiff",
    ) -> DatasetCreateResponse:
        if not db.get(Project, project_id):
            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
        if area_id is not None:
            area = db.get(Area, area_id)
            if not area:
                raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
            if area.project_id != project_id:
                raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400)
        if not content:
            raise AppError(code="INVALID_UPLOAD", message="Raster artifact is empty", status_code=400)
        safe_filename = DatasetService._validate_upload_filename(filename)
        if DatasetService._extension_for_path(safe_filename) not in DatasetService.RASTER_EXTENSIONS:
            raise AppError(code="INVALID_UPLOAD", message="Raster artifacts require a GeoTIFF filename", status_code=415)
        temporal = DatasetService._validate_temporal_metadata(
            temporal_series_key=temporal_series_key,
            observed_at=observed_at,
            valid_from=valid_from,
            valid_to=valid_to,
            temporal_granularity=temporal_granularity,
            source_version=source_version,
        )

        dataset_id = uuid.uuid4()
        storage_info = StorageService.persist_dataset_file(
            project_id=str(project_id),
            dataset_id=str(dataset_id),
            dataset_type="raster",
            original_filename=safe_filename,
            content=content,
            content_type=content_type,
        )
        try:
            metadata = extract_raster_metadata(storage_info["storage_path"])
            dataset = Dataset(
                id=dataset_id,
                project_id=project_id,
                area_id=area_id,
                name=safe_filename,
                dataset_type="raster",
                source=source,
                dataset_role="source",
                source_name=source_name,
                source_metadata=source_metadata,
                provenance_metadata=provenance_metadata,
                imported_at=datetime.now(timezone.utc),
                **temporal,
                storage_path=storage_info["storage_path"],
                original_filename=storage_info["original_filename"],
                stored_filename=storage_info["stored_filename"],
                content_type=storage_info["content_type"],
                size_bytes=storage_info["size_bytes"],
                checksum_sha256=storage_info["checksum_sha256"],
                crs=metadata.get("crs"),
                bounds_json=DatasetService._extract_raster_bounds_json(metadata),
                resolution_json=DatasetService._extract_raster_resolution_json(metadata),
                bands_json=DatasetService._extract_raster_bands_json(metadata),
                metadata_json=metadata,
                status="ready",
            )
            db.add(dataset)
            db.add(
                DatasetVersion(
                    dataset_id=dataset.id,
                    version=1,
                    storage_path=dataset.storage_path,
                    source_version=dataset.source_version,
                    observed_at=dataset.observed_at,
                    valid_from=dataset.valid_from,
                    valid_to=dataset.valid_to,
                    checksum_sha256=dataset.checksum_sha256,
                    source_metadata=dataset.source_metadata,
                    provenance_metadata=dataset.provenance_metadata,
                )
            )
            db.commit()
            db.refresh(dataset)
            return DatasetService._to_response(dataset)
        except Exception:
            db.rollback()
            StorageService.remove_dataset_file(storage_info["storage_path"])
            raise

    @staticmethod
    def _governed_import_bytes(
        db: Session,
        *,
        project_id: UUID,
        filename: str,
        content: bytes,
        dataset_type: str,
        source: str,
        source_name: str,
        dataset_role: str,
        reference_layer_name: str | None,
        source_metadata: dict[str, Any] | None,
        provenance_metadata: dict[str, Any] | None,
        area_id: UUID | None,
        temporal_series_key: str | None,
        observed_at: datetime | None,
        valid_from: datetime | None,
        valid_to: datetime | None,
        temporal_granularity: str | None,
        source_version: str | None,
        content_type: str,
    ) -> DatasetCreateResponse:
        """Persist an adapter-owned source through one governed ingestion path.

        Acquisition adapters choose an entry from the server-owned registry;
        they cannot create authority identities dynamically.  The original
        artifact is deliberately retained if parsing or validation fails so
        the immutable checksum, source snapshot and quarantine record remain
        reviewable.
        """
        if not db.get(Project, project_id):
            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
        if area_id is not None:
            area = db.get(Area, area_id)
            if not area:
                raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
            if area.project_id != project_id:
                raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400)
        if not content:
            raise AppError(code="INVALID_UPLOAD", message=f"{dataset_type.title()} artifact is empty", status_code=400)

        canonical_type = DatasetService._canonical_dataset_type(dataset_type)
        normalized_role = DatasetService._normalize_dataset_role(dataset_role)
        if normalized_role == "reference" and canonical_type == "raster":
            raise AppError(code="INVALID_DATASET_ROLE", message="reference role requires vector dataset type", status_code=400)
        safe_filename = DatasetService._validate_upload_filename(filename)
        extension = DatasetService._extension_for_path(safe_filename)
        allowed_extensions = DatasetService.VECTOR_EXTENSIONS if canonical_type == "vector" else DatasetService.RASTER_EXTENSIONS
        if extension not in allowed_extensions:
            expected = ".geojson or .json" if canonical_type == "vector" else ".tif, .tiff or .geotiff"
            raise AppError(code="INVALID_UPLOAD", message=f"{canonical_type.title()} artifacts require {expected} files", status_code=415)

        source_key = SourceRegistryService.normalize_source_key(source_name)
        # This lookup deliberately happens before writing the artifact. A
        # typo in an internal adapter must not acquire an unregistered source
        # identity or silently downgrade itself to a manual source.
        SourceRegistryService.definition_for(source_key)
        temporal = DatasetService._validate_temporal_metadata(
            temporal_series_key=temporal_series_key,
            observed_at=observed_at,
            valid_from=valid_from,
            valid_to=valid_to,
            temporal_granularity=temporal_granularity,
            source_version=source_version,
        )
        computed_checksum_sha256 = StorageService.calculate_checksum_sha256(content)
        ingest_key = DatasetService._ingest_key(
            project_id=project_id,
            source_key=source_key,
            checksum_sha256=computed_checksum_sha256,
            dataset_type=canonical_type,
            dataset_role=normalized_role,
            area_id=area_id,
            reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
            source_version=temporal["source_version"],
        )
        existing = DatasetService._find_existing_ingest(db, project_id, ingest_key)
        if existing is not None:
            return DatasetService._to_response(existing)

        governed_source_metadata = dict(source_metadata or {})
        governed_provenance_metadata = dict(provenance_metadata or {})
        governed_source_metadata.update(
            {
                "ingest_origin": "governed_acquisition_adapter",
                "source_registry_key": source_key,
                "authority_claim_accepted": True,
            }
        )
        governed_source_metadata.setdefault(
            "temporal_unknown_reason",
            "The governed source did not publish a precise observation timestamp for this snapshot.",
        )
        governed_source_metadata.setdefault(
            "source_version_unknown_reason",
            "The governed source did not publish a stable source edition; the immutable checksum identifies this snapshot.",
        )
        governed_provenance_metadata.update(
            {
                "ingest_origin": "governed_acquisition_adapter",
                "source_registry_key": source_key,
                "ingest_key": ingest_key,
            }
        )

        dataset_id = uuid.uuid4()
        storage_info: dict[str, Any] | None = None
        storage_content = content
        source_evidence: dict[str, Any] | None = None
        imported_at = datetime.now(timezone.utc)
        metadata: dict[str, Any] = {"dataset_type": canonical_type}
        source_crs: str | None = None
        canonical_vector_payload: dict[str, Any] | None = None
        parser_error: tuple[str, str] | None = None
        try:
            if canonical_type == "vector":
                try:
                    payload = json.loads(content.decode("utf-8"))
                except UnicodeDecodeError as exc:
                    raise AppError(code="INVALID_UPLOAD", message="Vector artifact must be UTF-8 encoded", status_code=400) from exc
                raw_metadata = parse_geojson_payload(payload)
                source_crs = str(raw_metadata.get("crs") or "").strip() or None
                canonical_vector_payload = VectorFeatureService.canonicalize_geojson_payload(
                    payload,
                    source_crs=source_crs or DatasetService.CANONICAL_VECTOR_CRS,
                )
                metadata = parse_geojson_payload(canonical_vector_payload)
                metadata.update(
                    {
                        "dataset_type": "vector",
                        "source_crs": source_crs,
                        "source_bounds_json": raw_metadata.get("bounds_json"),
                        "source_crs_assumed": raw_metadata.get("crs_assumed", False),
                        "canonical_storage_crs": DatasetService.CANONICAL_VECTOR_CRS,
                    }
                )
                if DatasetService._vector_storage_requires_canonicalization(source_crs):
                    storage_content = DatasetService._canonical_vector_storage_bytes(canonical_vector_payload)
                    source_evidence = DatasetService._persist_vector_source_evidence(
                        project_id=project_id,
                        dataset_id=dataset_id,
                        original_filename=safe_filename,
                        content=content,
                        content_type=content_type,
                    )
            else:
                storage_info = StorageService.persist_dataset_file(
                    project_id=str(project_id),
                    dataset_id=str(dataset_id),
                    dataset_type=canonical_type,
                    original_filename=safe_filename,
                    content=content,
                    content_type=content_type,
                )
                metadata = extract_raster_metadata(storage_info["storage_path"])
                metadata["dataset_type"] = "raster"
                source_crs = str(metadata.get("crs") or "").strip() or None
        except (ValueError, json.JSONDecodeError, AppError) as exc:
            code = exc.code if isinstance(exc, AppError) else "INVALID_GEOJSON"
            parser_error = (code, str(exc))
            metadata = {"dataset_type": canonical_type, "processing_error": str(exc), "processing_code": code}

        if storage_info is None:
            storage_info = StorageService.persist_dataset_file(
                project_id=str(project_id),
                dataset_id=str(dataset_id),
                dataset_type=canonical_type,
                original_filename=safe_filename,
                content=storage_content,
                content_type=content_type,
            )
        computed_storage_checksum_sha256 = StorageService.calculate_checksum_sha256(storage_content)
        if source_evidence is not None:
            resolved_source_crs = source_crs or DatasetService.CANONICAL_VECTOR_CRS
            DatasetService._record_vector_source_evidence(
                source_metadata=governed_source_metadata,
                provenance_metadata=governed_provenance_metadata,
                source_crs=resolved_source_crs,
                evidence=source_evidence,
                canonical_checksum_sha256=computed_storage_checksum_sha256,
            )
            metadata.update(
                {
                    "source_artifact_checksum_sha256": source_evidence["checksum_sha256"],
                    "canonical_artifact_checksum_sha256": computed_storage_checksum_sha256,
                }
            )

        source_registry, source_snapshot = DatasetService._record_snapshot(
            db,
            source_key=source_key,
            checksum_sha256=storage_info["checksum_sha256"],
            source_version=temporal["source_version"],
            observed_at=temporal["observed_at"],
            valid_from=temporal["valid_from"],
            valid_to=temporal["valid_to"],
            source_crs=source_crs,
            source_metadata=governed_source_metadata,
            metadata=metadata,
        )
        contract_metadata = DatasetService._contract_metadata(
            metadata=metadata,
            source_metadata=governed_source_metadata,
            provenance_metadata=governed_provenance_metadata,
            source=source_registry,
        )
        if parser_error is not None:
            report = DatasetService._failed_validation_report(
                asset_id=ingest_key,
                dataset_type=canonical_type,
                code=parser_error[0],
                message=parser_error[1],
                now=imported_at,
            )
        elif canonical_type == "vector":
            resolved_source_crs = source_crs or DatasetService.CANONICAL_VECTOR_CRS
            lineage = LineageEvidence()
            if resolved_source_crs.upper() != DatasetService.CANONICAL_VECTOR_CRS:
                lineage = LineageEvidence(
                    transformations=(
                        TransformationEvidence(
                            name="vector_crs_normalization",
                            version="1.0.0",
                            checksum_sha256=DatasetService._stable_hash(
                                {"source_crs": resolved_source_crs, "storage_crs": DatasetService.CANONICAL_VECTOR_CRS}
                            ),
                        ),
                    )
                )
            try:
                contract_metadata["source_schema_validation"] = DatasetService._validate_vector_source_schema(
                    source_registry,
                    canonical_vector_payload or {"type": "FeatureCollection", "features": []},
                )
                report = validate_registered_asset(
                    build_vector_ingest_input(
                        asset_id=ingest_key,
                        source_crs=resolved_source_crs,
                        storage_crs=DatasetService.CANONICAL_VECTOR_CRS,
                        feature_collection=canonical_vector_payload or {"type": "FeatureCollection", "features": []},
                        checksum_sha256=storage_info["checksum_sha256"],
                        computed_checksum_sha256=computed_storage_checksum_sha256,
                        content=storage_content,
                        source_registry_id=str(source_registry.id),
                        source_snapshot_id=str(source_snapshot.id),
                        imported_at=imported_at,
                        metadata=contract_metadata,
                        observed_at=temporal["observed_at"],
                        valid_from=temporal["valid_from"],
                        valid_to=temporal["valid_to"],
                        temporal_unknown_reason=governed_source_metadata["temporal_unknown_reason"],
                        source_version=temporal["source_version"],
                        source_version_unknown_reason=governed_source_metadata["source_version_unknown_reason"],
                        lineage=lineage,
                    )
                )
            except AppError as exc:
                report = DatasetService._failed_validation_report(
                    asset_id=ingest_key,
                    dataset_type="vector",
                    code=exc.code,
                    message=exc.message,
                    now=imported_at,
                    category="source_schema",
                )
        else:
            resolution_json = DatasetService._extract_raster_resolution_json(metadata)
            resolution = (
                {
                    "x": resolution_json["x"],
                    "y": resolution_json["y"],
                    "unit": DatasetService._resolution_unit_for_crs(source_crs),
                }
                if resolution_json
                else None
            )
            report = validate_registered_asset(
                build_raster_ingest_input(
                    asset_id=ingest_key,
                    source_crs=source_crs,
                    storage_crs=source_crs,
                    raster_profile=metadata,
                    bounds=DatasetService._extract_raster_bounds_json(metadata),
                    resolution=resolution,
                    checksum_sha256=storage_info["checksum_sha256"],
                    computed_checksum_sha256=computed_checksum_sha256,
                    content=content,
                    source_registry_id=str(source_registry.id),
                    source_snapshot_id=str(source_snapshot.id),
                    imported_at=imported_at,
                    metadata=contract_metadata,
                    observed_at=temporal["observed_at"],
                    valid_from=temporal["valid_from"],
                    valid_to=temporal["valid_to"],
                    temporal_unknown_reason=governed_source_metadata["temporal_unknown_reason"],
                    source_version=temporal["source_version"],
                    source_version_unknown_reason=governed_source_metadata["source_version_unknown_reason"],
                )
            )

        bounds_json = metadata.get("bounds_json")
        resolution_json = metadata.get("resolution_json")
        bands_json = metadata.get("bands_json")
        if canonical_type == "raster":
            bounds_json = DatasetService._extract_raster_bounds_json(metadata)
            resolution_json = DatasetService._extract_raster_resolution_json(metadata)
            bands_json = DatasetService._extract_raster_bands_json(metadata)
        dataset = Dataset(
            id=dataset_id,
            project_id=project_id,
            area_id=area_id,
            name=safe_filename,
            dataset_type=canonical_type,
            source=source,
            dataset_role=normalized_role,
            source_name=source_key,
            reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
            source_metadata=governed_source_metadata,
            provenance_metadata=governed_provenance_metadata,
            imported_at=imported_at,
            ingest_key=ingest_key,
            **temporal,
            storage_path=storage_info["storage_path"],
            original_filename=storage_info["original_filename"],
            stored_filename=storage_info["stored_filename"],
            content_type=storage_info["content_type"],
            size_bytes=storage_info["size_bytes"],
            checksum_sha256=storage_info["checksum_sha256"],
            crs=(DatasetService.CANONICAL_VECTOR_CRS if canonical_type == "vector" else source_crs),
            bounds_json=bounds_json,
            resolution_json=resolution_json,
            bands_json=bands_json,
            metadata_json=contract_metadata,
            status="validating",
        )
        dataset_version = DatasetService._new_dataset_version(dataset, ingest_key=ingest_key)
        try:
            db.add(dataset)
            db.add(dataset_version)
            db.flush()
            DatasetService._apply_validation_report(
                db,
                dataset=dataset,
                dataset_version=dataset_version,
                report=report,
                source=source_registry,
                snapshot=source_snapshot,
                artifact_path=storage_info["storage_path"],
            )
            if report.validation_status == ValidationStatus.PASSED and canonical_vector_payload is not None:
                VectorFeatureService.persist_geojson_features(
                    db=db,
                    dataset_id=dataset.id,
                    payload=canonical_vector_payload,
                    feature_class=reference_layer_name if normalized_role == "reference" else None,
                    source_crs=DatasetService.CANONICAL_VECTOR_CRS,
                    commit=False,
                )
            db.commit()
            db.refresh(dataset)
        except Exception:
            db.rollback()
            # Leave the staged artifact untouched. A failed persistence
            # transaction is not evidence that the source bytes were safe to
            # delete or that an acquisition can be repeated silently.
            raise
        return DatasetService._to_response(dataset)

    @staticmethod
    def import_vector_bytes(
        db: Session,
        *,
        project_id: UUID,
        filename: str,
        content: bytes,
        source: str,
        source_name: str,
        dataset_role: str,
        reference_layer_name: str | None,
        source_metadata: dict[str, Any],
        provenance_metadata: dict[str, Any],
        area_id: UUID | None = None,
        temporal_series_key: str | None = None,
        observed_at: datetime | None = None,
        valid_from: datetime | None = None,
        valid_to: datetime | None = None,
        temporal_granularity: str | None = None,
        source_version: str | None = None,
        content_type: str = "application/geo+json",
    ) -> DatasetCreateResponse:
        if not DatasetService._registry_persistence_available(db):
            return DatasetService._import_vector_bytes_legacy(
                db,
                project_id=project_id,
                filename=filename,
                content=content,
                source=source,
                source_name=source_name,
                dataset_role=dataset_role,
                reference_layer_name=reference_layer_name,
                source_metadata=source_metadata,
                provenance_metadata=provenance_metadata,
                area_id=area_id,
                temporal_series_key=temporal_series_key,
                observed_at=observed_at,
                valid_from=valid_from,
                valid_to=valid_to,
                temporal_granularity=temporal_granularity,
                source_version=source_version,
                content_type=content_type,
            )
        return DatasetService._governed_import_bytes(
            db,
            project_id=project_id,
            filename=filename,
            content=content,
            dataset_type="vector",
            source=source,
            source_name=source_name,
            dataset_role=dataset_role,
            reference_layer_name=reference_layer_name,
            source_metadata=source_metadata,
            provenance_metadata=provenance_metadata,
            area_id=area_id,
            temporal_series_key=temporal_series_key,
            observed_at=observed_at,
            valid_from=valid_from,
            valid_to=valid_to,
            temporal_granularity=temporal_granularity,
            source_version=source_version,
            content_type=content_type,
        )

    @staticmethod
    def import_raster_bytes(
        db: Session,
        *,
        project_id: UUID,
        filename: str,
        content: bytes,
        source: str,
        source_name: str,
        source_metadata: dict[str, Any],
        provenance_metadata: dict[str, Any],
        area_id: UUID | None = None,
        temporal_series_key: str | None = None,
        observed_at: datetime | None = None,
        valid_from: datetime | None = None,
        valid_to: datetime | None = None,
        temporal_granularity: str | None = None,
        source_version: str | None = None,
        content_type: str = "image/tiff",
    ) -> DatasetCreateResponse:
        if not DatasetService._registry_persistence_available(db):
            return DatasetService._import_raster_bytes_legacy(
                db,
                project_id=project_id,
                filename=filename,
                content=content,
                source=source,
                source_name=source_name,
                source_metadata=source_metadata,
                provenance_metadata=provenance_metadata,
                area_id=area_id,
                temporal_series_key=temporal_series_key,
                observed_at=observed_at,
                valid_from=valid_from,
                valid_to=valid_to,
                temporal_granularity=temporal_granularity,
                source_version=source_version,
                content_type=content_type,
            )
        return DatasetService._governed_import_bytes(
            db,
            project_id=project_id,
            filename=filename,
            content=content,
            dataset_type="raster",
            source=source,
            source_name=source_name,
            dataset_role="source",
            reference_layer_name=None,
            source_metadata=source_metadata,
            provenance_metadata=provenance_metadata,
            area_id=area_id,
            temporal_series_key=temporal_series_key,
            observed_at=observed_at,
            valid_from=valid_from,
            valid_to=valid_to,
            temporal_granularity=temporal_granularity,
            source_version=source_version,
            content_type=content_type,
        )

    @staticmethod
    def import_partitioned_vector_artifact(
        db: Session,
        *,
        project_id: UUID,
        area_id: UUID,
        artifact_path: str | Path,
        partition_paths: list[str | Path],
        original_filename: str,
        source: str,
        dataset_role: str,
        source_name: str,
        reference_layer_name: str | None,
        metadata_json: dict[str, Any],
        source_metadata: dict[str, Any],
        provenance_metadata: dict[str, Any],
        temporal_series_key: str,
        observed_at: datetime,
        temporal_granularity: str = "snapshot",
        source_version: str | None = None,
        batch_size: int = 1000,
    ) -> DatasetCreateResponse:
        if not DatasetService._registry_persistence_available(db):
            raise AppError(
                code="SOURCE_REGISTRY_PERSISTENCE_UNAVAILABLE",
                message="Partitioned authoritative imports require registry and provenance persistence.",
                status_code=503,
            )
        if not db.get(Project, project_id):
            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
        area = db.get(Area, area_id)
        if not area:
            raise AppError(code="AREA_NOT_FOUND", message="Area not found", status_code=404)
        if area.project_id != project_id:
            raise AppError(code="INVALID_DATASET_SCOPE", message="Area does not belong to this project", status_code=400)
        if not partition_paths:
            raise AppError(
                code="INVALID_GEOJSON_PARTITIONS",
                message="At least one GeoJSON partition is required",
                status_code=400,
            )

        filename = DatasetService._validate_upload_filename(original_filename)
        if DatasetService._extension_for_path(filename) not in DatasetService.VECTOR_EXTENSIONS:
            raise AppError(code="INVALID_UPLOAD", message="Vector artifacts require .geojson or .json files", status_code=415)
        normalized_role = DatasetService._normalize_dataset_role(dataset_role)
        source_key = SourceRegistryService.normalize_source_key(source_name)
        # A partitioned operator artifact is never allowed to manufacture a
        # source identity from its caller-provided label.
        SourceRegistryService.definition_for(source_key)
        temporal = DatasetService._validate_temporal_metadata(
            temporal_series_key=temporal_series_key,
            observed_at=observed_at,
            valid_from=observed_at,
            valid_to=None,
            temporal_granularity=temporal_granularity,
            source_version=source_version,
        )
        metadata = dict(metadata_json)
        expected_feature_count = int(metadata.get("feature_count") or 0)
        if expected_feature_count <= 0:
            raise AppError(
                code="INVALID_GEOJSON_PARTITIONS",
                message="Partition metadata must declare a positive feature_count",
                status_code=400,
            )

        artifact_checksum_sha256 = DatasetService._calculate_file_checksum_sha256(artifact_path)
        ingest_key = DatasetService._ingest_key(
            project_id=project_id,
            source_key=source_key,
            checksum_sha256=artifact_checksum_sha256,
            dataset_type="vector",
            dataset_role=normalized_role,
            area_id=area_id,
            reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
            source_version=temporal["source_version"],
        )
        existing = DatasetService._find_existing_ingest(db, project_id, ingest_key)
        if existing is not None:
            return DatasetService._to_response(existing)

        governed_source_metadata = dict(source_metadata or {})
        governed_provenance_metadata = dict(provenance_metadata or {})
        governed_source_metadata.update(
            {
                "ingest_origin": "governed_partitioned_acquisition_adapter",
                "source_registry_key": source_key,
                "authority_claim_accepted": True,
                "partitioned_artifact": True,
            }
        )
        governed_source_metadata.setdefault(
            "temporal_unknown_reason",
            "The governed source did not publish a precise observation timestamp for this snapshot.",
        )
        governed_source_metadata.setdefault(
            "source_version_unknown_reason",
            "The governed source did not publish a stable source edition; the immutable checksum identifies this snapshot.",
        )
        governed_provenance_metadata.update(
            {
                "ingest_origin": "governed_partitioned_acquisition_adapter",
                "source_registry_key": source_key,
                "ingest_key": ingest_key,
                "artifact_checksum_sha256": artifact_checksum_sha256,
                "combined_artifact_checksum_sha256": artifact_checksum_sha256,
                "partition_count": len(partition_paths),
            }
        )

        dataset_id = uuid.uuid4()
        storage_info = StorageService.persist_dataset_file_from_path(
            project_id=str(project_id),
            dataset_id=str(dataset_id),
            dataset_type="vector",
            original_filename=filename,
            source_path=artifact_path,
            content_type="application/geo+json",
        )
        # The source file may have changed while it was copied. Re-key on the
        # bytes actually retained; never bind a snapshot to a stale pre-copy
        # checksum.
        persisted_checksum_sha256 = str(storage_info["checksum_sha256"])
        if persisted_checksum_sha256 != artifact_checksum_sha256:
            ingest_key = DatasetService._ingest_key(
                project_id=project_id,
                source_key=source_key,
                checksum_sha256=persisted_checksum_sha256,
                dataset_type="vector",
                dataset_role=normalized_role,
                area_id=area_id,
                reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
                source_version=temporal["source_version"],
            )
            existing = DatasetService._find_existing_ingest(db, project_id, ingest_key)
            if existing is not None:
                StorageService.remove_dataset_file(str(storage_info["storage_path"]))
                return DatasetService._to_response(existing)
            governed_provenance_metadata["ingest_key"] = ingest_key
            governed_provenance_metadata["artifact_checksum_sha256"] = persisted_checksum_sha256
            governed_provenance_metadata["combined_artifact_checksum_sha256"] = persisted_checksum_sha256

        storage_crs = str(
            metadata.get("canonical_storage_crs")
            or metadata.get("storage_crs")
            or metadata.get("crs")
            or DatasetService.CANONICAL_VECTOR_CRS
        ).strip()
        source_crs = str(metadata.get("source_crs") or storage_crs).strip() or None
        metadata.update(
            {
                "dataset_type": "vector",
                "canonical_storage_crs": DatasetService.CANONICAL_VECTOR_CRS,
                "partitioned_artifact": True,
                "partition_count": len(partition_paths),
            }
        )
        source_registry, source_snapshot = DatasetService._record_snapshot(
            db,
            source_key=source_key,
            checksum_sha256=persisted_checksum_sha256,
            source_version=temporal["source_version"],
            observed_at=temporal["observed_at"],
            valid_from=temporal["valid_from"],
            valid_to=temporal["valid_to"],
            source_crs=source_crs,
            source_metadata=governed_source_metadata,
            metadata=metadata,
        )
        # Registry persistence is checked above, so absence here is an
        # infrastructure fault rather than a state that can be imported.
        if source_registry is None or source_snapshot is None:  # pragma: no cover - defensive invariant
            raise AppError(
                code="SOURCE_REGISTRY_PERSISTENCE_UNAVAILABLE",
                message="Source registry persistence did not return a governed snapshot.",
                status_code=503,
            )
        contract_metadata = DatasetService._contract_metadata(
            metadata=metadata,
            source_metadata=governed_source_metadata,
            provenance_metadata=governed_provenance_metadata,
            source=source_registry,
        )
        partition_records = _PartitionedGeoJsonRecords(
            partition_paths,
            expected_feature_count=expected_feature_count,
            declared_partition_checksums=governed_provenance_metadata.get("partition_checksums"),
            source_schema=_SourceVectorSchema.from_source(source_registry),
        )
        lineage = LineageEvidence()
        if source_crs and storage_crs.upper() != source_crs.upper():
            lineage = LineageEvidence(
                transformations=(
                    TransformationEvidence(
                        name="partitioned_vector_crs_normalization",
                        version="1.0.0",
                        checksum_sha256=DatasetService._stable_hash(
                            {
                                "source_crs": source_crs,
                                "storage_crs": storage_crs,
                                "partition_count": len(partition_paths),
                            }
                        ),
                    ),
                )
            )
        declared_artifact_checksum = str(governed_provenance_metadata.get("artifact_sha256") or "").strip().lower()
        artifact_binding_error: tuple[str, str] | None = None
        if not declared_artifact_checksum:
            artifact_binding_error = (
                "ARTIFACT_CHECKSUM_REQUIRED",
                "Partitioned ingestion requires the acquisition manifest's combined artifact checksum.",
            )
        elif not _CHECKSUM_SHA256.fullmatch(declared_artifact_checksum):
            artifact_binding_error = (
                "ARTIFACT_CHECKSUM_INVALID",
                "Declared partitioned artifact checksum must be a lowercase SHA-256 value.",
            )
        elif declared_artifact_checksum != persisted_checksum_sha256:
            artifact_binding_error = (
                "ARTIFACT_CHECKSUM_MISMATCH",
                "Declared artifact checksum does not match the retained partitioned artifact.",
            )
        try:
            partition_audit = partition_records.audit()
            contract_metadata["partitioned_geometry_audit"] = partition_audit.to_metadata()
            contract_metadata["source_schema_validation"] = partition_audit.source_schema_validation
            governed_provenance_metadata["partition_checksum_manifest_sha256"] = DatasetService._stable_hash(
                partition_audit.partition_checksums_sha256
            )
            governed_provenance_metadata["partitioned_artifact_binding_sha256"] = DatasetService._stable_hash(
                {
                    "combined_artifact_checksum_sha256": persisted_checksum_sha256,
                    "partition_checksum_manifest_sha256": governed_provenance_metadata[
                        "partition_checksum_manifest_sha256"
                    ],
                    "feature_count": partition_audit.feature_count,
                    "storage_crs": storage_crs,
                }
            )
            if artifact_binding_error is not None:
                report = DatasetService._failed_validation_report(
                    asset_id=ingest_key,
                    dataset_type="vector",
                    code=artifact_binding_error[0],
                    message=artifact_binding_error[1],
                    now=datetime.now(timezone.utc),
                    category="checksum",
                )
            else:
                report = validate_registered_asset(
                    DataAssetValidationInput(
                        asset_id=ingest_key,
                        data_contract_key=VECTOR_GEOJSON_CONTRACT_KEY,
                        data_contract_version=VECTOR_GEOJSON_CONTRACT_VERSION,
                        kind=ContractKind.VECTOR,
                        source_crs=source_crs,
                        storage_crs=storage_crs,
                        bounds=contract_metadata.get("bounds_json"),
                        checksum_sha256=persisted_checksum_sha256,
                        computed_checksum_sha256=persisted_checksum_sha256,
                        metadata=contract_metadata,
                        # The partition-bounded audit above validates every
                        # source feature. The generic contract receives only
                        # compact aggregate geometry evidence and therefore
                        # cannot materialize the complete regional artifact.
                        geometry_records=(partition_audit.representative_record,),
                        source_registry_id=str(source_registry.id),
                        source_snapshot_id=str(source_snapshot.id),
                        lineage=lineage,
                        imported_at=datetime.now(timezone.utc),
                        observed_at=temporal["observed_at"],
                        valid_from=temporal["valid_from"],
                        valid_to=temporal["valid_to"],
                        temporal_unknown_reason=governed_source_metadata["temporal_unknown_reason"],
                        source_version=temporal["source_version"],
                        source_version_unknown_reason=governed_source_metadata[
                            "source_version_unknown_reason"
                        ],
                    )
                )
        except AppError as exc:
            report = DatasetService._failed_validation_report(
                asset_id=ingest_key,
                dataset_type="vector",
                code=exc.code,
                message=exc.message,
                now=datetime.now(timezone.utc),
                category="source_schema" if exc.code.startswith("SOURCE_SCHEMA") else "parser",
            )
        dataset = Dataset(
            id=dataset_id,
            project_id=project_id,
            area_id=area_id,
            name=filename,
            dataset_type="vector",
            source=source,
            dataset_role=normalized_role,
            source_name=source_key,
            reference_layer_name=reference_layer_name if normalized_role == "reference" else None,
            source_metadata=governed_source_metadata,
            provenance_metadata=governed_provenance_metadata,
            imported_at=datetime.now(timezone.utc),
            **temporal,
            ingest_key=ingest_key,
            storage_path=storage_info["storage_path"],
            original_filename=storage_info["original_filename"],
            stored_filename=storage_info["stored_filename"],
            content_type=storage_info["content_type"],
            size_bytes=storage_info["size_bytes"],
            checksum_sha256=persisted_checksum_sha256,
            crs=storage_crs,
            bounds_json=metadata.get("bounds_json"),
            metadata_json=contract_metadata,
            status="validating",
        )
        dataset_version = DatasetService._new_dataset_version(dataset, ingest_key=ingest_key)
        try:
            db.add(dataset)
            db.add(dataset_version)
            db.flush()
            if report.validation_status == ValidationStatus.PASSED:
                try:
                    begin_nested = getattr(db, "begin_nested", None)
                    if callable(begin_nested):
                        with begin_nested():
                            persisted_count = VectorFeatureService.persist_geojson_partitions(
                                db,
                                dataset.id,
                                partition_paths,
                                feature_class=reference_layer_name if normalized_role == "reference" else None,
                                batch_size=batch_size,
                                source_crs=storage_crs,
                            )
                            if persisted_count != expected_feature_count:
                                raise AppError(
                                    code="PARTITION_FEATURE_COUNT_MISMATCH",
                                    message=(
                                        f"Regional artifact declares {expected_feature_count} features but "
                                        f"{persisted_count} queryable features were indexed"
                                    ),
                                    status_code=400,
                                )
                    else:  # lightweight test sessions only; production uses a savepoint
                        persisted_count = VectorFeatureService.persist_geojson_partitions(
                            db,
                            dataset.id,
                            partition_paths,
                            feature_class=reference_layer_name if normalized_role == "reference" else None,
                            batch_size=batch_size,
                            source_crs=storage_crs,
                        )
                        if persisted_count != expected_feature_count:
                            raise AppError(
                                code="PARTITION_FEATURE_COUNT_MISMATCH",
                                message=(
                                    f"Regional artifact declares {expected_feature_count} features but "
                                    f"{persisted_count} queryable features were indexed"
                                ),
                                status_code=400,
                            )
                except AppError as exc:
                    report = DatasetService._failed_validation_report(
                        asset_id=ingest_key,
                        dataset_type="vector",
                        code=exc.code,
                        message=exc.message,
                        now=datetime.now(timezone.utc),
                    )
            DatasetService._apply_validation_report(
                db,
                dataset=dataset,
                dataset_version=dataset_version,
                report=report,
                source=source_registry,
                snapshot=source_snapshot,
                artifact_path=str(storage_info["storage_path"]),
            )
            db.commit()
            db.refresh(dataset)
        except Exception:
            db.rollback()
            # Retain staged bytes for forensic review. A transaction error is
            # not evidence that the source artifact may be safely discarded.
            raise
        return DatasetService._to_response(dataset)

    @staticmethod
    def refresh_metadata(db: Session, dataset_id: UUID) -> DatasetCreateResponse:
        dataset = DatasetService._get_dataset(db, dataset_id)
        if dataset.quarantine_status == "quarantined" or dataset.status == "quarantined":
            raise AppError(
                code="DATASET_QUARANTINED",
                message="Quarantined datasets cannot be refreshed into an eligible state; re-ingest a new governed snapshot.",
                status_code=409,
            )
        # A governed dataset's source snapshot and validation report bind the
        # exact bytes, CRS and extracted metadata that were inspected at
        # ingest.  Re-reading a mutable storage path here would otherwise let
        # an in-place replacement change the operational artifact while its
        # persisted checksum/report still says ``passed``.  Such a change must
        # create a new immutable source snapshot through the governed ingest
        # path; metadata refresh remains intentionally available only to rows
        # without Phase-2 contract evidence.
        has_governed_contract_evidence = any(
            (
                dataset.source_registry_id is not None,
                dataset.source_snapshot_id is not None,
                bool(str(dataset.data_contract_key or "").strip()),
                bool(str(dataset.data_contract_version or "").strip()),
                dataset.validation_report_json is not None,
                dataset.validation_status == "passed",
            )
        )
        if has_governed_contract_evidence:
            raise AppError(
                code="GOVERNED_DATASET_REINGEST_REQUIRED",
                message=(
                    "Governed dataset metadata is immutable evidence. Re-ingest the artifact to create a new "
                    "source snapshot and validation report."
                ),
                status_code=409,
            )
        if not dataset.storage_path:
            raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404)
        if not Path(dataset.storage_path).exists():
            raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404)

        try:
            if DatasetService._is_vector_type(dataset.dataset_type):
                metadata = parse_geojson_payload(load_dataset_text(dataset.storage_path))
            elif DatasetService._is_raster_type(dataset.dataset_type):
                metadata = extract_raster_metadata(dataset.storage_path)
            else:
                raise AppError(code="INVALID_DATASET_TYPE", message="Cannot refresh metadata for this dataset type", status_code=400)
        except ValueError as exc:
            raise AppError(code="INVALID_GEOJSON", message=str(exc), status_code=400) from exc
        except AppError as exc:
            if DatasetService._is_raster_type(dataset.dataset_type) and exc.code == "RASTER_PROCESSING_UNAVAILABLE":
                metadata = {"processing_error": exc.message, "processing_code": exc.code}
            else:
                raise

        bounds_json = metadata.get("bounds_json") if isinstance(metadata, dict) else dataset.bounds_json
        resolution_json = metadata.get("resolution_json") if isinstance(metadata, dict) else dataset.resolution_json
        bands_json = metadata.get("bands_json") if isinstance(metadata, dict) else dataset.bands_json
        if DatasetService._is_raster_type(dataset.dataset_type) and isinstance(metadata, dict):
            bounds_json = DatasetService._extract_raster_bounds_json(metadata)
            resolution_json = DatasetService._extract_raster_resolution_json(metadata)
            bands_json = DatasetService._extract_raster_bands_json(metadata)

        # Metadata extraction is observational only. It must never turn an
        # unvalidated historical row into a ready, authoritative dataset.
        if DatasetService._is_vector_type(dataset.dataset_type):
            dataset.crs = DatasetService.CANONICAL_VECTOR_CRS
        else:
            dataset.crs = metadata.get("crs") if isinstance(metadata, dict) else dataset.crs
        dataset.bounds_json = bounds_json
        existing_metadata = dataset.metadata_json if isinstance(dataset.metadata_json, dict) else {}
        dataset.metadata_json = {**existing_metadata, **metadata}
        dataset.resolution_json = resolution_json
        dataset.bands_json = bands_json

        db.add(dataset)
        db.commit()
        db.refresh(dataset)

        return DatasetService._to_response(dataset)

    @staticmethod
    def update_temporal_metadata(db: Session, dataset_id: UUID, payload: DatasetTemporalUpdate) -> DatasetCreateResponse:
        dataset = DatasetService._get_dataset(db, dataset_id)
        if dataset.quarantine_status == "quarantined" or dataset.status == "quarantined":
            raise AppError(
                code="DATASET_QUARANTINED",
                message="Quarantined datasets require a new governed ingest rather than an in-place temporal edit.",
                status_code=409,
            )
        temporal = DatasetService._validate_temporal_metadata(**payload.model_dump())
        if all(getattr(dataset, field) == value for field, value in temporal.items()):
            return DatasetService._to_response(dataset)

        for field, value in temporal.items():
            setattr(dataset, field, value)

        # Observation/source-version fields are contract inputs. Their edit
        # invalidates the prior report, so later training/inference gates fail
        # closed until a governed re-ingest persists a new snapshot/report.
        dataset.status = "validating"
        dataset.validation_status = "not_validated"
        dataset.validation_report_json = None
        dataset.provenance_status = "incomplete"
        dataset.quarantine_status = "not_quarantined"

        latest_version = (
            db.query(DatasetVersion)
            .filter(DatasetVersion.dataset_id == dataset.id)
            .order_by(DatasetVersion.version.desc())
            .first()
        )
        db.add(dataset)
        db.add(
            DatasetVersion(
                dataset_id=dataset.id,
                version=(latest_version.version + 1) if latest_version else 1,
                storage_path=dataset.storage_path,
                source_version=dataset.source_version,
                observed_at=dataset.observed_at,
                valid_from=dataset.valid_from,
                valid_to=dataset.valid_to,
                checksum_sha256=dataset.checksum_sha256,
                ingest_key=(
                    f"{dataset.ingest_key}:temporal:{(latest_version.version + 1) if latest_version else 1}"
                    if dataset.ingest_key
                    else None
                ),
                source_metadata=dataset.source_metadata,
                provenance_metadata=dataset.provenance_metadata,
                source_registry_id=dataset.source_registry_id,
                source_snapshot_id=dataset.source_snapshot_id,
                data_contract_key=dataset.data_contract_key,
                data_contract_version=dataset.data_contract_version,
                validation_status="not_validated",
                validation_report_json=None,
                provenance_status="incomplete",
                lineage_status=dataset.lineage_status,
            )
        )
        db.commit()
        db.refresh(dataset)
        return DatasetService._to_response(dataset)

    @staticmethod
    def list_versions(db: Session, dataset_id: UUID) -> list[DatasetVersionRead]:
        DatasetService._get_dataset(db, dataset_id)
        rows = (
            db.query(DatasetVersion)
            .filter(DatasetVersion.dataset_id == dataset_id)
            .order_by(DatasetVersion.version.desc())
            .all()
        )
        return [DatasetVersionRead.model_validate(row) for row in rows]

    @staticmethod
    def get_dataset(db: Session, dataset_id: UUID) -> Dataset:
        dataset = db.get(Dataset, dataset_id)
        if not dataset:
            raise AppError(code="DATASET_NOT_FOUND", message="Dataset not found", status_code=404)
        return dataset

    @staticmethod
    def _get_dataset(db: Session, dataset_id: UUID) -> Dataset:
        return DatasetService.get_dataset(db, dataset_id)

    @staticmethod
    def get_dataset_geojson(db: Session, dataset_id: UUID) -> dict:
        dataset = DatasetService._get_dataset(db, dataset_id)
        if not DatasetService._is_vector_type(dataset.dataset_type):
            raise AppError(code="INVALID_DATASET_TYPE", message="Dataset is not a vector dataset", status_code=400)
        if not dataset.storage_path:
            raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404)
        if not pathlib.Path(dataset.storage_path).exists():
            raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404)

        raw = load_dataset_text(dataset.storage_path)
        # Only the parse can be "not valid JSON". Everything after it fails for
        # its own reasons and must say so, or an operator is sent to inspect a
        # file that parses perfectly well.
        try:
            payload = json.loads(raw)
        except Exception as exc:
            raise AppError(
                code="INVALID_GEOJSON",
                message="Stored dataset is not valid JSON",
                status_code=500,
            ) from exc

        try:
            metadata_value = getattr(dataset, "metadata_json", None)
            metadata = metadata_value if isinstance(metadata_value, dict) else {}
            provenance_value = getattr(dataset, "provenance_metadata", None)
            provenance = provenance_value if isinstance(provenance_value, dict) else {}
            canonical_evidence = provenance.get("canonical_consumption_artifact")
            canonical_checksum = (
                canonical_evidence.get("checksum_sha256")
                if isinstance(canonical_evidence, dict)
                else metadata.get("canonical_artifact_checksum_sha256")
            )
            # Post-normalization imports persist canonical bytes.  Reapplying
            # their original source CRS here would transform those coordinates
            # a second time.  Historical rows without this immutable binding
            # retain the legacy read-time canonicalization behavior until they
            # are re-ingested through the governed path.
            source_crs = (
                DatasetService.CANONICAL_VECTOR_CRS
                if canonical_checksum == getattr(dataset, "checksum_sha256", None)
                else (
                    metadata.get("source_crs")
                    or getattr(dataset, "crs", None)
                    or DatasetService.CANONICAL_VECTOR_CRS
                )
            )
            return VectorFeatureService.canonicalize_geojson_payload(payload, source_crs=str(source_crs))
        except AppError:
            # The canonicaliser's diagnosis is more precise than anything this
            # layer could substitute for it.
            raise
        except Exception as exc:
            raise AppError(
                code="DATASET_GEOJSON_UNREADABLE",
                message="The stored dataset could not be read as canonical GeoJSON",
                details={"dataset_id": str(dataset.id), "error_type": type(exc).__name__},
                status_code=500,
            ) from exc

    @staticmethod
    def inspect_vector_dataset(db: Session, dataset_id: UUID) -> dict[str, Any]:
        dataset = DatasetService._get_dataset(db, dataset_id)
        if not DatasetService._is_vector_type(dataset.dataset_type):
            raise AppError(code="INVALID_DATASET_TYPE", message="Dataset is not a vector dataset", status_code=400)
        if not dataset.storage_path or not Path(dataset.storage_path).exists():
            raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404)
        metadata = dataset.metadata_json or {}
        if not isinstance(metadata, dict):
            metadata = {}
        summary = DatasetService._extract_vector_summary(dataset.dataset_type, metadata)
        return {
            "dataset": {
                "id": str(dataset.id),
                "name": dataset.name,
                "dataset_type": dataset.dataset_type,
                "status": dataset.status,
                "source": dataset.source,
                "storage": DatasetStorageResponse(
                    original_filename=dataset.original_filename,
                    stored_filename=dataset.stored_filename,
                    content_type=dataset.content_type,
                    size_bytes=dataset.size_bytes,
                    checksum_sha256=dataset.checksum_sha256,
                ).model_dump(),
                "feature_count": metadata.get("feature_count"),
                "crs": metadata.get("crs"),
            },
            "summary": summary.model_dump() if summary else None,
            "metadata": metadata,
        }

    @staticmethod
    def vector_summary(db: Session, dataset_id: UUID) -> dict[str, Any]:
        dataset = DatasetService._get_dataset(db, dataset_id)
        if not DatasetService._is_vector_type(dataset.dataset_type):
            raise AppError(code="INVALID_DATASET_TYPE", message="Dataset is not a vector dataset", status_code=400)

        metadata = dataset.metadata_json or {}
        if not isinstance(metadata, dict):
            metadata = {}
        summary = DatasetService._extract_vector_summary(dataset.dataset_type, metadata)
        if not summary:
            raise AppError(code="INVALID_GEOJSON", message="Vector summary unavailable", status_code=422)
        return summary.model_dump()

    @staticmethod
    def raster_metadata(db: Session, dataset_id: UUID) -> dict[str, Any]:
        dataset = DatasetService._get_dataset(db, dataset_id)
        if dataset.quarantine_status == "quarantined" or dataset.status == "quarantined":
            raise AppError(
                code="DATASET_QUARANTINED",
                message="Quarantined datasets cannot be read as production-ready raster metadata.",
                status_code=409,
            )
        if not DatasetService._is_raster_type(dataset.dataset_type):
            raise AppError(code="INVALID_DATASET_TYPE", message="Dataset is not a raster dataset", status_code=400)
        if not dataset.storage_path:
            raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404)
        if not Path(dataset.storage_path).exists():
            raise AppError(code="DATASET_FILE_MISSING", message="Stored file missing", status_code=404)

        if isinstance(dataset.metadata_json, dict) and dataset.metadata_json.get("driver"):
            return dataset.metadata_json

        metadata = extract_raster_metadata(dataset.storage_path)
        dataset.metadata_json = dict(dataset.metadata_json or {})
        dataset.metadata_json.update(metadata)
        db.add(dataset)
        db.commit()
        db.refresh(dataset)
        return metadata