geointel/scripts/training_dataset_eligibility.py

"""Fail-closed provenance gate for Belgian building-training inputs.

The database data-contract validator decides whether a dataset can be stored.
This module is intentionally a second, independent gate at the point where
persisted datasets become irreversible training pairs.  It has no database
side effects, making the decision reproducible in a corpus manifest and easy
to re-check before CUDA training starts.

Legacy data may only pass through this module in explicit fixture mode.  That
mode is deliberately limited to records marked as fixtures and must never be
used for an operational corpus.
"""

from __future__ import annotations

import hashlib
import json
import re
from collections.abc import Mapping
from pathlib import Path
from typing import Any, Callable, Literal
from uuid import UUID


TRAINING_ELIGIBILITY_POLICY_VERSION = "geointel-training-source-eligibility/v1"
DatasetRole = Literal["raster", "reference"]

_SHA256_RE = re.compile(r"^[0-9a-f]{64}$", re.IGNORECASE)
_ALLOWED_SOURCE_CLASSIFICATIONS = {
    "authoritative",
    "corroborative",
    "contextual",
    "derived",
}
_FIXTURE_SOURCES = {"fixture", "test", "test_fixture", "test-fixture"}


class TrainingEligibilityError(ValueError):
    """Raised when persisted inputs cannot be used in an operational corpus."""


class TrainingEligibilityResult:
    """Stable, JSON-serializable decision for one persisted dataset."""

    __slots__ = ("role", "eligible", "fixture_mode", "reasons", "evidence")

    def __init__(
        self,
        *,
        role: DatasetRole,
        eligible: bool,
        fixture_mode: bool,
        reasons: tuple[str, ...],
        evidence: dict[str, Any],
    ) -> None:
        self.role = role
        self.eligible = eligible
        self.fixture_mode = fixture_mode
        self.reasons = reasons
        self.evidence = evidence

    def as_dict(self) -> dict[str, Any]:
        return {
            "policy_version": TRAINING_ELIGIBILITY_POLICY_VERSION,
            "role": self.role,
            "eligible": self.eligible,
            "fixture_mode": self.fixture_mode,
            "reasons": list(self.reasons),
            "evidence": self.evidence,
        }


def _value(record: Any, field: str, default: Any = None) -> Any:
    if isinstance(record, Mapping):
        return record.get(field, default)
    return getattr(record, field, default)


def _normalise_status(value: Any) -> str:
    return str(value or "").strip().lower()


def _normalise_mapping(value: Any) -> dict[str, Any]:
    return dict(value) if isinstance(value, Mapping) else {}


def _source_identity(dataset: Any) -> tuple[str, str]:
    return (
        _normalise_status(_value(dataset, "source")),
        _normalise_status(_value(dataset, "source_name")),
    )


def _is_explicit_fixture(dataset: Any, registry: Any) -> bool:
    source, source_name = _source_identity(dataset)
    metadata = _normalise_mapping(_value(dataset, "metadata_json"))
    provenance = _normalise_mapping(_value(dataset, "provenance_metadata"))
    usage_policy = _normalise_mapping(_value(registry, "usage_policy_json"))
    return bool(
        source in _FIXTURE_SOURCES
        or source_name in _FIXTURE_SOURCES
        or metadata.get("fixture") is True
        or provenance.get("fixture") is True
        or usage_policy.get("fixture_only") is True
    )


def _dataset_evidence(dataset: Any, registry: Any, snapshot: Any) -> dict[str, Any]:
    usage_policy = _normalise_mapping(_value(registry, "usage_policy_json"))
    allowed_tasks = usage_policy.get("allowed_tasks")
    validation_authority = _normalise_mapping(usage_policy.get("validation_authority"))
    return {
        "dataset_id": str(_value(dataset, "id") or ""),
        "dataset_type": _normalise_status(_value(dataset, "dataset_type")),
        "dataset_role": _normalise_status(_value(dataset, "dataset_role")),
        "source": _normalise_status(_value(dataset, "source")),
        "source_name": _normalise_status(_value(dataset, "source_name")),
        "checksum_sha256": _value(dataset, "checksum_sha256"),
        "data_contract_key": _value(dataset, "data_contract_key"),
        "data_contract_version": _value(dataset, "data_contract_version"),
        "validation_status": _normalise_status(_value(dataset, "validation_status")),
        "provenance_status": _normalise_status(_value(dataset, "provenance_status")),
        "lineage_status": _normalise_status(_value(dataset, "lineage_status")),
        "quarantine_status": _normalise_status(_value(dataset, "quarantine_status")),
        "dataset_status": _normalise_status(_value(dataset, "status")),
        "dataset_source_registry_id": str(_value(dataset, "source_registry_id") or ""),
        "dataset_source_snapshot_id": str(_value(dataset, "source_snapshot_id") or ""),
        "source_registry_id": str(_value(registry, "id") or ""),
        "source_key": _normalise_status(_value(registry, "source_key")),
        "source_classification": _normalise_status(_value(registry, "classification")),
        "source_freshness_status": _normalise_status(_value(registry, "freshness_status")),
        "source_ingest_status": _normalise_status(_value(registry, "ingest_status")),
        "source_training_allowed": usage_policy.get("training_allowed"),
        "source_ground_truth_allowed": usage_policy.get("ground_truth_allowed"),
        "source_allowed_tasks": list(allowed_tasks) if isinstance(allowed_tasks, list) else [],
        "source_validation_authority": validation_authority,
        "source_building_validation_authority": validation_authority.get("building_validation"),
        "source_snapshot_id": str(_value(snapshot, "id") or _value(dataset, "source_snapshot_id") or ""),
        "snapshot_source_registry_id": str(_value(snapshot, "source_registry_id") or ""),
        "snapshot_key": _value(snapshot, "snapshot_key"),
        "snapshot_checksum_sha256": _value(snapshot, "checksum_sha256"),
        "snapshot_freshness_status": _normalise_status(_value(snapshot, "freshness_status")),
        "snapshot_ingest_status": _normalise_status(_value(snapshot, "ingest_status")),
    }


def evaluate_dataset_training_eligibility(
    dataset: Any,
    *,
    role: DatasetRole,
    fixture_mode: bool = False,
) -> TrainingEligibilityResult:
    """Evaluate a source Dataset without mutating it.

    Operational calls require server-attested source registry and snapshot
    relationships.  ``fixture_mode`` can relax only legacy provenance fields,
    and only for an explicitly marked fixture.  A failed validation or a
    quarantine is never relaxed.
    """

    registry = _value(dataset, "source_registry")
    snapshot = _value(dataset, "source_snapshot")
    evidence = _dataset_evidence(dataset, registry, snapshot)
    reasons: list[str] = []

    dataset_type = evidence["dataset_type"]
    dataset_role = evidence["dataset_role"]
    if role == "raster" and dataset_type != "raster":
        reasons.append("dataset_type_not_raster")
    if role == "reference":
        if dataset_type != "vector":
            reasons.append("dataset_type_not_vector")
        if dataset_role != "reference":
            reasons.append("dataset_role_not_reference")

    if evidence["dataset_status"] != "ready":
        reasons.append("dataset_not_ready")
    if evidence["quarantine_status"] == "quarantined":
        reasons.append("dataset_quarantined")

    explicit_fixture = _is_explicit_fixture(dataset, registry)
    if fixture_mode and not explicit_fixture:
        reasons.append("fixture_mode_requires_explicit_fixture")
    validation_status = evidence["validation_status"]
    if validation_status == "failed":
        reasons.append("validation_failed")
    elif validation_status != "passed" and not (fixture_mode and explicit_fixture):
        reasons.append("validation_not_passed")

    if not fixture_mode:
        if evidence["provenance_status"] != "complete":
            reasons.append("provenance_not_complete")
        if evidence["lineage_status"] not in {"complete", "not_applicable"}:
            reasons.append("lineage_not_complete")
        if not evidence["data_contract_key"] or not evidence["data_contract_version"]:
            reasons.append("data_contract_not_versioned")
        checksum = str(evidence["checksum_sha256"] or "")
        if not _SHA256_RE.fullmatch(checksum):
            reasons.append("dataset_checksum_invalid")
        if registry is None:
            reasons.append("source_registry_missing")
        if snapshot is None:
            reasons.append("source_snapshot_missing")
        if registry is not None:
            if (
                evidence["dataset_source_registry_id"]
                and evidence["dataset_source_registry_id"] != evidence["source_registry_id"]
            ):
                reasons.append("dataset_source_registry_binding_mismatch")
            classification = evidence["source_classification"]
            if classification not in _ALLOWED_SOURCE_CLASSIFICATIONS:
                reasons.append("source_classification_not_allowed")
            if evidence["source_training_allowed"] is not True:
                reasons.append("source_not_allowed_for_training")
            if evidence["source_ingest_status"] == "quarantined":
                reasons.append("source_registry_quarantined")
        if snapshot is not None:
            if (
                evidence["dataset_source_snapshot_id"]
                and evidence["dataset_source_snapshot_id"] != evidence["source_snapshot_id"]
            ):
                reasons.append("dataset_source_snapshot_binding_mismatch")
            if (
                registry is not None
                and evidence["snapshot_source_registry_id"]
                and evidence["snapshot_source_registry_id"] != evidence["source_registry_id"]
            ):
                reasons.append("source_snapshot_registry_mismatch")
            if evidence["snapshot_ingest_status"] != "ingested":
                reasons.append("source_snapshot_not_ingested")
            if evidence["snapshot_freshness_status"] not in {"current", "not_applicable"}:
                reasons.append("source_snapshot_freshness_not_approved")
            snapshot_checksum = str(evidence["snapshot_checksum_sha256"] or "")
            if not _SHA256_RE.fullmatch(snapshot_checksum):
                reasons.append("source_snapshot_checksum_invalid")
            elif snapshot_checksum.lower() != checksum.lower():
                reasons.append("source_snapshot_checksum_mismatch")
        if role == "reference":
            if evidence["source_classification"] != "authoritative":
                reasons.append("reference_source_not_authoritative")
            if evidence["source_ground_truth_allowed"] is not True:
                reasons.append("reference_source_not_ground_truth_allowed")
            if "building_validation" not in evidence["source_allowed_tasks"]:
                reasons.append("reference_source_not_approved_for_building_validation")
            if evidence["source_building_validation_authority"] != "primary":
                reasons.append("reference_building_validation_not_primary")

    return TrainingEligibilityResult(
        role=role,
        eligible=not reasons,
        fixture_mode=fixture_mode,
        reasons=tuple(sorted(set(reasons))),
        evidence=evidence,
    )


def assert_dataset_training_eligible(
    dataset: Any,
    *,
    role: DatasetRole,
    fixture_mode: bool = False,
    sample_slug: str | None = None,
) -> TrainingEligibilityResult:
    result = evaluate_dataset_training_eligibility(dataset, role=role, fixture_mode=fixture_mode)
    if result.eligible:
        return result
    label = f" for sample {sample_slug!r}" if sample_slug else ""
    raise TrainingEligibilityError(
        f"{role} dataset is not eligible for training{label}: {', '.join(result.reasons)}"
    )


def training_pair_evidence(
    *,
    raster: Any,
    reference: Any,
    fixture_mode: bool = False,
) -> dict[str, Any]:
    """Return manifest-ready evidence for one raster/reference pair."""

    raster_result = evaluate_dataset_training_eligibility(
        raster,
        role="raster",
        fixture_mode=fixture_mode,
    )
    reference_result = evaluate_dataset_training_eligibility(
        reference,
        role="reference",
        fixture_mode=fixture_mode,
    )
    return {
        "policy_version": TRAINING_ELIGIBILITY_POLICY_VERSION,
        "eligible": raster_result.eligible and reference_result.eligible,
        "fixture_mode": fixture_mode,
        "raster": raster_result.as_dict(),
        "reference": reference_result.as_dict(),
    }


def manifest_training_eligibility_failures(
    manifest: Mapping[str, Any],
    *,
    fixture_mode: bool = False,
) -> list[str]:
    """Re-check immutable eligibility evidence before an actual training run."""

    failures: list[str] = []
    eligibility = manifest.get("training_eligibility")
    if not isinstance(eligibility, Mapping):
        return ["manifest_training_eligibility_missing"]
    if eligibility.get("policy_version") != TRAINING_ELIGIBILITY_POLICY_VERSION:
        failures.append("manifest_training_eligibility_policy_invalid")
    if eligibility.get("status") != "eligible":
        failures.append("manifest_training_eligibility_not_eligible")
    manifest_fixture_mode = eligibility.get("fixture_mode") is True
    if manifest_fixture_mode != fixture_mode:
        failures.append("manifest_fixture_mode_mismatch")

    samples = manifest.get("samples")
    if not isinstance(samples, list) or not samples:
        failures.append("manifest_samples_missing")
        return failures
    for sample in samples:
        if not isinstance(sample, Mapping):
            failures.append("manifest_sample_invalid")
            continue
        slug = str(sample.get("sample_slug") or "<unknown>")
        pair = sample.get("training_eligibility")
        if not isinstance(pair, Mapping):
            failures.append(f"{slug}:training_eligibility_missing")
            continue
        if pair.get("policy_version") != TRAINING_ELIGIBILITY_POLICY_VERSION:
            failures.append(f"{slug}:training_eligibility_policy_invalid")
        if pair.get("fixture_mode") is not fixture_mode:
            failures.append(f"{slug}:fixture_mode_mismatch")
        if pair.get("eligible") is not True:
            failures.append(f"{slug}:training_pair_not_eligible")
        for role in ("raster", "reference"):
            decision = pair.get(role)
            if not isinstance(decision, Mapping):
                failures.append(f"{slug}:{role}_eligibility_missing")
                continue
            if decision.get("eligible") is not True:
                failures.append(f"{slug}:{role}_not_eligible")
            reasons = decision.get("reasons")
            if isinstance(reasons, list) and reasons:
                failures.append(f"{slug}:{role}_has_rejection_reasons")
    return sorted(set(failures))


def _default_live_dataset_access() -> tuple[Callable[[], Any], type[Any]]:
    """Load the database dependencies only for an operational live re-check.

    This module is also imported by pure filesystem tooling and fixture tests.
    Keeping the import lazy means those paths do not accidentally open a
    database connection, while a normal release/verify invocation still fails
    closed if the live registry cannot be checked.
    """

    import sys

    repo_root = Path(__file__).resolve().parents[1]
    app_root = repo_root / "backend"
    if str(app_root) not in sys.path:
        sys.path.insert(0, str(app_root))
    from app.db.session import SessionLocal
    from app.models import Dataset

    return SessionLocal, Dataset


def live_manifest_training_eligibility_failures(
    manifest: Mapping[str, Any],
    *,
    fixture_mode: bool = False,
    session_factory: Callable[[], Any] | None = None,
    dataset_model: type[Any] | None = None,
) -> list[str]:
    """Re-evaluate every frozen corpus parent against the live database.

    A frozen manifest proves what was eligible when it was created, not what
    remains eligible now.  Operational releases therefore resolve the exact
    raster/reference Dataset ids again immediately before seal, retry, resume
    and CUDA training.  A later quarantine, failed contract, snapshot change
    or missing record is a revocation and cannot be masked by the old manifest.

    Fixture-only corpora deliberately have no operational authority and are
    never used for a production release; their isolated tests may skip this
    live database boundary.
    """

    if fixture_mode:
        return []
    failures: list[str] = []
    samples = manifest.get("samples")
    if not isinstance(samples, list) or not samples:
        return ["live_manifest_samples_missing"]
    if session_factory is None or dataset_model is None:
        try:
            default_factory, default_model = _default_live_dataset_access()
        except Exception:
            return ["live_training_dataset_access_unavailable"]
        session_factory = session_factory or default_factory
        dataset_model = dataset_model or default_model

    try:
        db = session_factory()
    except Exception:
        return ["live_training_dataset_access_unavailable"]
    try:
        for sample in samples:
            if not isinstance(sample, Mapping):
                failures.append("live_manifest_sample_invalid")
                continue
            slug = str(sample.get("sample_slug") or "<unknown>")
            recorded_pair = sample.get("training_eligibility")
            for role, field_name in (("raster", "raster_dataset_id"), ("reference", "reference_dataset_id")):
                raw_id = sample.get(field_name)
                if not isinstance(raw_id, str) or not raw_id.strip():
                    failures.append(f"{slug}:{role}_dataset_id_missing_for_live_check")
                    continue
                try:
                    dataset_id = UUID(raw_id)
                except (TypeError, ValueError, AttributeError):
                    failures.append(f"{slug}:{role}_dataset_id_invalid_for_live_check")
                    continue
                try:
                    dataset = db.get(dataset_model, dataset_id)
                except Exception:
                    failures.append(f"{slug}:{role}_live_lookup_failed")
                    continue
                if dataset is None:
                    failures.append(f"{slug}:{role}_live_dataset_missing")
                    continue
                result = evaluate_dataset_training_eligibility(
                    dataset,
                    role=role,  # type: ignore[arg-type]
                    fixture_mode=False,
                )
                if not result.eligible:
                    for reason in result.reasons:
                        failures.append(f"{slug}:{role}_live_revoked:{reason}")
                if isinstance(recorded_pair, Mapping):
                    recorded_role = recorded_pair.get(role)
                    if isinstance(recorded_role, Mapping):
                        recorded_evidence = recorded_role.get("evidence")
                        if isinstance(recorded_evidence, Mapping) and str(recorded_evidence.get("dataset_id") or "") != raw_id:
                            failures.append(f"{slug}:{role}_manifest_dataset_binding_mismatch")
    finally:
        close = getattr(db, "close", None)
        if callable(close):
            close()
    return sorted(set(failures))


def assert_manifest_training_eligible(
    manifest: Mapping[str, Any],
    *,
    fixture_mode: bool = False,
) -> None:
    failures = manifest_training_eligibility_failures(manifest, fixture_mode=fixture_mode)
    if failures:
        raise TrainingEligibilityError(
            "Corpus manifest is not eligible for training: " + ", ".join(failures)
        )


def _file_sha256(path: Path) -> str:
    digest = hashlib.sha256()
    with path.open("rb") as stream:
        for chunk in iter(lambda: stream.read(1024 * 1024), b""):
            digest.update(chunk)
    return digest.hexdigest()


def frozen_manifest_training_eligibility_failures(
    manifest_path: Path,
    *,
    fixture_mode: bool = False,
    verify_live: bool = False,
    session_factory: Callable[[], Any] | None = None,
    dataset_model: type[Any] | None = None,
) -> list[str]:
    """Verify an immutable manifest and its source-eligibility decision together."""

    failures: list[str] = []
    try:
        manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return ["corpus_manifest_unreadable"]
    if not isinstance(manifest, Mapping):
        return ["corpus_manifest_invalid"]
    failures.extend(manifest_training_eligibility_failures(manifest, fixture_mode=fixture_mode))

    freeze_path = manifest_path.parent / "corpus-freeze.json"
    try:
        freeze = json.loads(freeze_path.read_text(encoding="utf-8"))
    except (OSError, json.JSONDecodeError):
        return sorted(set(failures + ["corpus_freeze_missing_or_invalid"]))
    if not isinstance(freeze, Mapping):
        return sorted(set(failures + ["corpus_freeze_missing_or_invalid"]))
    if freeze.get("immutable") is not True:
        failures.append("corpus_freeze_not_immutable")
    if freeze.get("manifest_sha256") != _file_sha256(manifest_path):
        failures.append("corpus_manifest_checksum_mismatch")
    if freeze.get("training_eligibility_policy") != TRAINING_ELIGIBILITY_POLICY_VERSION:
        failures.append("corpus_freeze_policy_invalid")
    if bool(freeze.get("fixture_mode")) != fixture_mode:
        failures.append("corpus_freeze_fixture_mode_mismatch")
    if verify_live:
        failures.extend(
            live_manifest_training_eligibility_failures(
                manifest,
                fixture_mode=fixture_mode,
                session_factory=session_factory,
                dataset_model=dataset_model,
            )
        )
    return sorted(set(failures))


def assert_frozen_manifest_training_eligible(
    manifest_path: Path,
    *,
    fixture_mode: bool = False,
    verify_live: bool = False,
    session_factory: Callable[[], Any] | None = None,
    dataset_model: type[Any] | None = None,
) -> dict[str, Any]:
    """Load only a frozen, policy-valid corpus manifest for a training entrypoint."""

    failures = frozen_manifest_training_eligibility_failures(
        manifest_path,
        fixture_mode=fixture_mode,
        verify_live=verify_live,
        session_factory=session_factory,
        dataset_model=dataset_model,
    )
    if failures:
        raise TrainingEligibilityError(
            "Corpus manifest is not eligible for training: " + ", ".join(failures)
        )
    payload = json.loads(manifest_path.read_text(encoding="utf-8"))
    assert isinstance(payload, dict)
    return payload