geointel/scripts/manage_alz_agriculture_release.py

#!/usr/bin/env python3
"""Plan, stage, review and explicitly apply one definitive ALZ edition.

Planning reads only the canonical source-catalog report. Staging downloads and
prepares one official v3 archive without database persistence. Review records
a named approval. Apply requires the exact plan and review hashes and reuses
the canonical agricultural DatasetService upload path.
"""

from __future__ import annotations

import argparse
from datetime import datetime, timezone
from hashlib import sha256
import json
import os
from pathlib import Path
import re
import subprocess
import sys
from typing import Any
from urllib.error import HTTPError, URLError
from urllib.parse import urlparse
from urllib.request import Request, urlopen

from geographic_scopes import GEOGRAPHIC_SCOPES
from provision_agricultural_parcel_history import (
    AgriculturalReleaseConfig,
    MAX_ARCHIVE_BYTES,
    OUTPUT_CRS,
    SOURCE_CRS,
    STABLE_REQUIRED_FIELDS,
    artifact_paths,
    resolve_release_config,
    reusable_artifact,
    sha256_file,
)


DEFAULT_API_URL = "http://127.0.0.1:8000/api/v1"
DEFAULT_SCOPE = "kempen-transport-region"
DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-evidence/agricultural-use-parcels")
DEFAULT_EVIDENCE_ROOT = Path("/app/storage/operator-evidence/alz-agriculture-refresh")
SOURCE_NAME = "agentschap_landbouw_zeevisserij_agricultural_parcels"
REMOTE_VERSION_PATTERN = re.compile(r"^(20[0-9]{2})-v3$")
LOCAL_VERSION_PATTERN = re.compile(r"^(20[0-9]{2})-definitive$")
SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
PROVISIONAL_VERSION_PATTERN = re.compile(r"\b(20[0-9]{2}-v[12])\b")
ACTIONABLE_STATUSES = {"update_available", "not_loaded"}


def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="Governed definitive ALZ release: plan, stage, review, then checksum-confirmed apply."
    )
    parser.add_argument("action", choices=("plan", "stage", "review", "apply"))
    parser.add_argument("--project-id", required=True, help="GeoIntel project UUID for the governed scope")
    parser.add_argument("--scope", choices=(DEFAULT_SCOPE,), default=DEFAULT_SCOPE)
    parser.add_argument("--api-url", default=os.environ.get("GEOINTEL_API_URL", DEFAULT_API_URL))
    parser.add_argument("--confirm-edition", help="Exact official definitive edition, for example 2026-v3")
    parser.add_argument("--confirm-plan-sha256", help="Exact staged-plan hash required for review/apply")
    parser.add_argument("--confirm-review-sha256", help="Exact review-evidence hash required for apply")
    parser.add_argument("--approve", action="store_true", help="Explicitly approve the staged evidence during review")
    parser.add_argument("--reviewer", help="Named human/operator approving the staged evidence")
    parser.add_argument("--review-note", default="", help="Optional bounded review note")
    parser.add_argument("--plan-path", type=Path, help="Override the governed staged-plan path")
    parser.add_argument("--review-path", type=Path, help="Override the governed review-evidence path")
    parser.add_argument(
        "--output-root",
        type=Path,
        default=Path(os.environ.get("GEOINTEL_AGRICULTURE_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)),
    )
    parser.add_argument(
        "--evidence-root",
        type=Path,
        default=Path(os.environ.get("GEOINTEL_ALZ_REFRESH_EVIDENCE_ROOT", DEFAULT_EVIDENCE_ROOT)),
    )
    parser.add_argument("--refresh-catalog", action="store_true", help="Bypass the short official-catalog cache")
    parser.add_argument("--request-timeout", type=int, default=900)
    parser.add_argument("--api-timeout", type=int, default=180)
    parser.add_argument("--import-timeout", type=int, default=3600)
    parser.add_argument("--max-features", type=int, default=250_000)
    parser.add_argument("--max-archive-mb", type=int, default=250)
    return parser.parse_args()


def api_data(api_url: str, path: str, timeout: int) -> dict[str, Any]:
    endpoint = f"{api_url.rstrip('/')}/{path.lstrip('/')}"
    request = Request(
        endpoint,
        headers={"Accept": "application/json", "User-Agent": "GeoIntel-ALZ-release/1.0"},
    )
    try:
        with urlopen(request, timeout=timeout) as response:
            payload = json.load(response)
    except HTTPError as exc:
        body = exc.read().decode("utf-8", errors="replace")
        raise RuntimeError(f"GeoIntel API returned HTTP {exc.code}: {body[-1000:]}") from exc
    except URLError as exc:
        raise RuntimeError(f"GeoIntel API is unreachable: {exc.reason}") from exc
    if not isinstance(payload, dict) or not isinstance(payload.get("data"), dict):
        raise RuntimeError("GeoIntel API response is not a canonical data envelope")
    return payload["data"]


def validate_project_scope(args: argparse.Namespace) -> None:
    project = api_data(args.api_url, f"projects/{args.project_id}", args.api_timeout)
    expected_name = GEOGRAPHIC_SCOPES[args.scope].project_name
    if project.get("name") != expected_name:
        raise RuntimeError(
            f"Project {args.project_id} is '{project.get('name')}', but scope {args.scope} requires '{expected_name}'"
        )


def parse_catalog_datetime(value: Any) -> datetime:
    if not isinstance(value, str) or not value.strip():
        raise RuntimeError("The official ALZ catalog did not provide a publication date")
    try:
        parsed = datetime.fromisoformat(value.strip().replace("Z", "+00:00"))
    except ValueError as exc:
        raise RuntimeError("The official ALZ catalog publication date is invalid") from exc
    if parsed.tzinfo is None:
        parsed = parsed.replace(tzinfo=timezone.utc)
    return parsed.astimezone(timezone.utc)


def release_from_catalog_item(item: dict[str, Any]) -> AgriculturalReleaseConfig:
    match = REMOTE_VERSION_PATTERN.fullmatch(str(item.get("remote_version") or ""))
    if not match:
        raise RuntimeError("The official ALZ catalog did not provide one definitive YYYY-v3 edition")
    year = int(match.group(1))
    published_at = parse_catalog_datetime(item.get("remote_published_at"))
    archive_url = (
        "https://www.landbouwvlaanderen.be/bestanden/gis/"
        f"agpa_{year}_{published_at.date().isoformat()}_public.zip"
    )
    return resolve_release_config(year, archive_url=archive_url)


def fetch_release_decision_from_item(args: argparse.Namespace, item: dict[str, Any]) -> dict[str, Any]:
    allowed_degraded = item.get("error_code") == "CATALOG_ALZ_CURRENT_SNAPSHOT_MISSING"
    if (
        item.get("source_name") != SOURCE_NAME
        or item.get("reachable") is not True
        or item.get("status") not in {"available", "degraded"}
        or (item.get("status") == "degraded" and not allowed_degraded)
        or "definitive_archive" not in set(item.get("matched_layers") or [])
        or not SHA256_PATTERN.fullmatch(str(item.get("capabilities_sha256") or ""))
        or item.get("metadata_identifier") != "alz-agricultural-use-parcels"
    ):
        raise RuntimeError("The official ALZ catalog is not safely available for release planning")
    release = release_from_catalog_item(item)
    local_version = str(item.get("local_source_version") or "")
    local_match = LOCAL_VERSION_PATTERN.fullmatch(local_version)
    if local_version and not local_match:
        status = "blocked_local_version"
    elif not local_version:
        status = "not_loaded"
    elif int(local_match.group(1)) == release.year:
        status = "current"
    elif int(local_match.group(1)) < release.year:
        status = "update_available"
    else:
        status = "blocked_remote_older"
    message = str(item.get("message") or "")
    provisional_match = PROVISIONAL_VERSION_PATTERN.search(message)
    return {
        "schema_version": 1,
        "status": status,
        "project_id": args.project_id,
        "scope": args.scope,
        "generated_at": datetime.now(timezone.utc).isoformat(),
        "catalog_checked_at": item.get("checked_at"),
        "local_source_version": local_version or None,
        "release": {
            "year": release.year,
            "edition": release.definitive_version,
            "archive_url": release.archive_url,
        },
        "catalog_identity": {
            "metadata_identifier": item["metadata_identifier"],
            "metadata_url": item.get("metadata_url"),
            "remote_version": release.definitive_version,
            "remote_published_at": item.get("remote_published_at"),
            "capabilities_sha256": item["capabilities_sha256"],
        },
        "provisional_release": provisional_match.group(1) if provisional_match else None,
        "provisional_release_importable": False,
        "automatic_download": False,
        "automatic_import": False,
        "destructive_replacement": False,
        "next_action": "stage" if status in ACTIONABLE_STATUSES else None,
    }


def fetch_release_decision(args: argparse.Namespace, *, refresh: bool) -> dict[str, Any]:
    query = "true" if refresh else "false"
    report = api_data(
        args.api_url,
        f"projects/{args.project_id}/datasets/source-catalog-probes?refresh={query}",
        args.api_timeout,
    )
    matches = [item for item in report.get("items") or [] if item.get("source_name") == SOURCE_NAME]
    if len(matches) != 1:
        raise RuntimeError("Source catalog report did not contain exactly one ALZ agriculture contract")
    return fetch_release_decision_from_item(args, matches[0])


def require_release_confirmation(args: argparse.Namespace, release: AgriculturalReleaseConfig) -> None:
    if args.confirm_edition != release.definitive_version:
        raise RuntimeError(f"Explicit --confirm-edition {release.definitive_version} is required")


def internal_base_url(api_url: str) -> str:
    value = api_url.rstrip("/")
    if value.endswith("/api/v1"):
        value = value[:-7]
    parsed = urlparse(value)
    if parsed.hostname not in {"127.0.0.1", "localhost", "::1"}:
        raise RuntimeError("Stage/apply must run inside GeoIntel against the local backend API")
    return value


def default_plan_path(args: argparse.Namespace, year: int) -> Path:
    return args.evidence_root / args.scope / str(year) / "staged-plan.json"


def default_review_path(args: argparse.Namespace, year: int) -> Path:
    return args.evidence_root / args.scope / str(year) / "review-evidence.json"


def governed_evidence_path(args: argparse.Namespace, path: Path) -> Path:
    if not path.resolve().is_relative_to(args.evidence_root.resolve()):
        raise RuntimeError(f"Release evidence path is outside the governed evidence root: {path}")
    return path


def build_operator_command(
    args: argparse.Namespace,
    release: AgriculturalReleaseConfig,
    *,
    fetch_only: bool,
) -> list[str]:
    command = [
        sys.executable,
        str(Path(__file__).resolve().parent / "provision_agricultural_parcel_history.py"),
        "--base-url",
        internal_base_url(args.api_url),
        "--scope",
        args.scope,
        "--years",
        str(release.year),
        "--archive-url",
        release.archive_url,
        "--output-root",
        str(args.output_root),
        "--request-timeout",
        str(args.request_timeout),
        "--import-timeout",
        str(args.import_timeout),
        "--max-features",
        str(args.max_features),
        "--max-archive-mb",
        str(args.max_archive_mb),
    ]
    if fetch_only:
        command.extend(("--force", "--fetch-only"))
    return command


def run_operator(command: list[str], *, action: str) -> dict[str, Any]:
    print(f"ALZ agriculture: {action}...", file=sys.stderr, flush=True)
    completed = subprocess.run(command, check=False, capture_output=True, text=True, encoding="utf-8")
    if completed.returncode != 0:
        detail = completed.stderr.strip() or completed.stdout.strip() or "operator returned no diagnostics"
        raise RuntimeError(f"ALZ agriculture {action} failed with exit {completed.returncode}: {detail[-3000:]}")
    try:
        payload = json.loads(completed.stdout)
    except json.JSONDecodeError as exc:
        raise RuntimeError(f"ALZ agriculture {action} returned invalid JSON") from exc
    if payload.get("status") != "ok":
        raise RuntimeError(f"ALZ agriculture {action} did not report success")
    return payload


def canonical_sha256(payload: dict[str, Any], hash_field: str) -> str:
    content = {key: value for key, value in payload.items() if key != hash_field}
    encoded = json.dumps(content, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
    return sha256(encoded).hexdigest()


def write_json(path: Path, payload: dict[str, Any]) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_suffix(path.suffix + ".partial")
    temporary.write_text(json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True) + "\n", encoding="utf-8")
    temporary.replace(path)


def _previous_manifest(output_root: Path, scope_key: str, year: int) -> tuple[Path, dict[str, Any]] | None:
    scope_root = output_root / scope_key
    if not scope_root.is_dir():
        return None
    previous_years = sorted(
        (int(path.name) for path in scope_root.iterdir() if path.is_dir() and path.name.isdigit() and int(path.name) < year),
        reverse=True,
    )
    for previous_year in previous_years:
        manifests = list((scope_root / str(previous_year)).glob("*.manifest.json"))
        if len(manifests) != 1:
            continue
        payload = json.loads(manifests[0].read_text(encoding="utf-8"))
        if payload.get("year") == previous_year and payload.get("scope_key") == scope_key:
            return manifests[0], payload
    return None


def _relative_change(current: float, previous: float) -> float | None:
    if previous <= 0:
        return None
    return round((current - previous) / previous, 8)


def validate_staged_release(args: argparse.Namespace, release: AgriculturalReleaseConfig) -> dict[str, Any]:
    paths = artifact_paths(args.output_root, args.scope, release.year, archive_url=release.archive_url)
    resolved_output = args.output_root.resolve()
    for key in ("archive", "artifact", "codelist", "manifest"):
        if not paths[key].resolve().is_relative_to(resolved_output):
            raise RuntimeError(f"Staged ALZ {key} is outside the governed output root: {paths[key]}")
    manifest = reusable_artifact(
        paths,
        year=release.year,
        scope_key=args.scope,
        archive_url=release.archive_url,
    )
    if manifest is None:
        raise RuntimeError("Staged ALZ artifacts are incomplete or no longer match their checksums")
    fields = set(manifest.get("source_fields") or [])
    if (
        manifest.get("source_url") != release.archive_url
        or manifest.get("source_crs") != SOURCE_CRS
        or manifest.get("output_crs") != OUTPUT_CRS
        or not STABLE_REQUIRED_FIELDS.issubset(fields)
        or int(manifest.get("feature_count") or 0) < 1
        or float(manifest.get("clipped_area_ha") or 0) <= 0
        or int(manifest.get("source_feature_count") or 0) < int(manifest.get("feature_count") or 0)
    ):
        raise RuntimeError("Staged ALZ manifest does not satisfy the definitive release contract")
    max_archive_bytes = min(args.max_archive_mb * 1024 * 1024, MAX_ARCHIVE_BYTES)
    if paths["archive"].stat().st_size > max_archive_bytes:
        raise RuntimeError("Staged ALZ archive exceeds the configured release limit")
    codelist = json.loads(paths["codelist"].read_text(encoding="utf-8"))
    crop_entries = codelist.get("crop_entries") if isinstance(codelist, dict) else None
    if codelist.get("year") != release.year or not isinstance(crop_entries, list) or not crop_entries:
        raise RuntimeError("Staged ALZ crop-code evidence is incomplete")
    previous_result = _previous_manifest(args.output_root, args.scope, release.year)
    baseline = None
    if previous_result is not None:
        previous_path, previous = previous_result
        if not previous_path.resolve().is_relative_to(resolved_output):
            raise RuntimeError("Previous ALZ baseline manifest is outside the governed output root")
        current_count = int(manifest["feature_count"])
        previous_count = int(previous.get("feature_count") or 0)
        current_area = float(manifest["clipped_area_ha"])
        previous_area = float(previous.get("clipped_area_ha") or 0)
        baseline = {
            "year": int(previous["year"]),
            "manifest_path": str(previous_path),
            "manifest_sha256": sha256_file(previous_path),
            "feature_count": previous_count,
            "clipped_area_ha": previous_area,
            "feature_count_change_ratio": _relative_change(current_count, previous_count),
            "clipped_area_change_ratio": _relative_change(current_area, previous_area),
        }
    return {
        "manifest_path": str(paths["manifest"]),
        "manifest_sha256": sha256_file(paths["manifest"]),
        "archive_path": str(paths["archive"]),
        "archive_sha256": manifest["source_archive_sha256"],
        "archive_size_bytes": manifest["source_archive_size_bytes"],
        "artifact_path": str(paths["artifact"]),
        "artifact_sha256": manifest["artifact_sha256"],
        "crop_code_list_path": str(paths["codelist"]),
        "crop_code_list_sha256": manifest["crop_code_list_sha256"],
        "crop_entry_count": len(crop_entries),
        "crop_code_conflicts": codelist.get("code_title_conflicts") or {},
        "source_feature_count": manifest["source_feature_count"],
        "feature_count": manifest["feature_count"],
        "clipped_feature_count": manifest["clipped_feature_count"],
        "clipped_area_ha": manifest["clipped_area_ha"],
        "source_fields": sorted(fields),
        "member_nis_codes": manifest.get("member_nis_codes") or [],
        "baseline": baseline,
    }


def build_staged_plan(
    args: argparse.Namespace,
    decision: dict[str, Any],
    release: AgriculturalReleaseConfig,
    operator_result: dict[str, Any],
) -> dict[str, Any]:
    payload: dict[str, Any] = {
        "schema_version": 1,
        "status": "staged",
        "created_at": datetime.now(timezone.utc).isoformat(),
        "project_id": args.project_id,
        "scope": args.scope,
        "local_source_version_before_apply": decision.get("local_source_version"),
        "release": decision["release"],
        "catalog_identity": decision["catalog_identity"],
        "evidence": validate_staged_release(args, release),
        "operator_result": operator_result,
        "review_required": True,
        "apply_requires_plan_sha256": True,
        "apply_requires_review_sha256": True,
        "automatic_import": False,
        "destructive_replacement": False,
        "existing_snapshots_retained": True,
    }
    payload["plan_sha256"] = canonical_sha256(payload, "plan_sha256")
    return payload


def _release_from_payload(payload: dict[str, Any]) -> AgriculturalReleaseConfig:
    release = payload.get("release") or {}
    year = release.get("year") if isinstance(release, dict) else None
    edition = release.get("edition") if isinstance(release, dict) else None
    archive_url = release.get("archive_url") if isinstance(release, dict) else None
    if (
        not isinstance(year, int)
        or isinstance(year, bool)
        or edition != f"{year}-v3"
        or not isinstance(archive_url, str)
    ):
        raise RuntimeError("Release evidence is missing required ALZ identity fields")
    return resolve_release_config(year, archive_url=archive_url)


def load_staged_plan(
    args: argparse.Namespace,
    release: AgriculturalReleaseConfig,
) -> tuple[Path, dict[str, Any]]:
    path = governed_evidence_path(args, args.plan_path or default_plan_path(args, release.year))
    if not path.is_file():
        raise RuntimeError(f"Staged ALZ plan is missing: {path}")
    payload = json.loads(path.read_text(encoding="utf-8"))
    actual_sha = canonical_sha256(payload, "plan_sha256")
    if payload.get("plan_sha256") != actual_sha:
        raise RuntimeError("Staged ALZ plan checksum is invalid")
    if args.confirm_plan_sha256 != actual_sha:
        raise RuntimeError(f"Explicit --confirm-plan-sha256 {actual_sha} is required")
    if (
        payload.get("status") != "staged"
        or payload.get("project_id") != args.project_id
        or payload.get("scope") != args.scope
        or _release_from_payload(payload) != release
    ):
        raise RuntimeError("Staged ALZ plan identity is invalid")
    current_evidence = validate_staged_release(args, release)
    if current_evidence != payload.get("evidence"):
        raise RuntimeError("Staged ALZ artifacts no longer match the approved plan")
    return path, payload


def require_catalog_unchanged(plan: dict[str, Any], decision: dict[str, Any]) -> None:
    if plan.get("release") != decision.get("release") or plan.get("catalog_identity") != decision.get("catalog_identity"):
        raise RuntimeError("The official ALZ release evidence changed; create and review a new staged plan")


def build_review_evidence(args: argparse.Namespace, plan_path: Path, plan: dict[str, Any]) -> dict[str, Any]:
    reviewer = str(args.reviewer or "").strip()
    note = str(args.review_note or "").strip()
    if not args.approve or len(reviewer) < 2 or len(reviewer) > 120:
        raise RuntimeError("Review requires --approve and a named --reviewer between 2 and 120 characters")
    if len(note) > 1000:
        raise RuntimeError("Review note must not exceed 1000 characters")
    payload: dict[str, Any] = {
        "schema_version": 1,
        "status": "approved",
        "reviewed_at": datetime.now(timezone.utc).isoformat(),
        "reviewer": reviewer,
        "review_note": note or None,
        "project_id": args.project_id,
        "scope": args.scope,
        "release": plan["release"],
        "staged_plan_path": str(plan_path),
        "staged_plan_sha256": plan["plan_sha256"],
        "reviewed_checks": [
            "official_catalog_identity",
            "source_archive_checksum",
            "geopackage_schema_and_crs",
            "scope_feature_and_area_accounting",
            "crop_code_list_and_conflicts",
            "previous_definitive_edition_delta",
            "provisional_snapshot_exclusion",
            "immutable_dataset_apply",
        ],
    }
    payload["review_sha256"] = canonical_sha256(payload, "review_sha256")
    return payload


def load_review_evidence(
    args: argparse.Namespace,
    release: AgriculturalReleaseConfig,
    plan: dict[str, Any],
) -> tuple[Path, dict[str, Any]]:
    path = governed_evidence_path(args, args.review_path or default_review_path(args, release.year))
    if not path.is_file():
        raise RuntimeError(f"Approved ALZ review evidence is missing: {path}")
    payload = json.loads(path.read_text(encoding="utf-8"))
    actual_sha = canonical_sha256(payload, "review_sha256")
    if payload.get("review_sha256") != actual_sha:
        raise RuntimeError("ALZ review evidence checksum is invalid")
    if args.confirm_review_sha256 != actual_sha:
        raise RuntimeError(f"Explicit --confirm-review-sha256 {actual_sha} is required")
    if (
        payload.get("status") != "approved"
        or payload.get("project_id") != args.project_id
        or payload.get("scope") != args.scope
        or payload.get("release") != plan.get("release")
        or payload.get("staged_plan_sha256") != plan.get("plan_sha256")
        or not str(payload.get("reviewer") or "").strip()
    ):
        raise RuntimeError("ALZ review evidence does not authorize this staged plan")
    return path, payload


def _release_result(operator_result: dict[str, Any], year: int) -> dict[str, Any]:
    matches = [item for item in operator_result.get("years") or [] if int(item.get("year") or 0) == year]
    if len(matches) != 1:
        raise RuntimeError("Agriculture operator did not return exactly one result for the approved edition")
    return matches[0]


def main() -> int:
    args = parse_args()
    try:
        if min(args.request_timeout, args.api_timeout, args.import_timeout, args.max_features, args.max_archive_mb) <= 0:
            raise ValueError("All timeout and safety limits must be positive")
        validate_project_scope(args)
        refresh = args.refresh_catalog or args.action in {"stage", "review", "apply"}
        decision = fetch_release_decision(args, refresh=refresh)
        release = _release_from_payload(decision)

        if args.action == "plan":
            print(json.dumps({"status": "ok", "action": "plan", "decision": decision}, ensure_ascii=False, indent=2))
            return 0

        require_release_confirmation(args, release)
        if args.action == "stage":
            if decision.get("status") not in ACTIONABLE_STATUSES:
                raise RuntimeError(f"ALZ release is not safely stageable: {decision.get('status')}")
            operator_result = run_operator(build_operator_command(args, release, fetch_only=True), action="staging")
            staged_result = _release_result(operator_result, release.year)
            if staged_result.get("status") != "prepared":
                raise RuntimeError("ALZ staging did not produce prepared release evidence")
            plan = build_staged_plan(args, decision, release, operator_result)
            plan_path = governed_evidence_path(args, args.plan_path or default_plan_path(args, release.year))
            write_json(plan_path, plan)
            print(json.dumps({"status": "staged", "plan_path": str(plan_path), **plan}, ensure_ascii=False, indent=2))
            return 0

        plan_path, plan = load_staged_plan(args, release)
        require_catalog_unchanged(plan, decision)
        if args.action == "review":
            review = build_review_evidence(args, plan_path, plan)
            review_path = governed_evidence_path(args, args.review_path or default_review_path(args, release.year))
            write_json(review_path, review)
            print(json.dumps({"status": "approved", "review_path": str(review_path), **review}, ensure_ascii=False, indent=2))
            return 0

        review_path, review = load_review_evidence(args, release, plan)
        operator_result = run_operator(build_operator_command(args, release, fetch_only=False), action="applying")
        applied_result = _release_result(operator_result, release.year)
        if applied_result.get("status") not in {"imported", "existing"} or not applied_result.get("dataset_id"):
            raise RuntimeError("ALZ apply did not return one persisted immutable Dataset")
        final_decision = fetch_release_decision(args, refresh=True)
        if (
            final_decision.get("status") != "current"
            or final_decision.get("local_source_version") != f"{release.year}-definitive"
        ):
            raise RuntimeError("Applied ALZ Dataset did not become the current local definitive edition")
        evidence = {
            "schema_version": 1,
            "status": "applied",
            "applied_at": datetime.now(timezone.utc).isoformat(),
            "project_id": args.project_id,
            "scope": args.scope,
            "release": plan["release"],
            "staged_plan_path": str(plan_path),
            "staged_plan_sha256": plan["plan_sha256"],
            "review_path": str(review_path),
            "review_sha256": review["review_sha256"],
            "reviewer": review["reviewer"],
            "dataset_id": applied_result["dataset_id"],
            "dataset_status": applied_result["status"],
            "feature_count": applied_result.get("feature_count"),
            "final_catalog_decision": final_decision,
            "existing_snapshots_retained": True,
        }
        evidence["applied_evidence_sha256"] = canonical_sha256(evidence, "applied_evidence_sha256")
        evidence_path = plan_path.with_name("applied-evidence.json")
        write_json(evidence_path, evidence)
        print(json.dumps({"status": "applied", "evidence_path": str(evidence_path), **evidence}, ensure_ascii=False, indent=2))
        return 0
    except (OSError, RuntimeError, ValueError, KeyError, json.JSONDecodeError) as exc:
        print(
            json.dumps({"status": "error", "action": args.action, "message": str(exc)}, ensure_ascii=False),
            file=sys.stderr,
        )
        return 1


if __name__ == "__main__":
    raise SystemExit(main())