Add governed ALZ release promotion

scripts/README.md

@@ -1498,6 +1498,45 @@ Datasets. `--force` refreshes retained evidence but cannot silently replace a
 conflicting persisted annual checksum. Use `--scope mol` for an independent
 municipal series.
 
+### Governed future definitive ALZ release
+
+Run the four phases only inside the GeoIntel container. The project id must
+belong to `Kempen Regional Workbench`:
+
+```bash
+docker exec geointel python /app/scripts/manage_alz_agriculture_release.py plan \
+  --project-id <KEMPEN_PROJECT_ID> --refresh-catalog
+
+docker exec geointel python /app/scripts/manage_alz_agriculture_release.py stage \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --confirm-edition <YYYY-v3_FROM_PLAN>
+
+docker exec geointel python /app/scripts/manage_alz_agriculture_release.py review \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --confirm-edition <YYYY-v3_FROM_PLAN> \
+  --confirm-plan-sha256 <SHA256_FROM_STAGE> \
+  --approve --reviewer "<OPERATOR_NAME>" \
+  --review-note "Schema, gewascodes, scope en jaarverschillen nagekeken"
+
+docker exec geointel python /app/scripts/manage_alz_agriculture_release.py apply \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --confirm-edition <YYYY-v3_FROM_PLAN> \
+  --confirm-plan-sha256 <SHA256_FROM_STAGE> \
+  --confirm-review-sha256 <SHA256_FROM_REVIEW>
+```
+
+`plan` is read-only. `stage` derives the exact archive from the official
+catalog campaign/publication date, downloads within the 250 MiB ceiling and
+runs the existing provisioner with `--force --fetch-only`. The staged plan
+binds archive, GeoJSON, schema, CRS, crop-code list, scope counts and previous
+definitive-edition manifest hash/deltas. ZIP member count and extracted size
+are bounded before the GeoPackage is read. `review` imports nothing. `apply` revalidates the
+catalog and every byte before using the canonical Dataset upload route.
+
+Only a definitive `YYYY-v3` is eligible. Current or older editions, v1/v2
+snapshots, changed catalog/source evidence and paths outside the governed
+roots fail closed. Existing annual Datasets remain immutable and queryable.
+
 ## Buildings and Addresses Register snapshot
 
 Prepare and audit the current official Mol snapshot without persistence:

scripts/manage_alz_agriculture_release.py

@@ -0,0 +1,616 @@
+#!/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())

scripts/provision_agricultural_parcel_history.py

@@ -10,17 +10,19 @@ directly and never uses the provisional current-campaign snapshot.
 from __future__ import annotations
 
 import argparse
+from dataclasses import dataclass
 import hashlib
 import json
 import math
 import os
-import shutil
+import re
 import sys
 import tempfile
 import zipfile
 from datetime import date, datetime, timezone
 from pathlib import Path
 from typing import Any, Iterable
+from urllib.parse import urlsplit
 
 import requests
 from pyproj import Transformer
@@ -51,6 +53,10 @@ SOURCE_CRS = "EPSG:31370"
 OUTPUT_CRS = "EPSG:4326"
 SCHEMA_VERSION = 1
 MAX_ARCHIVE_BYTES = 250 * 1024 * 1024
+MAX_EXTRACTED_BYTES = 2 * 1024 * 1024 * 1024
+MAX_ARCHIVE_MEMBERS = 32
+ARCHIVE_HOST = "www.landbouwvlaanderen.be"
+ARCHIVE_PATH_PATTERN = re.compile(r"^/bestanden/gis/agpa_(20[0-9]{2})_(20[0-9]{2}-[0-9]{2}-[0-9]{2})_public\.zip$")
 
 ARCHIVE_URLS = {
     2008: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2008_2022-03-23_public.zip",
@@ -74,6 +80,16 @@ ARCHIVE_URLS = {
 }
 SUPPORTED_YEARS = tuple(ARCHIVE_URLS)
 
+
+@dataclass(frozen=True)
+class AgriculturalReleaseConfig:
+    year: int
+    archive_url: str
+
+    @property
+    def definitive_version(self) -> str:
+        return f"{self.year}-v3"
+
 STABLE_REQUIRED_FIELDS = {
     "agpakey",
     "parcelnumber",
@@ -131,6 +147,10 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL))
     parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY)
     parser.add_argument("--years", default=",".join(str(year) for year in SUPPORTED_YEARS))
+    parser.add_argument(
+        "--archive-url",
+        help="Exact official archive URL for one explicitly confirmed future definitive edition.",
+    )
     parser.add_argument("--output-root", type=Path, default=Path(os.environ.get("GEOINTEL_AGRICULTURE_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)))
     parser.add_argument("--request-timeout", type=int, default=900)
     parser.add_argument("--import-timeout", type=int, default=3600)
@@ -201,21 +221,29 @@ def response_data(response: requests.Response) -> Any:
 def api_items(session: requests.Session, url: str, timeout: int) -> list[dict[str, Any]]:
     items: list[dict[str, Any]] = []
     offset = 0
-    while True:
+    expected_total: int | None = None
+    while expected_total is None or offset < expected_total:
         response = session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout)
         data = response_data(response)
         if isinstance(data, dict):
             page = data.get("items") or data.get("results") or []
-            total = int(data.get("total") or len(page))
+            page_total = int(data.get("total") if data.get("total") is not None else len(page))
         else:
             page = data
-            total = len(page) if isinstance(page, list) else 0
+            page_total = len(page) if isinstance(page, list) else 0
         if not isinstance(page, list):
             raise RuntimeError(f"Expected a list response from {url}")
+        if expected_total is None:
+            expected_total = page_total
+        elif page_total != expected_total:
+            raise RuntimeError("GeoIntel pagination total changed while reading the agricultural workspace")
         items.extend(item for item in page if isinstance(item, dict))
-        if not page or len(items) >= total:
-            return items
+        if not page:
+            break
         offset += len(page)
+    if expected_total is not None and len(items) != expected_total:
+        raise RuntimeError(f"GeoIntel list response for {url} returned {len(items)} of {expected_total} items")
+    return items
 
 
 def locate_workspace(
@@ -251,6 +279,53 @@ def parse_years(raw: str) -> list[int]:
     return years
 
 
+def validate_archive_url(url: str, *, expected_year: int) -> str:
+    parsed = urlsplit(url)
+    match = ARCHIVE_PATH_PATTERN.fullmatch(parsed.path)
+    if (
+        parsed.scheme != "https"
+        or parsed.hostname != ARCHIVE_HOST
+        or parsed.port not in {None, 443}
+        or parsed.username
+        or parsed.password
+        or parsed.query
+        or parsed.fragment
+        or not match
+        or int(match.group(1)) != expected_year
+    ):
+        raise ValueError("Agricultural release archive is outside the official ALZ URL contract")
+    try:
+        datetime.strptime(match.group(2), "%Y-%m-%d")
+    except ValueError as exc:
+        raise ValueError("Agricultural release archive contains an invalid publication date") from exc
+    return url
+
+
+def resolve_release_config(year: int, *, archive_url: str | None = None) -> AgriculturalReleaseConfig:
+    known_url = ARCHIVE_URLS.get(year)
+    if known_url is not None:
+        if archive_url is not None and archive_url != known_url:
+            raise ValueError(f"The official retained archive identity for {year} may not be overridden")
+        return AgriculturalReleaseConfig(year=year, archive_url=known_url)
+    if year <= max(SUPPORTED_YEARS) or not archive_url:
+        raise ValueError(
+            f"One future definitive edition after {max(SUPPORTED_YEARS)} may be supplied with --archive-url"
+        )
+    return AgriculturalReleaseConfig(year=year, archive_url=validate_archive_url(archive_url, expected_year=year))
+
+
+def resolve_release_configs(raw_years: str, *, archive_url: str | None = None) -> list[AgriculturalReleaseConfig]:
+    if archive_url is None:
+        return [resolve_release_config(year) for year in parse_years(raw_years)]
+    try:
+        years = sorted({int(value.strip()) for value in raw_years.split(",") if value.strip()})
+    except ValueError as exc:
+        raise ValueError("Years must be a comma-separated list of integers") from exc
+    if len(years) != 1:
+        raise ValueError("--archive-url requires exactly one explicitly selected definitive year")
+    return [resolve_release_config(years[0], archive_url=archive_url)]
+
+
 def polygonal_geometry(geometry):
     if geometry is None or geometry.is_empty:
         return None
@@ -308,6 +383,10 @@ def download_archive(
     temporary.unlink(missing_ok=True)
     response = session.get(url, timeout=timeout, stream=True)
     response.raise_for_status()
+    final_url = str(getattr(response, "url", "") or url)
+    expected_year = int(ARCHIVE_PATH_PATTERN.fullmatch(urlsplit(url).path).group(1))
+    if validate_archive_url(final_url, expected_year=expected_year) != url:
+        raise RuntimeError("Official archive download redirected to a different release identity")
     content_length = int(response.headers.get("content-length") or 0)
     if content_length > max_bytes:
         raise RuntimeError(f"Official archive exceeds the configured {max_bytes // (1024 * 1024)} MiB limit")
@@ -331,7 +410,12 @@ def download_archive(
 
 def archive_geopackage_member(path: Path) -> str:
     with zipfile.ZipFile(path) as archive:
-        members = [item.filename for item in archive.infolist() if not item.is_dir() and item.filename.lower().endswith(".gpkg")]
+        entries = archive.infolist()
+        if len(entries) > MAX_ARCHIVE_MEMBERS:
+            raise RuntimeError(f"Official archive contains more than {MAX_ARCHIVE_MEMBERS} members")
+        if sum(item.file_size for item in entries if not item.is_dir()) > MAX_EXTRACTED_BYTES:
+            raise RuntimeError("Official archive exceeds the extracted-size safety limit")
+        members = [item.filename for item in entries if not item.is_dir() and item.filename.lower().endswith(".gpkg")]
     if len(members) != 1:
         raise RuntimeError(f"Official archive must contain exactly one GeoPackage; found {len(members)}")
     member = members[0]
@@ -351,7 +435,12 @@ def extract_geopackage(archive_path: Path, destination_dir: Path) -> Path:
     member = validate_archive(archive_path)
     destination = destination_dir / Path(member).name
     with zipfile.ZipFile(archive_path) as archive, archive.open(member) as source, destination.open("wb") as target:
-        shutil.copyfileobj(source, target, length=1024 * 1024)
+        extracted_bytes = 0
+        for chunk in iter(lambda: source.read(1024 * 1024), b""):
+            extracted_bytes += len(chunk)
+            if extracted_bytes > MAX_EXTRACTED_BYTES:
+                raise RuntimeError("Official GeoPackage exceeded the extracted-size safety limit while streaming")
+            target.write(chunk)
     if destination.stat().st_size == 0:
         raise RuntimeError("Extracted official GeoPackage is empty")
     return destination
@@ -470,23 +559,39 @@ def normalize_frame(frame, *, year: int, boundary_lambert72, max_features: int)
     }
 
 
-def artifact_paths(output_root: Path, scope_key: str, year: int) -> dict[str, Path]:
+def artifact_paths(
+    output_root: Path,
+    scope_key: str,
+    year: int,
+    *,
+    archive_url: str | None = None,
+) -> dict[str, Path]:
+    release = resolve_release_config(year, archive_url=archive_url)
     directory = output_root / scope_key / str(year)
     return {
         "directory": directory,
-        "archive": directory / Path(ARCHIVE_URLS[year]).name,
+        "archive": directory / Path(urlsplit(release.archive_url).path).name,
         "artifact": directory / f"agricultural_use_parcels_{year}_{scope_key}.geojson",
         "codelist": directory / f"agricultural_use_parcels_{year}_crop_codes.json",
         "manifest": directory / f"agricultural_use_parcels_{year}_{scope_key}.manifest.json",
     }
 
 
-def reusable_artifact(paths: dict[str, Path], *, year: int, scope_key: str) -> dict[str, Any] | None:
+def reusable_artifact(
+    paths: dict[str, Path],
+    *,
+    year: int,
+    scope_key: str,
+    archive_url: str | None = None,
+) -> dict[str, Any] | None:
     if not all(paths[key].is_file() for key in ("archive", "artifact", "codelist", "manifest")):
         return None
     manifest = json.loads(paths["manifest"].read_text(encoding="utf-8"))
     if manifest.get("schema_version") != SCHEMA_VERSION or manifest.get("year") != year or manifest.get("scope_key") != scope_key:
         return None
+    release = resolve_release_config(year, archive_url=archive_url)
+    if manifest.get("source_url") != release.archive_url:
+        return None
     if manifest.get("source_archive_sha256") != sha256_file(paths["archive"]):
         return None
     if manifest.get("artifact_sha256") != sha256_file(paths["artifact"]):
@@ -508,16 +613,18 @@ def prepare_year(
     max_archive_bytes: int,
     max_features: int,
     force: bool,
+    archive_url: str | None = None,
 ) -> tuple[dict[str, Path], dict[str, Any]]:
-    paths = artifact_paths(output_root, scope.key, year)
+    release = resolve_release_config(year, archive_url=archive_url)
+    paths = artifact_paths(output_root, scope.key, year, archive_url=release.archive_url)
     paths["directory"].mkdir(parents=True, exist_ok=True)
     if not force:
-        reused = reusable_artifact(paths, year=year, scope_key=scope.key)
+        reused = reusable_artifact(paths, year=year, scope_key=scope.key, archive_url=release.archive_url)
         if reused is not None:
             return paths, reused
     download = download_archive(
         session,
-        ARCHIVE_URLS[year],
+        release.archive_url,
         paths["archive"],
         timeout=request_timeout,
         max_bytes=max_archive_bytes,
@@ -546,7 +653,7 @@ def prepare_year(
         "scope_name": scope.display_name,
         "scope_type": scope.scope_type,
         "member_nis_codes": list(scope.nis_codes),
-        "source_url": ARCHIVE_URLS[year],
+        "source_url": release.archive_url,
         "catalog_url": CATALOG_URL,
         "data_catalog_url": DATA_CATALOG_URL,
         "attribution": ATTRIBUTION,
@@ -635,7 +742,7 @@ def upload_artifact(
         "operator_tool": "provision_agricultural_parcel_history.py",
         "operator_explicit_fetch": True,
         "geometry_clipped_to_area": True,
-        "source_archive_url": ARCHIVE_URLS[year],
+        "source_archive_url": manifest["source_url"],
         "source_archive_path": str(paths["archive"]),
         "source_archive_sha256": manifest["source_archive_sha256"],
         "crop_code_list_path": str(paths["codelist"]),
@@ -684,7 +791,7 @@ def existing_dataset_for_year(datasets: list[dict[str, Any]], *, area_id: str, y
 def main() -> int:
     args = parse_args()
     try:
-        years = parse_years(args.years)
+        releases = resolve_release_configs(args.years, archive_url=args.archive_url)
         if args.max_features < 1 or args.max_archive_mb < 1:
             raise ValueError("Feature and archive safety limits must be positive")
         scope = GEOGRAPHIC_SCOPES[args.scope]
@@ -693,7 +800,8 @@ def main() -> int:
             project_id, area_id, boundary, datasets = locate_workspace(api_session, base_url, scope, args.import_timeout)
             results: list[dict[str, Any]] = []
             with build_session() as official_session:
-                for year in years:
+                for release in releases:
+                    year = release.year
                     paths, manifest = prepare_year(
                         official_session,
                         year=year,
@@ -704,6 +812,7 @@ def main() -> int:
                         max_archive_bytes=min(args.max_archive_mb * 1024 * 1024, MAX_ARCHIVE_BYTES),
                         max_features=args.max_features,
                         force=args.force,
+                        archive_url=release.archive_url,
                     )
                     existing = existing_dataset_for_year(datasets, area_id=area_id, year=year)
                     if existing is not None:

scripts/run_readiness_check.sh

@@ -54,6 +54,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_waterinfo_station_history.py
 ${PYTHON_BIN} -m py_compile scripts/provision_mol_bwk_natura2000.py
 ${PYTHON_BIN} -m py_compile scripts/provision_regional_bwk_natura2000.py
 ${PYTHON_BIN} -m py_compile scripts/provision_agricultural_parcel_history.py
+${PYTHON_BIN} -m py_compile scripts/manage_alz_agriculture_release.py
 ${PYTHON_BIN} -m py_compile scripts/provision_buildings_addresses_register.py
 ${PYTHON_BIN} -m py_compile scripts/provision_mol_dhmv.py
 ${PYTHON_BIN} -m py_compile scripts/provision_regional_dhmv.py