Add governed Statbel release promotion

CHANGELOG.md

@@ -7,6 +7,25 @@
 
 # Changelog
 
+## Sprint 228 Governed Statbel population release promotion (2026-07-17)
+
+- Added an operator-only `plan -> stage -> review -> apply` coordinator for
+  future official Statbel population editions. Plan is read-only, stage is
+  preflight/filesystem-only, review requires a named approval and apply
+  requires exact plan plus review SHA-256 values.
+- Derived population and matching geometry URLs only from the allowlisted
+  catalog year/layout contract. Current, older, unavailable, ambiguous or
+  catalog-drifted releases fail closed; the coordinator accepts no arbitrary
+  provider URL and creates no scheduler or background fetch.
+- Extended the existing population provisioner to accept one complete
+  explicitly governed future release, bound downloads before buffering and
+  discover the latest retained baseline across dynamically added years.
+- Revalidated every source, manifest and snapshot byte immediately before the
+  canonical DatasetService upload. Dataset discovery now paginates completely,
+  preserving idempotence in workspaces with more than 200 datasets.
+- Added focused release-state, trust-boundary, tamper, review and apply tests,
+  Docker packaging and readiness compilation without API or migration changes.
+
 ## Sprint 227 Statbel population import compatibility preflight (2026-07-16)
 
 - Added a local, fail-closed preflight for staged official Statbel population

backend/README.md

@@ -1073,6 +1073,46 @@ excluded from spatial metrics. The 2025 REDEGEO contract deliberately compares
 explicit municipality fields; it does not assume that `CD_SECTOR` still starts
 with the current `CD_REFNIS` after municipal mergers.
 
+Future official editions use the separate four-phase release coordinator. The
+project id must belong to `Kempen Regional Workbench`:
+
+```bash
+docker exec geointel python /app/scripts/manage_statbel_population_release.py plan \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --refresh-catalog
+
+docker exec geointel python /app/scripts/manage_statbel_population_release.py stage \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --confirm-edition <YEAR_FROM_PLAN> \
+  --confirm-layout <LAYOUT_FROM_PLAN>
+
+docker exec geointel python /app/scripts/manage_statbel_population_release.py review \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --confirm-edition <YEAR_FROM_PLAN> \
+  --confirm-layout <LAYOUT_FROM_PLAN> \
+  --confirm-plan-sha256 <SHA256_FROM_STAGE> \
+  --approve --reviewer "<OPERATOR_NAME>" \
+  --review-note "Schema, totalen, ZZZZ en geometrieherstel nagekeken"
+
+docker exec geointel python /app/scripts/manage_statbel_population_release.py apply \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --confirm-edition <YEAR_FROM_PLAN> \
+  --confirm-layout <LAYOUT_FROM_PLAN> \
+  --confirm-plan-sha256 <SHA256_FROM_STAGE> \
+  --confirm-review-sha256 <SHA256_FROM_REVIEW>
+```
+
+`plan` is read-only and creates no file. `stage` always uses bounded fresh
+downloads and `--fetch-only`; `review` imports nothing; `apply` revalidates
+the current catalog, plan, review, source archives, preflight manifest and
+derived snapshot before using the existing upload API. An already-current
+release cannot be staged. A repeated successful apply resolves the existing
+Dataset through complete paginated lookup rather than creating a duplicate.
+
 Historical land-use work can be bounded explicitly:
 
 ```bash

backend/tests/test_sprint228_statbel_release_management.py

@@ -0,0 +1,437 @@
+from __future__ import annotations
+
+import argparse
+from hashlib import sha256
+import importlib.util
+import json
+from pathlib import Path
+import sys
+
+import pytest
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPTS = ROOT / "scripts"
+if str(SCRIPTS) not in sys.path:
+    sys.path.insert(0, str(SCRIPTS))
+
+
+def load_script(name: str):
+    path = SCRIPTS / name
+    module_name = f"test_{path.stem}_sprint228"
+    spec = importlib.util.spec_from_file_location(module_name, path)
+    assert spec is not None
+    assert spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[module_name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+MANAGER = load_script("manage_statbel_population_release.py")
+OPERATOR = load_script("provision_mol_population_history.py")
+
+
+def arguments(tmp_path: Path, **overrides) -> argparse.Namespace:
+    values = {
+        "action": "plan",
+        "project_id": "00000000-0000-0000-0000-000000000001",
+        "scope": "kempen-transport-region",
+        "api_url": "http://127.0.0.1:8000/api/v1",
+        "confirm_edition": None,
+        "confirm_layout": None,
+        "confirm_plan_sha256": None,
+        "confirm_review_sha256": None,
+        "approve": False,
+        "reviewer": None,
+        "review_note": "",
+        "plan_path": None,
+        "review_path": None,
+        "output_root": tmp_path / "operator-data" / "regional-timeseries",
+        "scope_output_root": tmp_path / "operator-data" / "geographic-scopes",
+        "evidence_root": tmp_path / "operator-evidence" / "statbel-population-refresh",
+        "refresh_catalog": False,
+        "request_timeout": 300,
+        "api_timeout": 180,
+        "import_timeout": 3600,
+    }
+    values.update(overrides)
+    return argparse.Namespace(**values)
+
+
+def catalog_item(*, remote: str = "2026", local: str | None = "2025", catalog_hash: str = "a" * 64) -> dict:
+    return {
+        "source_name": "statbel",
+        "status": "available",
+        "reachable": True,
+        "error_code": None,
+        "remote_version": remote,
+        "local_source_version": local,
+        "remote_title": f"Bevolking per statistische sector {remote} (nieuwe REDEGEO-sectorindeling)",
+        "message": "De officiele catalogus bevestigt de nieuwe REDEGEO-sectorindeling.",
+        "matched_layers": ["population_txt_current", "landing_page", "cc_by_4_0"],
+        "capabilities_sha256": catalog_hash,
+        "metadata_identifier": f"NodeID{remote}",
+        "metadata_url": f"https://statbel.fgov.be/nl/open-data/bevolking-statistische-sector-{remote}",
+        "checked_at": "2026-07-17T08:00:00Z",
+    }
+
+
+def decision(args: argparse.Namespace, *, remote: str = "2026", local: str | None = "2025") -> dict:
+    return MANAGER.fetch_release_decision_from_item(args, catalog_item(remote=remote, local=local))
+
+
+def release_2026():
+    return MANAGER.release_from_catalog_item(catalog_item())
+
+
+def write_staged_artifacts(args: argparse.Namespace, release) -> None:
+    scope = MANAGER.GEOGRAPHIC_SCOPES[args.scope]
+    output_dir = MANAGER.population_output_dir(args)
+    raw_dir = output_dir / "raw" / str(release.year)
+    raw_dir.mkdir(parents=True, exist_ok=True)
+    population_path = raw_dir / "OPENDATA_SECTOREN_2026_NEW.zip"
+    geometry_path = raw_dir / "sh_statbel_statistical_sectors_31370_20260101.geojson.zip"
+    population_path.write_bytes(b"official population archive")
+    geometry_path.write_bytes(b"official geometry archive")
+    snapshot = MANAGER.snapshot_path(output_dir, scope, release.year)
+    snapshot.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
+    manifest = {
+        "schema_version": 1,
+        "status": "passed",
+        "import_eligible": True,
+        "release": {"year": 2026, "population_layout": "new"},
+        "artifacts": {
+            "population": {
+                "source_url": release.population_url,
+                "archive_sha256": sha256(population_path.read_bytes()).hexdigest(),
+                "archive_size_bytes": population_path.stat().st_size,
+                "retained_path": str(population_path),
+            },
+            "geometry": {
+                "source_url": release.geometry_url,
+                "archive_sha256": sha256(geometry_path.read_bytes()).hexdigest(),
+                "archive_size_bytes": geometry_path.stat().st_size,
+                "retained_path": str(geometry_path),
+            },
+            "derived_snapshot": {
+                "sha256": sha256(snapshot.read_bytes()).hexdigest(),
+                "size_bytes": snapshot.stat().st_size,
+                "feature_count": 700,
+            },
+        },
+        "scope_accounting": {
+            "scope_key": scope.key,
+            "spatial_sector_count": 700,
+            "spatial_population_total": 510000,
+            "unlocated_population_total": 300,
+            "accounted_population_total": 510300,
+        },
+        "national_accounting": {"population_total": 12000000},
+        "baseline": {"year": 2025, "annual_change_ratio": 0.007},
+        "schemas": {"geometry_repair_count": 2},
+    }
+    manifest_path = MANAGER.preflight_manifest_path(output_dir, scope, release.year)
+    manifest_path.write_text(json.dumps(manifest), encoding="utf-8")
+
+
+def staged_plan(args: argparse.Namespace, release, release_decision: dict) -> tuple[Path, dict]:
+    write_staged_artifacts(args, release)
+    result = {
+        "status": "ok",
+        "scope": args.scope,
+        "snapshots": [
+            {
+                "year": release.year,
+                "status": "prepared",
+                "preflight_status": "passed",
+                "feature_count": 700,
+            }
+        ],
+    }
+    plan = MANAGER.build_staged_plan(args, release_decision, release, result)
+    path = MANAGER.default_plan_path(args, release.year)
+    MANAGER.write_json(path, plan)
+    return path, plan
+
+
+def test_future_release_config_is_strict_and_previous_snapshot_is_discovered(tmp_path: Path) -> None:
+    release = OPERATOR.resolve_release_config(
+        2026,
+        layout="new",
+        population_url=(
+            "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/"
+            "OPENDATA_SECTOREN_2026_NEW.zip"
+        ),
+        geometry_url=(
+            "https://statbel.fgov.be/sites/default/files/files/opendata/Statistische%20sectoren/"
+            "sh_statbel_statistical_sectors_31370_20260101.geojson.zip"
+        ),
+    )
+    scope = OPERATOR.GEOGRAPHIC_SCOPES["mol"]
+    baseline = OPERATOR.snapshot_path(tmp_path, scope, 2026)
+    baseline.write_text("{}", encoding="utf-8")
+
+    assert release.year == 2026
+    assert release.layout == "new"
+    assert OPERATOR.previous_snapshot_path(tmp_path, scope, 2027) == baseline
+    with pytest.raises(ValueError, match="supplied together"):
+        OPERATOR.resolve_release_config(2026, layout="new")
+
+
+def test_population_workspace_pagination_reads_every_dataset() -> None:
+    rows = [{"id": index} for index in range(401)]
+
+    class Response:
+        ok = True
+        status_code = 200
+        text = ""
+
+        def __init__(self, payload: dict) -> None:
+            self.payload = payload
+
+        def json(self) -> dict:
+            return {"data": self.payload}
+
+    class Session:
+        def get(self, _url: str, *, params: dict, timeout: int):
+            assert timeout == 30
+            offset = int(params["offset"])
+            limit = int(params["limit"])
+            return Response({"items": rows[offset : offset + limit], "total": len(rows)})
+
+    assert OPERATOR.list_paginated_items(Session(), "http://backend/datasets", timeout=30) == rows
+
+
+def test_population_archive_download_is_bounded_before_streaming() -> None:
+    class Response:
+        url = (
+            "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/"
+            "OPENDATA_SECTOREN_2026_NEW.zip"
+        )
+        headers = {"Content-Length": str(OPERATOR.MAX_POPULATION_ARCHIVE_BYTES + 1)}
+
+        def __enter__(self):
+            return self
+
+        def __exit__(self, *_args):
+            return False
+
+        def raise_for_status(self) -> None:
+            return None
+
+        def iter_content(self, *, chunk_size: int):
+            raise AssertionError(f"download should fail before streaming {chunk_size}")
+
+    class Session:
+        def get(self, *_args, **_kwargs):
+            return Response()
+
+    with pytest.raises(RuntimeError, match="download limit"):
+        OPERATOR.download_archive(
+            Session(),
+            url=Response.url,
+            year=2026,
+            layout="new",
+            max_bytes=OPERATOR.MAX_POPULATION_ARCHIVE_BYTES,
+            timeout=30,
+        )
+
+
+@pytest.mark.parametrize(
+    ("remote", "local", "expected"),
+    [
+        ("2026", "2025", "update_available"),
+        ("2025", "2025", "current"),
+        ("2026", None, "not_loaded"),
+        ("2025", "2026", "blocked_remote_older"),
+    ],
+)
+def test_catalog_decision_orders_remote_and_local_editions(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    remote: str,
+    local: str | None,
+    expected: str,
+) -> None:
+    args = arguments(tmp_path)
+    monkeypatch.setattr(
+        MANAGER,
+        "api_data",
+        lambda *_args, **_kwargs: {"items": [catalog_item(remote=remote, local=local)]},
+    )
+
+    result = MANAGER.fetch_release_decision(args, refresh=True)
+
+    assert result["status"] == expected
+    assert result["release"]["year"] == int(remote)
+    assert result["automatic_download"] is False
+    assert result["automatic_import"] is False
+
+
+def test_operator_commands_separate_staging_from_apply(tmp_path: Path) -> None:
+    args = arguments(tmp_path)
+    release = release_2026()
+
+    stage = MANAGER.build_operator_command(args, release, fetch_only=True)
+    apply = MANAGER.build_operator_command(args, release, fetch_only=False)
+
+    assert "--force" in stage
+    assert "--fetch-only" in stage
+    assert "--force" not in apply
+    assert "--fetch-only" not in apply
+    assert stage[stage.index("--population-layout") + 1] == "new"
+    assert stage[stage.index("--population-url") + 1] == release.population_url
+    assert stage[stage.index("--geometry-url") + 1] == release.geometry_url
+
+
+def test_staged_plan_and_review_require_exact_hashes(tmp_path: Path) -> None:
+    args = arguments(tmp_path)
+    release = release_2026()
+    release_decision = decision(args)
+    plan_path, plan = staged_plan(args, release, release_decision)
+    args.confirm_plan_sha256 = plan["plan_sha256"]
+
+    loaded_path, loaded = MANAGER.load_staged_plan(args, release)
+    assert loaded_path == plan_path
+    assert loaded["evidence"]["scope_accounting"]["accounted_population_total"] == 510300
+
+    args.approve = True
+    args.reviewer = "GeoIntel operator"
+    args.review_note = "Schema, totalen en ZZZZ-accounting nagekeken."
+    review = MANAGER.build_review_evidence(args, plan_path, plan)
+    review_path = MANAGER.default_review_path(args, release.year)
+    MANAGER.write_json(review_path, review)
+    args.confirm_review_sha256 = review["review_sha256"]
+
+    loaded_review_path, loaded_review = MANAGER.load_review_evidence(args, release, plan)
+    assert loaded_review_path == review_path
+    assert loaded_review["status"] == "approved"
+    assert "scope_and_national_accounting" in loaded_review["reviewed_checks"]
+
+
+def test_tampered_source_or_review_is_rejected(tmp_path: Path) -> None:
+    args = arguments(tmp_path)
+    release = release_2026()
+    release_decision = decision(args)
+    plan_path, plan = staged_plan(args, release, release_decision)
+    args.confirm_plan_sha256 = plan["plan_sha256"]
+    population_path = Path(plan["evidence"]["population_archive"]["retained_path"])
+    population_path.write_bytes(population_path.read_bytes() + b"tampered")
+
+    with pytest.raises(RuntimeError, match="population archive"):
+        MANAGER.load_staged_plan(args, release)
+
+    write_staged_artifacts(args, release)
+    args.approve = True
+    args.reviewer = "Operator"
+    review = MANAGER.build_review_evidence(args, plan_path, plan)
+    review_path = MANAGER.default_review_path(args, release.year)
+    MANAGER.write_json(review_path, review)
+    review_payload = json.loads(review_path.read_text(encoding="utf-8"))
+    review_payload["reviewer"] = "Someone else"
+    review_path.write_text(json.dumps(review_payload), encoding="utf-8")
+    args.confirm_review_sha256 = review["review_sha256"]
+
+    with pytest.raises(RuntimeError, match="checksum"):
+        MANAGER.load_review_evidence(args, release, plan)
+
+
+def test_catalog_drift_and_outside_evidence_path_are_rejected(tmp_path: Path) -> None:
+    args = arguments(tmp_path)
+    original = decision(args)
+    changed = json.loads(json.dumps(original))
+    changed["catalog_identity"]["capabilities_sha256"] = "b" * 64
+
+    with pytest.raises(RuntimeError, match="changed"):
+        MANAGER.require_catalog_unchanged(
+            {"release": original["release"], "catalog_identity": original["catalog_identity"]},
+            changed,
+        )
+    with pytest.raises(RuntimeError, match="outside"):
+        MANAGER.governed_evidence_path(args, tmp_path / "outside.json")
+
+
+def test_apply_flow_requires_approved_review_and_writes_evidence(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    args = arguments(
+        tmp_path,
+        action="apply",
+        confirm_edition="2026",
+        confirm_layout="new",
+    )
+    release = release_2026()
+    release_decision = decision(args)
+    plan_path, plan = staged_plan(args, release, release_decision)
+    args.confirm_plan_sha256 = plan["plan_sha256"]
+    args.approve = True
+    args.reviewer = "GeoIntel operator"
+    review = MANAGER.build_review_evidence(args, plan_path, plan)
+    review_path = MANAGER.default_review_path(args, release.year)
+    MANAGER.write_json(review_path, review)
+    args.confirm_review_sha256 = review["review_sha256"]
+    final_decision = json.loads(json.dumps(release_decision))
+    final_decision["status"] = "current"
+    final_decision["local_source_version"] = "2026"
+    decisions = iter((release_decision, final_decision))
+    monkeypatch.setattr(MANAGER, "parse_args", lambda: args)
+    monkeypatch.setattr(MANAGER, "validate_project_scope", lambda _args: None)
+    monkeypatch.setattr(MANAGER, "fetch_release_decision", lambda *_args, **_kwargs: next(decisions))
+    monkeypatch.setattr(
+        MANAGER,
+        "run_operator",
+        lambda *_args, **_kwargs: {
+            "status": "ok",
+            "snapshots": [{"year": 2026, "status": "imported", "dataset_id": "dataset-2026", "feature_count": 700}],
+        },
+    )
+
+    assert MANAGER.main() == 0
+    applied_path = plan_path.with_name("applied-evidence.json")
+    applied = json.loads(applied_path.read_text(encoding="utf-8"))
+    assert applied["dataset_id"] == "dataset-2026"
+    assert applied["review_sha256"] == review["review_sha256"]
+    assert len(applied["applied_evidence_sha256"]) == 64
+
+
+def test_current_release_cannot_be_staged(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    args = arguments(tmp_path, action="stage", confirm_edition="2025", confirm_layout="new")
+    current = decision(args, remote="2025", local="2025")
+    monkeypatch.setattr(MANAGER, "parse_args", lambda: args)
+    monkeypatch.setattr(MANAGER, "validate_project_scope", lambda _args: None)
+    monkeypatch.setattr(MANAGER, "fetch_release_decision", lambda *_args, **_kwargs: current)
+
+    assert MANAGER.main() == 1
+    assert "not safely stageable: current" in capsys.readouterr().err
+
+
+def test_plan_action_is_read_only_for_current_release(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+) -> None:
+    args = arguments(tmp_path, action="plan")
+    current = decision(args, remote="2025", local="2025")
+    monkeypatch.setattr(MANAGER, "parse_args", lambda: args)
+    monkeypatch.setattr(MANAGER, "validate_project_scope", lambda _args: None)
+    monkeypatch.setattr(MANAGER, "fetch_release_decision", lambda *_args, **_kwargs: current)
+
+    assert MANAGER.main() == 0
+    output = json.loads(capsys.readouterr().out)
+    assert output["status"] == "ok"
+    assert output["decision"]["status"] == "current"
+    assert not args.evidence_root.exists()
+
+
+def test_release_manager_is_packaged_and_release_checked() -> None:
+    dockerfile = (ROOT / "deploy/unraid/Dockerfile.all-in-one").read_text(encoding="utf-8")
+    readiness = (ROOT / "scripts/run_readiness_check.sh").read_text(encoding="utf-8")
+
+    assert "COPY scripts/manage_statbel_population_release.py" in dockerfile
+    assert "py_compile scripts/manage_statbel_population_release.py" in readiness

deploy/unraid/Dockerfile.all-in-one

@@ -83,6 +83,7 @@ COPY scripts/provision_mol_soil_map.py /app/scripts/provision_mol_soil_map.py
 COPY scripts/provision_regional_soil_map.py /app/scripts/provision_regional_soil_map.py
 COPY scripts/provision_mol_population_history.py /app/scripts/provision_mol_population_history.py
 COPY scripts/statbel_population_preflight.py /app/scripts/statbel_population_preflight.py
+COPY scripts/manage_statbel_population_release.py /app/scripts/manage_statbel_population_release.py
 COPY scripts/provision_mol_historical_landuse.py /app/scripts/provision_mol_historical_landuse.py
 COPY scripts/provision_regional_historical_landuse.py /app/scripts/provision_regional_historical_landuse.py
 COPY scripts/provision_official_landuse_timeseries.py /app/scripts/provision_official_landuse_timeseries.py

docs/API_CONTRACTS.md

@@ -450,6 +450,13 @@ reported only as transition evidence. `service_type=DCAT` uses
 A newer statistical-sector geometry edition is not interpreted as a newer
 population release.
 
+Future Statbel execution remains outside the HTTP request cycle in
+`scripts/manage_statbel_population_release.py`. It reuses this read-only
+catalog response for `plan`, then separates filesystem-only `stage`, named
+human `review` and checksum-confirmed `apply`. No additional public endpoint
+is introduced. Apply delegates to the existing dataset upload contract and
+creates a new immutable annual snapshot only after all evidence is unchanged.
+
 The endpoint accepts no arbitrary URL, feature query, area or layer. It does
 not fetch vector features, raster pixels or models, create jobs/datasets, write
 to PostGIS or trigger an import. The normal `source-freshness` endpoint remains

docs/CODEX_EXECUTION_LOG.md

@@ -1,3 +1,55 @@
+## Sprint 228 - Governed Statbel population release promotion (2026-07-17)
+
+Implemented:
+- Added `scripts/manage_statbel_population_release.py` with four separate
+  operator actions for the approved Kempen scope. `plan` is read-only; `stage`
+  requires exact year/layout confirmation and performs bounded fresh download
+  plus preflight only; `review` requires a named explicit approval; `apply`
+  requires the exact staged-plan and review-evidence SHA-256 values.
+- Derived the only accepted population and matching sector-geometry URLs from
+  the allowlisted catalog year/layout contract. The coordinator accepts no
+  arbitrary source URL or process, and current, older, unavailable, ambiguous
+  or catalog-drifted releases are not stageable.
+- Bound catalog identity, source archives, preflight manifest, derived
+  snapshot, scope/national accounting, ZZZZ totals, baseline trend and geometry
+  repairs into `staged-plan.json`. Bound the named human decision to that plan
+  in `review-evidence.json`; successful apply records the immutable Dataset in
+  `applied-evidence.json` without deleting prior snapshots.
+- Extended `provision_mol_population_history.py` with a complete all-or-none
+  release config for one future year. Existing 2021-2025 arguments remain
+  compatible. Source downloads now enforce response and streaming byte bounds,
+  validate final official URLs and discover the latest retained baseline from
+  actual snapshot files rather than a hardcoded year list.
+- Replaced the population operator's 200-row workspace lookup with complete,
+  total-consistent pagination. This keeps repeated apply idempotent in the live
+  project with 2,429 Datasets.
+- Added Docker/readiness packaging and updated source, API, persistence,
+  storage and operator documentation. No API route, migration, scheduler,
+  automatic fetch or frontend behavior changed.
+
+Validation:
+- 15 focused Sprint 228 tests and 38 combined Sprint 194/227/228 tests passed.
+  Coverage includes release ordering, future URL/layout validation, dynamic
+  baseline discovery, bounded downloads, >200-row pagination, stage/apply
+  command separation, plan/review hashes, source/review tampering, evidence-root
+  confinement, catalog drift, named review, apply evidence and current-edition
+  refusal.
+- Complete readiness passed with 830 backend tests, 110 documented routes,
+  one Alembic head `202607160001`, frontend typecheck and production build.
+  Static Alembic SQL, shell syntax, target Ruff and diff checks passed.
+- A compatibility run against the live Tower API used the temporary candidate
+  scripts before deployment. The official Statbel catalog reported
+  `NodeID6475`, remote/local edition `2025`, layout `new`, catalog hash
+  `64b17ce059a9c2f936d4b5741b20aed4b178409d8b5be81be86e6fdc9fe7c9d9` and
+  decision `current`; `plan` wrote no evidence. A deliberate `stage` attempt
+  for that current edition exited 1 with `not safely stageable: current` and
+  left the database Dataset count unchanged at 2,429.
+
+Boundary:
+- No newer population edition is currently advertised, so no real stage,
+  review or apply was executed. Their complete state machine is fixture-tested;
+  the first future release must still pass all four explicit operator phases.
+
 ## Sprint 227 - Statbel population import compatibility preflight (2026-07-16)
 
 Implemented:

docs/DATABASE_IMPLEMENTATION_PLAN.md

@@ -274,6 +274,15 @@ new Dataset plus DatasetVersion and vector_features through the existing
 DatasetService/VectorFeatureService transaction; prior snapshots remain
 unchanged and queryable for temporal comparison.
 
+Statbel population release management likewise adds no lifecycle table.
+Planning reads the existing source-catalog response. Stage, named review and
+their SHA-256-bound evidence are filesystem-only operator artifacts. Apply
+revalidates those artifacts and invokes the existing Dataset upload service,
+which creates the ordinary annual `datasets`, `dataset_versions` and
+`vector_features` records in one established persistence flow. An existing
+year remains idempotent and previous annual snapshots are never updated or
+deleted.
+
 ## Geometry normalization
 
 - User-drawn polygons arrive as EPSG:4326.

docs/DATA_SOURCES.md

@@ -222,6 +222,22 @@ bronarchieven en de afgeleide GeoJSON. Zij geeft alleen technische
 importgeschiktheid aan; zij vervangt of importeert nooit automatisch een
 bestaande Dataset.
 
+`scripts/manage_statbel_population_release.py` beheert een toekomstige editie
+in vier afzonderlijke operatorstappen. `plan` vergelijkt de nieuwste
+allowlisted DCAT-editie read-only met de lokale bronversie. `stage` vereist
+expliciete bevestiging van jaar en REDEGEO-layout, haalt de exact uit het
+cataloguscontract afgeleide Statbel-URL's begrensd op en voert alleen de
+preflight/artefactstaging uit. `review` schrijft pas na een benoemde menselijke
+goedkeuring SHA-256-gebonden reviewevidence. `apply` vereist zowel plan- als
+reviewhash, controleert catalogusidentiteit en alle bestanden opnieuw en maakt
+via de bestaande uploadservice hoogstens een nieuwe immutable Dataset.
+
+De coordinator accepteert geen vrije provider-URL, start geen achtergrondtaak
+en kan een actuele of oudere remote editie niet stagen. Een ontbrekende
+bijbehorende sectorgeometrie, gewijzigde catalogus, gewijzigde bronbyte,
+onverklaarde totalensprong of onvolledige review blokkeert de flow. Bestaande
+snapshots blijven beschikbaar voor historische vergelijking.
+
 ### Mol population history
 
 `scripts/provision_mol_population_history.py` imports official Statbel

docs/STORAGE_ARCHITECTURE.md

@@ -122,6 +122,23 @@ Dataset/DatasetVersion plus PostGIS `vector_features` through the canonical
 persistence services. Temporary standalone preflight output does not create a
 database record and may be removed explicitly after operator review.
 
+Governed release-decision evidence is separate from the source artifacts:
+
+```text
+storage/operator-evidence/statbel-population-refresh/{scope}/{year}/
+  staged-plan.json
+  review-evidence.json
+  applied-evidence.json
+```
+
+The staged plan binds the current official catalog identity to every retained
+source/snapshot/preflight hash and accounting summary. Review evidence binds a
+named approval to that exact plan. Applied evidence binds both approvals to
+the resulting Dataset id. These JSON files are operator audit artifacts, not
+database lifecycle entities. A plan/review path outside the configured
+evidence root or a retained archive outside the population output root is
+rejected. No evidence file authorizes deletion or in-place replacement.
+
 Waterinfo raw station layers, timeseries responses and checksum manifests live
 under `storage/operator-data/waterinfo/<scope>/`. These are immutable source
 evidence; queryable annual Point snapshots are normal Dataset/vector_feature

docs/TODO.md

@@ -44,6 +44,7 @@
 - [x] Add a fail-closed ALZ publication probe that distinguishes provisional v1/v2 snapshots from the definitive v3 historical edition and never downloads an archive.
 - [x] Add a fail-closed Statbel DCAT publication probe that distinguishes population year, sector-geometry year and the 2025 REDEGEO transition without downloading distributions.
 - [x] Add a fail-closed Statbel population import preflight with archive/schema/CRS/join/total/baseline checks, retained checksums and explicit ZZZZ accounting before any new Dataset import.
+- [x] Add an explicit Statbel population plan -> stage -> named review -> checksum-confirmed apply workflow that preserves every prior annual snapshot.
 - [ ] Extend catalogue probes only to additional sources that publish a stable official edition contract; do not add background polling or infer releases from HTTP dates alone.
 
 ## Governed source expansion backlog

scripts/README.md

@@ -1752,6 +1752,43 @@ to the DatasetService-based persistence path. Existing snapshots are retained.
 The coordinator accepts no provider URL, collection name, arbitrary process or
 automatic schedule.
 
+## Governed Statbel population release
+
+Use the population release coordinator only inside the GeoIntel container.
+It derives the exact population and matching sector-geometry URLs from the
+strict official catalog contract; operators cannot inject another provider
+URL.
+
+```bash
+docker exec geointel python /app/scripts/manage_statbel_population_release.py plan \
+  --project-id <KEMPEN_PROJECT_ID> --refresh-catalog
+
+docker exec geointel python /app/scripts/manage_statbel_population_release.py stage \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --confirm-edition <YEAR_FROM_PLAN> --confirm-layout <LAYOUT_FROM_PLAN>
+
+docker exec geointel python /app/scripts/manage_statbel_population_release.py review \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --confirm-edition <YEAR_FROM_PLAN> --confirm-layout <LAYOUT_FROM_PLAN> \
+  --confirm-plan-sha256 <SHA256_FROM_STAGE> \
+  --approve --reviewer "<OPERATOR_NAME>"
+
+docker exec geointel python /app/scripts/manage_statbel_population_release.py apply \
+  --project-id <KEMPEN_PROJECT_ID> \
+  --confirm-edition <YEAR_FROM_PLAN> --confirm-layout <LAYOUT_FROM_PLAN> \
+  --confirm-plan-sha256 <SHA256_FROM_STAGE> \
+  --confirm-review-sha256 <SHA256_FROM_REVIEW>
+```
+
+`plan` writes nothing. `stage` downloads within the population/geometry size
+bounds, invokes the existing provisioner with `--force --fetch-only` and emits
+`staged-plan.json`. `review` requires a named approval and emits
+`review-evidence.json`. `apply` refuses catalog drift or any changed source,
+manifest, snapshot, plan or review byte, then delegates to the canonical
+DatasetService upload. `applied-evidence.json` records the resulting Dataset
+without deleting any historical snapshot. Current, older, unavailable or
+ambiguous releases are fail-closed.
+
 ## Tower deployment
 
 Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime:

scripts/manage_statbel_population_release.py

@@ -0,0 +1,585 @@
+#!/usr/bin/env python3
+"""Plan, stage, review and explicitly apply one Statbel population edition.
+
+Planning is read-only. Staging downloads official artifacts and runs the
+existing fail-closed preflight without database persistence. Review records an
+explicit named approval. Apply requires both exact evidence hashes and reuses
+the canonical population 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_mol_population_history import (
+    PopulationReleaseConfig,
+    load_preflight_manifest,
+    preflight_manifest_path,
+    resolve_release_config,
+    sha256_path,
+    snapshot_path,
+)
+
+
+DEFAULT_API_URL = "http://127.0.0.1:8000/api/v1"
+DEFAULT_SCOPE = "kempen-transport-region"
+DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-data/regional-timeseries")
+DEFAULT_SCOPE_OUTPUT_ROOT = Path("/app/storage/operator-data/geographic-scopes")
+DEFAULT_EVIDENCE_ROOT = Path("/app/storage/operator-evidence/statbel-population-refresh")
+YEAR_PATTERN = re.compile(r"^20[0-9]{2}$")
+SHA256_PATTERN = re.compile(r"^[0-9a-f]{64}$")
+ACTIONABLE_STATUSES = {"update_available", "not_loaded"}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(
+        description="Governed Statbel population 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 population year required after plan")
+    parser.add_argument("--confirm-layout", choices=("standard", "new"), help="Exact planned REDEGEO layout")
+    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_REGIONAL_TIMESERIES_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)),
+    )
+    parser.add_argument(
+        "--scope-output-root",
+        type=Path,
+        default=Path(os.environ.get("GEOINTEL_SCOPE_OUTPUT_ROOT", DEFAULT_SCOPE_OUTPUT_ROOT)),
+    )
+    parser.add_argument(
+        "--evidence-root",
+        type=Path,
+        default=Path(os.environ.get("GEOINTEL_STATBEL_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=300)
+    parser.add_argument("--api-timeout", type=int, default=180)
+    parser.add_argument("--import-timeout", type=int, default=3600)
+    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-Statbel-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 _layout_from_catalog_item(item: dict[str, Any], year: int) -> str:
+    evidence = f"{item.get('remote_title') or ''} {item.get('message') or ''}".casefold()
+    if "nieuwe redegeo-sectorindeling" in evidence:
+        return "new"
+    if year == 2025:
+        raise RuntimeError("The official 2025 release is missing explicit new-REDEGEO evidence")
+    if "actuele sectorindeling" in evidence:
+        return "standard"
+    raise RuntimeError("The official Statbel catalog did not expose a recognized population layout")
+
+
+def release_from_catalog_item(item: dict[str, Any]) -> PopulationReleaseConfig:
+    version = str(item.get("remote_version") or "")
+    if not YEAR_PATTERN.fullmatch(version):
+        raise RuntimeError("The official Statbel catalog did not provide one valid population year")
+    year = int(version)
+    layout = _layout_from_catalog_item(item, year)
+    suffix = "_NEW" if layout == "new" else ""
+    return resolve_release_config(
+        year,
+        layout=layout,
+        population_url=(
+            "https://statbel.fgov.be/sites/default/files/files/opendata/bevolking/sectoren/"
+            f"OPENDATA_SECTOREN_{year}{suffix}.zip"
+        ),
+        geometry_url=(
+            "https://statbel.fgov.be/sites/default/files/files/opendata/Statistische%20sectoren/"
+            f"sh_statbel_statistical_sectors_31370_{year}0101.geojson.zip"
+        ),
+    )
+
+
+def fetch_release_decision_from_item(args: argparse.Namespace, item: dict[str, Any]) -> dict[str, Any]:
+    if (
+        item.get("source_name") != "statbel"
+        or item.get("status") != "available"
+        or item.get("reachable") is not True
+        or item.get("error_code")
+        or not SHA256_PATTERN.fullmatch(str(item.get("capabilities_sha256") or ""))
+        or not item.get("metadata_identifier")
+    ):
+        raise RuntimeError("The official Statbel catalog is not safely available for release planning")
+    required_evidence = {"population_txt_current", "landing_page", "cc_by_4_0"}
+    if not required_evidence.issubset(set(item.get("matched_layers") or [])):
+        raise RuntimeError("The official Statbel catalog is missing required release evidence")
+    release = release_from_catalog_item(item)
+    local_version = str(item.get("local_source_version") or "")
+    if local_version and not YEAR_PATTERN.fullmatch(local_version):
+        status = "blocked_local_version"
+    elif not local_version:
+        status = "not_loaded"
+    elif int(local_version) == release.year:
+        status = "current"
+    elif int(local_version) < release.year:
+        status = "update_available"
+    else:
+        status = "blocked_remote_older"
+    return {
+        "schema_version": 1,
+        "status": status,
+        "project_id": args.project_id,
+        "scope": args.scope,
+        "generated_at": datetime.now(timezone.utc).isoformat(),
+        "local_source_version": local_version or None,
+        "release": {
+            "year": release.year,
+            "layout": release.layout,
+            "population_url": release.population_url,
+            "geometry_url": release.geometry_url,
+        },
+        "catalog_identity": {
+            "metadata_identifier": item["metadata_identifier"],
+            "metadata_url": item.get("metadata_url"),
+            "remote_version": str(release.year),
+            "capabilities_sha256": item["capabilities_sha256"],
+            "catalog_checked_at": item.get("checked_at"),
+        },
+        "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") == "statbel"]
+    if len(matches) != 1:
+        raise RuntimeError("Source catalog report did not contain exactly one Statbel population contract")
+    return fetch_release_decision_from_item(args, matches[0])
+
+
+def require_release_confirmation(args: argparse.Namespace, release: PopulationReleaseConfig) -> None:
+    if args.confirm_edition != str(release.year):
+        raise RuntimeError(f"Explicit --confirm-edition {release.year} is required")
+    if args.confirm_layout != release.layout:
+        raise RuntimeError(f"Explicit --confirm-layout {release.layout} 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 population_output_dir(args: argparse.Namespace) -> Path:
+    return args.output_root / args.scope / "population"
+
+
+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: PopulationReleaseConfig,
+    *,
+    fetch_only: bool,
+) -> list[str]:
+    scope = GEOGRAPHIC_SCOPES[args.scope]
+    command = [
+        sys.executable,
+        str(Path(__file__).resolve().parent / "provision_mol_population_history.py"),
+        "--scope",
+        scope.key,
+        "--base-url",
+        internal_base_url(args.api_url),
+        "--project-name",
+        scope.project_name,
+        "--area-name",
+        scope.area_name,
+        "--years",
+        str(release.year),
+        "--population-url",
+        release.population_url,
+        "--geometry-url",
+        release.geometry_url,
+        "--population-layout",
+        release.layout,
+        "--scope-output-root",
+        str(args.scope_output_root),
+        "--output-dir",
+        str(population_output_dir(args)),
+        "--request-timeout",
+        str(args.request_timeout),
+        "--import-timeout",
+        str(args.import_timeout),
+    ]
+    if fetch_only:
+        command.extend(("--force", "--fetch-only"))
+    return command
+
+
+def run_operator(command: list[str], *, action: str) -> dict[str, Any]:
+    print(f"Statbel population: {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"Statbel population {action} failed with exit {completed.returncode}: {detail[-3000:]}")
+    try:
+        payload = json.loads(completed.stdout)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(f"Statbel population {action} returned invalid JSON") from exc
+    if payload.get("status") != "ok":
+        raise RuntimeError(f"Statbel population {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 validate_staged_release(
+    args: argparse.Namespace,
+    release: PopulationReleaseConfig,
+) -> dict[str, Any]:
+    scope = GEOGRAPHIC_SCOPES[args.scope]
+    output_dir = population_output_dir(args)
+    resolved_output = output_dir.resolve()
+    snapshot = snapshot_path(output_dir, scope, release.year)
+    manifest_path = preflight_manifest_path(output_dir, scope, release.year)
+    for path in (snapshot, manifest_path):
+        if not path.resolve().is_relative_to(resolved_output):
+            raise RuntimeError(f"Staged Statbel evidence is outside the governed output root: {path}")
+    manifest = load_preflight_manifest(manifest_path, snapshot, release.year, scope)
+    manifest_release = manifest.get("release") or {}
+    artifacts = manifest.get("artifacts") or {}
+    if (
+        manifest_release.get("population_layout") != release.layout
+        or (artifacts.get("population") or {}).get("source_url") != release.population_url
+        or (artifacts.get("geometry") or {}).get("source_url") != release.geometry_url
+    ):
+        raise RuntimeError("Staged Statbel manifest no longer matches the planned release identity")
+    for artifact_name in ("population", "geometry"):
+        retained = Path(str((artifacts.get(artifact_name) or {}).get("retained_path") or ""))
+        if not retained.resolve().is_relative_to(resolved_output):
+            raise RuntimeError(f"Staged {artifact_name} archive is outside the governed output root: {retained}")
+    return {
+        "manifest_path": str(manifest_path),
+        "manifest_sha256": sha256_path(manifest_path),
+        "snapshot_path": str(snapshot),
+        "snapshot_sha256": (artifacts.get("derived_snapshot") or {}).get("sha256"),
+        "snapshot_size_bytes": (artifacts.get("derived_snapshot") or {}).get("size_bytes"),
+        "feature_count": (artifacts.get("derived_snapshot") or {}).get("feature_count"),
+        "population_archive": artifacts.get("population"),
+        "geometry_archive": artifacts.get("geometry"),
+        "scope_accounting": manifest.get("scope_accounting"),
+        "national_accounting": manifest.get("national_accounting"),
+        "baseline": manifest.get("baseline"),
+        "geometry_repair_count": (manifest.get("schemas") or {}).get("geometry_repair_count"),
+    }
+
+
+def build_staged_plan(
+    args: argparse.Namespace,
+    decision: dict[str, Any],
+    release: PopulationReleaseConfig,
+    operator_result: dict[str, Any],
+) -> dict[str, Any]:
+    evidence = validate_staged_release(args, release)
+    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": evidence,
+        "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]) -> PopulationReleaseConfig:
+    release = payload.get("release") or {}
+    year = release.get("year") if isinstance(release, dict) else None
+    layout = release.get("layout") if isinstance(release, dict) else None
+    population_url = release.get("population_url") if isinstance(release, dict) else None
+    geometry_url = release.get("geometry_url") if isinstance(release, dict) else None
+    if (
+        not isinstance(year, int)
+        or isinstance(year, bool)
+        or not isinstance(layout, str)
+        or not isinstance(population_url, str)
+        or not isinstance(geometry_url, str)
+        or not all((layout, population_url, geometry_url))
+    ):
+        raise RuntimeError("Release evidence is missing required identity fields")
+    return resolve_release_config(
+        year,
+        layout=layout,
+        population_url=population_url,
+        geometry_url=geometry_url,
+    )
+
+
+def load_staged_plan(
+    args: argparse.Namespace,
+    release: PopulationReleaseConfig,
+) -> 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 Statbel 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 Statbel 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 Statbel plan identity is invalid")
+    current_evidence = validate_staged_release(args, release)
+    if current_evidence != payload.get("evidence"):
+        raise RuntimeError("Staged Statbel 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 Statbel 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_checksums",
+            "population_and_geometry_schemas",
+            "scope_and_national_accounting",
+            "unlocated_population_accounting",
+            "baseline_change_limit",
+            "bounded_geometry_repairs",
+            "immutable_dataset_apply",
+        ],
+    }
+    payload["review_sha256"] = canonical_sha256(payload, "review_sha256")
+    return payload
+
+
+def load_review_evidence(
+    args: argparse.Namespace,
+    release: PopulationReleaseConfig,
+    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 Statbel 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("Statbel 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("Statbel review evidence does not authorize this staged plan")
+    return path, payload
+
+
+def _snapshot_result(operator_result: dict[str, Any], year: int) -> dict[str, Any]:
+    matches = [item for item in operator_result.get("snapshots") or [] if int(item.get("year") or 0) == year]
+    if len(matches) != 1:
+        raise RuntimeError("Population 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) <= 0:
+            raise ValueError("All timeout 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"Statbel release is not safely stageable: {decision.get('status')}")
+            operator_result = run_operator(build_operator_command(args, release, fetch_only=True), action="staging")
+            staged_result = _snapshot_result(operator_result, release.year)
+            if staged_result.get("status") != "prepared" or staged_result.get("preflight_status") != "passed":
+                raise RuntimeError("Population staging did not produce passed preflight 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 = _snapshot_result(operator_result, release.year)
+        if applied_result.get("status") not in {"imported", "existing"} or not applied_result.get("dataset_id"):
+            raise RuntimeError("Population 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") != str(release.year):
+            raise RuntimeError("Applied population Dataset did not become the current local Statbel 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_mol_population_history.py

@@ -15,6 +15,7 @@ from __future__ import annotations
 
 import argparse
 import csv
+from dataclasses import dataclass
 from hashlib import sha256
 import io
 import json
@@ -36,7 +37,11 @@ from urllib3.util.retry import Retry
 
 from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope
 from statbel_population_preflight import (
+    MAX_GEOMETRY_ARCHIVE_BYTES,
+    MAX_POPULATION_ARCHIVE_BYTES,
     StatbelPreflightError,
+    validate_geometry_source_url,
+    validate_population_source_url,
     validate_statbel_release,
     write_manifest,
 )
@@ -67,6 +72,45 @@ POPULATION_URLS = {
 POPULATION_LAYOUTS = {year: ("new" if year == 2025 else "standard") for year in POPULATION_URLS}
 
 
+@dataclass(frozen=True)
+class PopulationReleaseConfig:
+    year: int
+    layout: str
+    population_url: str
+    geometry_url: str
+
+
+def resolve_release_config(
+    year: int,
+    *,
+    population_url: str | None = None,
+    geometry_url: str | None = None,
+    layout: str | None = None,
+) -> PopulationReleaseConfig:
+    overrides = (population_url, geometry_url, layout)
+    if any(value is not None for value in overrides):
+        if not all(value is not None for value in overrides):
+            raise ValueError("Population URL, geometry URL and population layout must be supplied together")
+        release = PopulationReleaseConfig(
+            year=year,
+            layout=str(layout),
+            population_url=str(population_url),
+            geometry_url=str(geometry_url),
+        )
+    elif year in POPULATION_URLS:
+        release = PopulationReleaseConfig(
+            year=year,
+            layout=POPULATION_LAYOUTS[year],
+            population_url=POPULATION_URLS[year],
+            geometry_url=SECTOR_URL.format(year=year),
+        )
+    else:
+        raise ValueError(f"Unsupported population year without an explicit governed release: {year}")
+    validate_population_source_url(release.population_url, release.year, release.layout)
+    validate_geometry_source_url(release.geometry_url, release.year)
+    return release
+
+
 def parse_args() -> argparse.Namespace:
     parser = argparse.ArgumentParser(description="Provision official annual Statbel population snapshots.")
     parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY)
@@ -74,6 +118,15 @@ def parse_args() -> argparse.Namespace:
     parser.add_argument("--project-name", default=None)
     parser.add_argument("--area-name", default=None, help="Case-insensitive fragment identifying the persisted Area.")
     parser.add_argument("--years", default="2021,2022,2023,2024,2025")
+    parser.add_argument(
+        "--population-url",
+        help="Exact official Statbel ZIP URL for one explicitly governed release; requires the other release flags",
+    )
+    parser.add_argument(
+        "--geometry-url",
+        help="Exact matching official statistical-sector GeoJSON ZIP URL for one explicitly governed release",
+    )
+    parser.add_argument("--population-layout", choices=("standard", "new"))
     parser.add_argument("--output-dir", type=Path, default=None)
     parser.add_argument("--boundary-path", type=Path, default=None)
     parser.add_argument(
@@ -107,6 +160,43 @@ def build_session() -> requests.Session:
     return session
 
 
+def download_archive(
+    session: requests.Session,
+    *,
+    url: str,
+    year: int,
+    layout: str | None,
+    max_bytes: int,
+    timeout: int,
+) -> bytes:
+    with session.get(url, timeout=timeout, stream=True) as response:
+        response.raise_for_status()
+        if layout is None:
+            validate_geometry_source_url(response.url, year)
+        else:
+            validate_population_source_url(response.url, year, layout)
+        content_length = response.headers.get("Content-Length")
+        if content_length:
+            try:
+                advertised_size = int(content_length)
+            except ValueError as exc:
+                raise RuntimeError(f"Official source returned an invalid Content-Length for {url}") from exc
+            if advertised_size <= 0 or advertised_size > max_bytes:
+                raise RuntimeError(f"Official source archive exceeds the {max_bytes}-byte download limit")
+        chunks: list[bytes] = []
+        size = 0
+        for chunk in response.iter_content(chunk_size=1024 * 1024):
+            if not chunk:
+                continue
+            size += len(chunk)
+            if size > max_bytes:
+                raise RuntimeError(f"Official source archive exceeds the {max_bytes}-byte download limit")
+            chunks.append(chunk)
+    if size <= 0:
+        raise RuntimeError(f"Official source archive is empty: {url}")
+    return b"".join(chunks)
+
+
 def response_data(response: requests.Response) -> Any:
     try:
         payload = response.json()
@@ -119,6 +209,29 @@ def response_data(response: requests.Response) -> Any:
     return payload["data"]
 
 
+def list_paginated_items(session: requests.Session, url: str, *, timeout: int) -> list[dict[str, Any]]:
+    items: list[dict[str, Any]] = []
+    offset = 0
+    total: int | None = None
+    while total is None or offset < total:
+        page = response_data(session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout))
+        page_items = page.get("items") if isinstance(page, dict) else None
+        if not isinstance(page_items, list):
+            raise RuntimeError(f"GeoIntel list response for {url} has no items array")
+        page_total = int(page.get("total", len(page_items)))
+        if total is None:
+            total = page_total
+        elif page_total != total:
+            raise RuntimeError("GeoIntel pagination total changed while reading the population workspace")
+        items.extend(page_items)
+        if not page_items:
+            break
+        offset += len(page_items)
+    if total is not None and len(items) != total:
+        raise RuntimeError(f"GeoIntel list response for {url} returned {len(items)} of {total} items")
+    return items
+
+
 def series_key(scope: GeographicScope) -> str:
     return f"statbel:population-statistical-sector:{scope.key}"
 
@@ -340,9 +453,13 @@ def preflight_manifest_path(output_dir: Path, scope: GeographicScope, year: int)
 
 
 def previous_snapshot_path(output_dir: Path, scope: GeographicScope, year: int) -> Path | None:
-    candidates = [snapshot_path(output_dir, scope, candidate) for candidate in POPULATION_URLS if candidate < year]
+    prefix = f"{scope.key.replace('-', '_')}_statbel_population_"
-    available = [path for path in candidates if path.is_file()]
+    available: list[tuple[int, Path]] = []
-    return max(available, key=lambda path: int(path.stem.rsplit("_", 1)[-1])) if available else None
+    for path in output_dir.glob(f"{prefix}*.geojson"):
+        suffix = path.stem.removeprefix(prefix)
+        if suffix.isdigit() and int(suffix) < year and path.is_file():
+            available.append((int(suffix), path))
+    return max(available, key=lambda item: item[0])[1] if available else None
 
 
 def load_preflight_manifest(path: Path, snapshot: Path, year: int, scope: GeographicScope) -> dict[str, Any]:
@@ -383,23 +500,24 @@ def stage_release(
     output_dir: Path,
     boundary,
     scope: GeographicScope,
+    release: PopulationReleaseConfig | None = None,
 ) -> tuple[Path, Path, dict[str, Any]]:
-    layout = POPULATION_LAYOUTS[year]
+    active_release = release or resolve_release_config(year)
-    population_url = POPULATION_URLS[year]
+    if active_release.year != year:
-    geometry_url = SECTOR_URL.format(year=year)
+        raise ValueError("Population release year does not match the staged year")
     result = validate_statbel_release(
         year=year,
-        layout=layout,
+        layout=active_release.layout,
         population_content=population_content,
-        population_url=population_url,
+        population_url=active_release.population_url,
         geometry_content=geometry_content,
-        geometry_url=geometry_url,
+        geometry_url=active_release.geometry_url,
         scope=scope,
         baseline_snapshot=previous_snapshot_path(output_dir, scope, year),
     )
     raw_dir = output_dir / "raw" / str(year)
-    population_archive_path = raw_dir / Path(unquote(urlsplit(population_url).path)).name
+    population_archive_path = raw_dir / Path(unquote(urlsplit(active_release.population_url).path)).name
-    geometry_archive_path = raw_dir / Path(unquote(urlsplit(geometry_url).path)).name
+    geometry_archive_path = raw_dir / Path(unquote(urlsplit(active_release.geometry_url).path)).name
     write_bytes_atomic(population_archive_path, population_content)
     write_bytes_atomic(geometry_archive_path, geometry_content)
 
@@ -444,18 +562,18 @@ def locate_workspace(
     area_name: str,
     timeout: int,
 ):
-    projects = response_data(session.get(f"{base_url}/api/v1/projects", params={"limit": 200}, timeout=timeout))
+    projects = list_paginated_items(session, f"{base_url}/api/v1/projects", timeout=timeout)
-    project = next((item for item in projects.get("items") or [] if item.get("name") == project_name), None)
+    project = next((item for item in projects if item.get("name") == project_name), None)
     if not project:
         raise RuntimeError(f"Project {project_name!r} is missing")
     project_id = str(project["id"])
-    areas = response_data(session.get(f"{base_url}/api/v1/projects/{project_id}/areas", params={"limit": 200}, timeout=timeout))
+    areas = list_paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/areas", timeout=timeout)
     area_fragment = area_name.strip().casefold()
-    matches = [item for item in areas.get("items") or [] if area_fragment in str(item.get("name") or "").casefold()]
+    matches = [item for item in areas if area_fragment in str(item.get("name") or "").casefold()]
     if len(matches) != 1:
         raise RuntimeError(f"Expected one official Area matching {area_name!r}, received {len(matches)}")
-    datasets = response_data(session.get(f"{base_url}/api/v1/projects/{project_id}/datasets", params={"limit": 200}, timeout=timeout))
+    datasets = list_paginated_items(session, f"{base_url}/api/v1/projects/{project_id}/datasets", timeout=timeout)
-    return project_id, str(matches[0]["id"]), list(datasets.get("items") or [])
+    return project_id, str(matches[0]["id"]), datasets
 
 
 def upload_snapshot(
@@ -468,7 +586,11 @@ def upload_snapshot(
     timeout: int,
     scope: GeographicScope,
     preflight_path: Path,
+    release: PopulationReleaseConfig | None = None,
 ) -> dict[str, Any]:
+    active_release = release or resolve_release_config(year)
+    if active_release.year != year:
+        raise ValueError("Population release year does not match the upload year")
     observed_at = f"{year}-01-01T00:00:00Z"
     preflight = load_preflight_manifest(preflight_path, path, year, scope)
     accounting = preflight["scope_accounting"]
@@ -506,8 +628,8 @@ def upload_snapshot(
         "operator_explicit_fetch": True,
         "scope_key": scope.key,
         "geometry_clipped_to_area": True,
-        "sector_geometry_url": SECTOR_URL.format(year=year),
+        "sector_geometry_url": active_release.geometry_url,
-        "population_url": POPULATION_URLS[year],
+        "population_url": active_release.population_url,
         "population_layout": preflight["release"]["population_layout"],
         "preflight_manifest_path": str(preflight_path),
         "preflight_manifest_sha256": sha256_path(preflight_path),
@@ -551,9 +673,23 @@ def main() -> int:
     except ValueError:
         print(json.dumps({"status": "error", "message": "Years must be comma-separated integers"}), file=sys.stderr)
         return 2
-    unsupported = [year for year in years if year not in POPULATION_URLS]
+    custom_release_requested = any((args.population_url, args.geometry_url, args.population_layout))
-    if unsupported or not years:
+    if not years or (custom_release_requested and len(years) != 1):
-        print(json.dumps({"status": "error", "message": f"Unsupported years: {unsupported}"}), file=sys.stderr)
+        message = "An explicit governed release requires exactly one population year" if custom_release_requested else "No years requested"
+        print(json.dumps({"status": "error", "message": message}), file=sys.stderr)
+        return 2
+    try:
+        releases = {
+            year: resolve_release_config(
+                year,
+                population_url=args.population_url if custom_release_requested else None,
+                geometry_url=args.geometry_url if custom_release_requested else None,
+                layout=args.population_layout if custom_release_requested else None,
+            )
+            for year in years
+        }
+    except (ValueError, StatbelPreflightError) as exc:
+        print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr)
         return 2
 
     output_dir = resolve_output_dir(args, scope)
@@ -565,21 +701,35 @@ def main() -> int:
         prepared: list[dict[str, Any]] = []
         with build_session() as source_session:
             for year in years:
+                release = releases[year]
                 path = snapshot_path(output_dir, scope, year)
                 manifest_path = preflight_manifest_path(output_dir, scope, year)
                 preflight_status = "passed"
                 if args.force or not path.exists():
-                    sectors_response = source_session.get(SECTOR_URL.format(year=year), timeout=args.request_timeout)
+                    geometry_content = download_archive(
-                    sectors_response.raise_for_status()
+                        source_session,
-                    population_response = source_session.get(POPULATION_URLS[year], timeout=args.request_timeout)
+                        url=release.geometry_url,
-                    population_response.raise_for_status()
+                        year=year,
+                        layout=None,
+                        max_bytes=MAX_GEOMETRY_ARCHIVE_BYTES,
+                        timeout=args.request_timeout,
+                    )
+                    population_content = download_archive(
+                        source_session,
+                        url=release.population_url,
+                        year=year,
+                        layout=release.layout,
+                        max_bytes=MAX_POPULATION_ARCHIVE_BYTES,
+                        timeout=args.request_timeout,
+                    )
                     path, manifest_path, _manifest = stage_release(
                         year=year,
-                        population_content=population_response.content,
+                        population_content=population_content,
-                        geometry_content=sectors_response.content,
+                        geometry_content=geometry_content,
                         output_dir=output_dir,
                         boundary=boundary,
                         scope=scope,
+                        release=release,
                     )
                 elif manifest_path.is_file():
                     load_preflight_manifest(manifest_path, path, year, scope)
@@ -648,6 +798,7 @@ def main() -> int:
                         args.import_timeout,
                         scope,
                         item["manifest_path"],
+                        releases[year],
                     )
                     results.append({"year": year, "dataset_id": dataset["id"], "feature_count": dataset.get("feature_count"), "status": "imported"})
     except (OSError, RuntimeError, requests.RequestException, ValueError, KeyError, zipfile.BadZipFile) as exc:

scripts/run_readiness_check.sh

@@ -46,6 +46,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_mol_municipality_workspace.py
 ${PYTHON_BIN} -m py_compile scripts/provision_mol_context_layers.py
 ${PYTHON_BIN} -m py_compile scripts/provision_mol_population_history.py
 ${PYTHON_BIN} -m py_compile scripts/statbel_population_preflight.py
+${PYTHON_BIN} -m py_compile scripts/manage_statbel_population_release.py
 ${PYTHON_BIN} -m py_compile scripts/provision_mol_historical_landuse.py
 ${PYTHON_BIN} -m py_compile scripts/provision_regional_historical_landuse.py
 ${PYTHON_BIN} -m py_compile scripts/provision_official_landuse_timeseries.py