Add governed GRB refresh workflow

CHANGELOG.md

@@ -7,6 +7,20 @@
 
 # Changelog
 
+## Sprint 223 Governed regional GRB refresh (2026-07-16)
+
+- Added a canonical, read-only regional GRB refresh plan for buildings, roads,
+  water and parcels, derived from the official catalog edition and persisted
+  temporal snapshots.
+- Added an explicit operator coordinator with separate `plan`, `stage` and
+  SHA-256-confirmed `apply` phases. Staging validates every municipality
+  partition before PostGIS persistence is possible.
+- Kept all provider URLs and collections allowlisted, reused the existing
+  resumable regional operators and DatasetService/VectorFeatureService import
+  path, and retained every older snapshot.
+- Added compact Status UI, deterministic guardrail tests and readiness syntax
+  coverage without changing migrations or enabling automatic refresh.
+
 ## Sprint 222 Official source edition probes (2026-07-16)
 
 - Added explicit, read-only GRB and most-recent orthophoto catalog probes using

backend/README.md

@@ -1469,3 +1469,41 @@ Runtime controls are `SOURCE_CATALOG_PROBE_ENABLED`,
 `SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS`. Metadata links remain restricted to
 the official HTTPS CSW path on `metadata.vlaanderen.be` even when an operator
 overrides the capabilities endpoint.
+
+## Governed regional GRB refresh
+
+`GET /api/v1/projects/{project_id}/datasets/grb-refresh-plan` combines the
+explicit official GRB edition probe with the four existing regional snapshot
+series. It is read-only: it reports local feature/storage impact and whether
+buildings, roads, water and parcels are current, updateable or require review.
+
+Regional refreshes use a two-phase operator flow inside the all-in-one
+container. First stage and validate every source partition without touching
+PostGIS:
+
+```bash
+docker exec geointel python /app/scripts/manage_grb_refresh.py stage \
+  --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --confirm-edition 2026-07-15 \
+  --layers buildings roads water parcels
+```
+
+The JSON result gives `plan_path`, exact feature deltas, artifact sizes and
+`plan_sha256`. Review that evidence, then apply those exact staged bytes:
+
+```bash
+docker exec geointel python /app/scripts/manage_grb_refresh.py apply \
+  --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --confirm-edition 2026-07-15 \
+  --confirm-plan-sha256 SHA256_FROM_STAGE
+```
+
+Both commands fail when project/scope, official edition, manifests, partition
+count or any checksum differs. Interrupted staging is safely resumable because
+the existing municipal manifests are reused. Apply imports through
+DatasetService/VectorFeatureService, creates new temporal Datasets and retains
+all previous snapshots. Do not add `--force` to this coordinator; a source
+refetch remains a separate deliberate recovery action in the lower-level
+operators.

backend/app/api/routes/datasets.py

@@ -46,6 +46,7 @@ from app.services.vector_feature_service import VectorFeatureService
 from app.services.dataset_service import DatasetService
 from app.services.source_freshness_service import SourceFreshnessService
 from app.services.source_catalog_probe_service import SourceCatalogProbeService
+from app.services.grb_refresh_plan_service import GrbRefreshPlanService
 from app.services.orthophoto_acquisition_service import OrthophotoAcquisitionService
 from app.services.dhmv_acquisition_service import DhmvAcquisitionService
 from app.services.terrain_analysis_service import TerrainAnalysisService
@@ -266,6 +267,22 @@ def probe_dataset_source_catalogs(
     return envelope(report.model_dump())
 
 
+@router.get("/datasets/grb-refresh-plan", response_model=dict)
+def plan_grb_dataset_refresh(
+    project_id: UUID,
+    scope: str = Query(default=GrbRefreshPlanService.SCOPE),
+    refresh_catalog: bool = Query(default=False),
+    db: Session = Depends(get_db),
+):
+    report = GrbRefreshPlanService.build(
+        db,
+        project_id,
+        scope=scope,
+        refresh_catalog=refresh_catalog,
+    )
+    return envelope(report.model_dump())
+
+
 @router.get("/datasets/{dataset_id}", response_model=dict)
 def get_dataset(
     project_id: UUID,

backend/app/schemas/__init__.py

@@ -16,6 +16,7 @@ from .source_catalog import (
     SourceCatalogProbeReport,
     SourceCatalogProbeSummary,
 )
+from .grb_refresh import GrbRefreshLayerPlan, GrbRefreshPlan, GrbRefreshPlanSummary
 from .detection import (
     DetectionListResponse,
     DetectionModelCapability,
@@ -149,6 +150,9 @@ __all__ = [
     "SourceCatalogProbeItem",
     "SourceCatalogProbeReport",
     "SourceCatalogProbeSummary",
+    "GrbRefreshLayerPlan",
+    "GrbRefreshPlan",
+    "GrbRefreshPlanSummary",
     "DetectionListResponse",
     "DetectionModelCapability",
     "DetectionModelsResponse",

backend/app/schemas/grb_refresh.py

@@ -0,0 +1,63 @@
+from __future__ import annotations
+
+from datetime import date, datetime
+from typing import Literal
+from uuid import UUID
+
+from pydantic import BaseModel
+
+
+GrbRefreshLayerStatus = Literal[
+    "current",
+    "update_available",
+    "not_loaded",
+    "review_required",
+    "remote_unavailable",
+]
+
+
+class GrbRefreshLayerPlan(BaseModel):
+    theme: Literal["buildings", "roads", "water", "parcels"]
+    display_name: str
+    collections: list[str]
+    temporal_series_key: str
+    status: GrbRefreshLayerStatus
+    local_dataset_id: UUID | None = None
+    local_source_version: str | None = None
+    local_observed_at: datetime | None = None
+    local_imported_at: datetime | None = None
+    local_feature_count: int | None = None
+    local_size_bytes: int | None = None
+    retained_after_refresh: bool = True
+    action_message: str
+
+
+class GrbRefreshPlanSummary(BaseModel):
+    layer_count: int
+    current_count: int
+    update_available_count: int
+    not_loaded_count: int
+    review_required_count: int
+    remote_unavailable_count: int
+    new_dataset_count_if_applied: int
+    retained_dataset_count: int
+    current_feature_count: int
+    current_size_bytes: int
+
+
+class GrbRefreshPlan(BaseModel):
+    project_id: UUID
+    scope: str
+    generated_at: datetime
+    remote_status: str
+    remote_version: str | None = None
+    remote_edition_date: date | None = None
+    catalog_checked_at: datetime | None = None
+    summary: GrbRefreshPlanSummary
+    layers: list[GrbRefreshLayerPlan]
+    execution_mode: Literal["operator_stage_then_apply"] = "operator_stage_then_apply"
+    staging_required: bool = True
+    automatic_import: bool = False
+    destructive_replacement: bool = False
+    message: str
+    limitations: list[str]

backend/app/services/grb_refresh_plan_service.py

@@ -0,0 +1,186 @@
+from __future__ import annotations
+
+import re
+from dataclasses import dataclass
+from datetime import date, datetime, timezone
+from uuid import UUID
+
+from sqlalchemy.orm import Session
+
+from app.core.errors import AppError
+from app.models import Dataset, Project
+from app.schemas.grb_refresh import GrbRefreshLayerPlan, GrbRefreshPlan, GrbRefreshPlanSummary
+from app.services.source_catalog_probe_service import SourceCatalogProbeService
+
+
+@dataclass(frozen=True)
+class _LayerDefinition:
+    theme: str
+    display_name: str
+    collections: tuple[str, ...]
+
+    @property
+    def series_key(self) -> str:
+        return f"grb:{self.theme}:kempen-transport-region"
+
+
+class GrbRefreshPlanService:
+    SCOPE = "kempen-transport-region"
+    LAYERS = (
+        _LayerDefinition("buildings", "Gebouwen", ("GBG",)),
+        _LayerDefinition("roads", "Wegen", ("Wegsegment",)),
+        _LayerDefinition("water", "Water", ("WTZ", "WLAS", "WGR")),
+        _LayerDefinition("parcels", "Percelen", ("ADP",)),
+    )
+    _EDITION_DATE = re.compile(r"(?<!\d)(20\d{2}-\d{2}-\d{2})(?!\d)")
+
+    @classmethod
+    def _parse_edition_date(cls, value: str | None) -> date | None:
+        match = cls._EDITION_DATE.search(value or "")
+        if not match:
+            return None
+        try:
+            return date.fromisoformat(match.group(1))
+        except ValueError:
+            return None
+
+    @staticmethod
+    def _dataset_feature_count(dataset: Dataset) -> int | None:
+        metadata = dataset.metadata_json if isinstance(dataset.metadata_json, dict) else {}
+        value = metadata.get("feature_count")
+        try:
+            return int(value) if value is not None else None
+        except (TypeError, ValueError):
+            return None
+
+    @staticmethod
+    def _latest_dataset(rows: list[Dataset], series_key: str) -> Dataset | None:
+        candidates = [
+            row
+            for row in rows
+            if row.temporal_series_key == series_key and row.status == "ready"
+        ]
+        if not candidates:
+            return None
+        minimum = datetime.min.replace(tzinfo=timezone.utc)
+        return max(candidates, key=lambda row: (row.observed_at or row.imported_at or row.created_at or minimum, str(row.id)))
+
+    @classmethod
+    def build(
+        cls,
+        db: Session,
+        project_id: UUID,
+        *,
+        scope: str = SCOPE,
+        refresh_catalog: bool = False,
+        now: datetime | None = None,
+    ) -> GrbRefreshPlan:
+        if scope != cls.SCOPE:
+            raise AppError(
+                code="GRB_REFRESH_SCOPE_UNSUPPORTED",
+                message=f"Only the governed scope '{cls.SCOPE}' is supported",
+                status_code=400,
+            )
+        if not db.get(Project, project_id):
+            raise AppError(code="PROJECT_NOT_FOUND", message="Project not found", status_code=404)
+
+        generated_at = now or datetime.now(timezone.utc)
+        catalog = SourceCatalogProbeService.audit_project(db, project_id, force=refresh_catalog)
+        grb_probe = next((item for item in catalog.items if item.source_name == "grb"), None)
+        remote_version = grb_probe.remote_version if grb_probe else None
+        remote_edition_date = cls._parse_edition_date(remote_version)
+        remote_available = bool(grb_probe and grb_probe.status == "available" and grb_probe.reachable)
+        rows = (
+            db.query(Dataset)
+            .filter(Dataset.project_id == project_id, Dataset.source_name == "grb")
+            .all()
+        )
+
+        layer_plans: list[GrbRefreshLayerPlan] = []
+        for definition in cls.LAYERS:
+            local = cls._latest_dataset(rows, definition.series_key)
+            local_date = cls._parse_edition_date(local.source_version if local else None)
+            if not remote_available:
+                status = "remote_unavailable"
+                action = "De officiële catalogus is niet bereikbaar; er wordt geen vernieuwingsbeslissing genomen."
+            elif remote_edition_date is None:
+                status = "review_required"
+                action = "De officiële editie bevat geen herkenbare datum en vereist menselijke beoordeling."
+            elif local is None:
+                status = "not_loaded"
+                action = "Deze laag kan als nieuwe, afzonderlijke GRB-snapshot worden voorbereid."
+            elif local_date is None:
+                status = "review_required"
+                action = "De lokale editie is niet veilig datumvergelijkbaar; controleer de provenance vóór staging."
+            elif local_date == remote_edition_date:
+                status = "current"
+                action = "De lokale snapshot gebruikt dezelfde officiële editie; geen import nodig."
+            elif local_date < remote_edition_date:
+                status = "update_available"
+                action = "Stage eerst alle bronartifacts en controleer exacte aantallen en checksums vóór import."
+            else:
+                status = "review_required"
+                action = "De lokale editie lijkt nieuwer dan de catalogus; automatische terugval is verboden."
+
+            layer_plans.append(
+                GrbRefreshLayerPlan(
+                    theme=definition.theme,
+                    display_name=definition.display_name,
+                    collections=list(definition.collections),
+                    temporal_series_key=definition.series_key,
+                    status=status,
+                    local_dataset_id=local.id if local else None,
+                    local_source_version=local.source_version if local else None,
+                    local_observed_at=local.observed_at if local else None,
+                    local_imported_at=local.imported_at if local else None,
+                    local_feature_count=cls._dataset_feature_count(local) if local else None,
+                    local_size_bytes=local.size_bytes if local else None,
+                    action_message=action,
+                )
+            )
+
+        counts = {status: sum(item.status == status for item in layer_plans) for status in (
+            "current", "update_available", "not_loaded", "review_required", "remote_unavailable"
+        )}
+        actionable = counts["update_available"] + counts["not_loaded"]
+        summary = GrbRefreshPlanSummary(
+            layer_count=len(layer_plans),
+            current_count=counts["current"],
+            update_available_count=counts["update_available"],
+            not_loaded_count=counts["not_loaded"],
+            review_required_count=counts["review_required"],
+            remote_unavailable_count=counts["remote_unavailable"],
+            new_dataset_count_if_applied=actionable,
+            retained_dataset_count=sum(item.local_dataset_id is not None for item in layer_plans),
+            current_feature_count=sum(item.local_feature_count or 0 for item in layer_plans),
+            current_size_bytes=sum(item.local_size_bytes or 0 for item in layer_plans),
+        )
+        if actionable:
+            message = (
+                f"{actionable} GRB-laag{' is' if actionable == 1 else 'en zijn'} voorbereidbaar voor editie "
+                f"{remote_edition_date.isoformat() if remote_edition_date else remote_version}. "
+                "Staging berekent eerst de exacte impact; import vereist daarna de plan-checksum."
+            )
+        elif counts["current"] == len(layer_plans):
+            message = "Alle beheerde regionale GRB-lagen gebruiken de officiële cataloguseditie."
+        else:
+            message = "Er is menselijke beoordeling nodig voordat een GRB-staging kan starten."
+
+        return GrbRefreshPlan(
+            project_id=project_id,
+            scope=scope,
+            generated_at=generated_at,
+            remote_status=grb_probe.status if grb_probe else "unavailable",
+            remote_version=remote_version,
+            remote_edition_date=remote_edition_date,
+            catalog_checked_at=grb_probe.checked_at if grb_probe else None,
+            summary=summary,
+            layers=layer_plans,
+            message=message,
+            limitations=[
+                "Dit endpoint is read-only en start geen download, import of databasejob.",
+                "Staging bewaart bronartifacts buiten PostGIS; apply vereist de exacte staged plan-checksum.",
+                "Een refresh maakt nieuwe immutable Datasets en verwijdert of overschrijft oude snapshots niet.",
+                "Exacte feature- en opslagverschillen zijn pas bekend nadat alle regionale partitions staged en gevalideerd zijn.",
+            ],
+        )

backend/tests/test_sprint223_governed_grb_refresh.py

@@ -0,0 +1,250 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+from types import SimpleNamespace
+import uuid
+
+import pytest
+
+from app.core.errors import AppError
+from app.models import Dataset
+from app.services.grb_refresh_plan_service import GrbRefreshPlanService
+
+
+NOW = datetime(2026, 7, 16, 16, 0, tzinfo=timezone.utc)
+PROJECT_ID = uuid.uuid4()
+
+
+class _Query:
+    def __init__(self, rows: list[Dataset]) -> None:
+        self.rows = rows
+
+    def filter(self, *_args):
+        return self
+
+    def all(self) -> list[Dataset]:
+        return self.rows
+
+
+class _Db:
+    def __init__(self, rows: list[Dataset]) -> None:
+        self.rows = rows
+
+    def get(self, _model, identifier):
+        return SimpleNamespace(id=identifier)
+
+    def query(self, _model):
+        return _Query(self.rows)
+
+
+def _dataset(theme: str, version: str = "2026-07-14", count: int = 100) -> Dataset:
+    return Dataset(
+        id=uuid.uuid4(),
+        project_id=PROJECT_ID,
+        name=f"grb_{theme}.geojson",
+        dataset_type="vector",
+        source="operator_official_import",
+        source_name="grb",
+        dataset_role="reference",
+        reference_layer_name=theme,
+        source_version=version,
+        temporal_series_key=f"grb:{theme}:kempen-transport-region",
+        observed_at=datetime.fromisoformat(f"{version}T00:00:00+00:00"),
+        imported_at=NOW,
+        metadata_json={"feature_count": count},
+        size_bytes=1000,
+        status="ready",
+    )
+
+
+def _catalog(status: str = "available", version: str | None = "Toestand 2026-07-15"):
+    return SimpleNamespace(
+        items=[
+            SimpleNamespace(
+                source_name="grb",
+                status=status,
+                reachable=status == "available",
+                remote_version=version,
+                checked_at=NOW,
+            )
+        ]
+    )
+
+
+def test_refresh_plan_marks_all_older_immutable_snapshots_as_update_available(monkeypatch) -> None:
+    rows = [_dataset(theme, count=(index + 1) * 100) for index, theme in enumerate(("buildings", "roads", "water", "parcels"))]
+    monkeypatch.setattr(
+        "app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project",
+        lambda *_args, **_kwargs: _catalog(),
+    )
+
+    plan = GrbRefreshPlanService.build(_Db(rows), PROJECT_ID, now=NOW)
+
+    assert plan.remote_edition_date.isoformat() == "2026-07-15"
+    assert plan.summary.update_available_count == 4
+    assert plan.summary.new_dataset_count_if_applied == 4
+    assert plan.summary.retained_dataset_count == 4
+    assert plan.summary.current_feature_count == 1000
+    assert plan.summary.current_size_bytes == 4000
+    assert all(item.status == "update_available" for item in plan.layers)
+    assert all(item.retained_after_refresh for item in plan.layers)
+    assert plan.automatic_import is False
+    assert plan.destructive_replacement is False
+
+
+@pytest.mark.parametrize(
+    ("rows", "catalog", "expected"),
+    [
+        ([_dataset("buildings", "2026-07-15")], _catalog(), "current"),
+        ([], _catalog(), "not_loaded"),
+        ([_dataset("buildings")], _catalog("unavailable"), "remote_unavailable"),
+        ([_dataset("buildings")], _catalog("available", "Onbekende toestand"), "review_required"),
+    ],
+)
+def test_refresh_plan_status_matrix(monkeypatch, rows, catalog, expected) -> None:
+    monkeypatch.setattr(
+        "app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project",
+        lambda *_args, **_kwargs: catalog,
+    )
+    plan = GrbRefreshPlanService.build(_Db(rows), PROJECT_ID, now=NOW)
+    buildings = next(item for item in plan.layers if item.theme == "buildings")
+    assert buildings.status == expected
+
+
+def test_refresh_plan_rejects_an_unapproved_scope(monkeypatch) -> None:
+    monkeypatch.setattr(
+        "app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project",
+        lambda *_args, **_kwargs: _catalog(),
+    )
+    with pytest.raises(AppError) as raised:
+        GrbRefreshPlanService.build(_Db([]), PROJECT_ID, scope="mol", now=NOW)
+    assert raised.value.code == "GRB_REFRESH_SCOPE_UNSUPPORTED"
+
+
+def test_refresh_plan_route_uses_the_canonical_envelope(monkeypatch) -> None:
+    from app.api.routes import datasets as dataset_routes
+
+    monkeypatch.setattr(
+        "app.services.grb_refresh_plan_service.SourceCatalogProbeService.audit_project",
+        lambda *_args, **_kwargs: _catalog(),
+    )
+    response = dataset_routes.plan_grb_dataset_refresh(
+        project_id=PROJECT_ID,
+        scope="kempen-transport-region",
+        refresh_catalog=False,
+        db=_Db([]),
+    )
+    assert list(response) == ["data"]
+    assert response["data"]["project_id"] == PROJECT_ID
+    assert response["data"]["automatic_import"] is False
+
+
+def _load_operator_module():
+    root = Path(__file__).resolve().parents[2]
+    scripts = root / "scripts"
+    if str(scripts) not in sys.path:
+        sys.path.insert(0, str(scripts))
+    spec = importlib.util.spec_from_file_location("manage_grb_refresh_s223", scripts / "manage_grb_refresh.py")
+    assert spec and spec.loader
+    module = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(module)
+    return module
+
+
+def test_operator_requires_exact_edition_and_plan_hash() -> None:
+    module = _load_operator_module()
+    assert module.require_confirmed_edition({"remote_edition_date": "2026-07-15"}, "2026-07-15") == "2026-07-15"
+    with pytest.raises(RuntimeError, match="confirm-edition"):
+        module.require_confirmed_edition({"remote_edition_date": "2026-07-15"}, "2026-07-14")
+
+    payload = {"status": "staged", "layers": []}
+    first = module.canonical_plan_sha256(payload)
+    assert first == module.canonical_plan_sha256({**payload, "plan_sha256": first})
+    assert first != module.canonical_plan_sha256({"status": "staged", "layers": [{"theme": "roads"}]})
+
+
+def test_operator_validates_every_staged_artifact_and_partition(tmp_path) -> None:
+    module = _load_operator_module()
+    manifest_dir = tmp_path / "kempen-transport-region" / "buildings" / "2026-07-15"
+    partition_dir = manifest_dir / "partitions"
+    partition_dir.mkdir(parents=True)
+    artifact = manifest_dir / "buildings.geojson"
+    partition = partition_dir / "13025_mol.geojson"
+    artifact.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
+    partition.write_text('{"type":"FeatureCollection","features":[]}', encoding="utf-8")
+    manifest = {
+        "status": "complete",
+        "scope": "kempen-transport-region",
+        "theme": "buildings",
+        "observed_at": "2026-07-15",
+        "reference_truncated": False,
+        "member_count": 1,
+        "feature_count": 1,
+        "artifact_filename": artifact.name,
+        "artifact_sha256": module.sha256_file(artifact),
+        "artifact_size_bytes": artifact.stat().st_size,
+        "partitions": [{"filename": partition.name, "sha256": module.sha256_file(partition)}],
+    }
+    path = manifest_dir / "regional_buildings_manifest.json"
+    path.write_text(json.dumps(manifest), encoding="utf-8")
+
+    assert module.validate_manifest(
+        path,
+        output_root=tmp_path,
+        scope="kempen-transport-region",
+        theme="buildings",
+        edition="2026-07-15",
+    )["feature_count"] == 1
+
+    partition.write_text("changed", encoding="utf-8")
+    with pytest.raises(RuntimeError, match="partition checksum"):
+        module.validate_manifest(
+            path,
+            output_root=tmp_path,
+            scope="kempen-transport-region",
+            theme="buildings",
+            edition="2026-07-15",
+        )
+
+
+def test_operator_builds_only_allowlisted_local_subprocess_commands(tmp_path) -> None:
+    module = _load_operator_module()
+    args = SimpleNamespace(
+        scope="kempen-transport-region",
+        api_url="http://127.0.0.1:8000/api/v1",
+        output_root=tmp_path,
+        request_timeout=180,
+        api_timeout=180,
+        batch_size=1000,
+        page_limit=1000,
+        max_features_per_member=100000,
+        max_total_features=1500000,
+    )
+    commands = module.build_operator_commands(
+        args,
+        ["buildings", "roads", "water", "parcels"],
+        "2026-07-15",
+        fetch_only=True,
+    )
+    flattened = [argument for _label, command in commands for argument in command]
+    assert len(commands) == 2
+    assert "--fetch-only" in flattened
+    assert "provision_regional_grb_buildings.py" in " ".join(flattened)
+    assert "provision_regional_grb_context.py" in " ".join(flattened)
+    assert "--force" not in flattened
+
+
+def test_refresh_api_and_frontend_remain_explicit_only() -> None:
+    root = Path(__file__).resolve().parents[2]
+    service = (root / "backend" / "app" / "services" / "grb_refresh_plan_service.py").read_text(encoding="utf-8")
+    hook = (root / "frontend" / "src" / "hooks" / "useSourceFreshness.ts").read_text(encoding="utf-8")
+    operator = (root / "scripts" / "manage_grb_refresh.py").read_text(encoding="utf-8")
+    assert "DatasetService" not in service
+    assert "VectorFeatureService" not in service
+    assert "void probeCatalogs(" not in hook
+    assert 'choices=("plan", "stage", "apply")' in operator
+    assert "--confirm-plan-sha256" in operator

docs/API_CONTRACTS.md

@@ -432,6 +432,30 @@ not fetch vector features, raster pixels or models, create jobs/datasets, write
 to PostGIS or trigger an import. The normal `source-freshness` endpoint remains
 local-only and never invokes this probe implicitly.
 
+### GET `/api/v1/projects/{project_id}/datasets/grb-refresh-plan`
+
+Builds a read-only refresh decision for the governed
+`kempen-transport-region` GRB snapshot series. Query parameter
+`refresh_catalog=true` explicitly bypasses the catalog cache. The response
+maps the official dated edition to `buildings`, `roads`, `water` and `parcels`
+and reports each latest local Dataset, source version, observation date,
+feature count, artifact size and refresh state.
+
+Layer state is `current`, `update_available`, `not_loaded`, `review_required`
+or `remote_unavailable`. The summary reports how many new immutable Datasets
+would be created and how many existing snapshots remain retained. The endpoint
+does not fetch GRB features, stage files, create a Job, write to PostGIS or
+start an operator process. Exact remote deltas remain unavailable until every
+municipality partition has been staged and validated.
+
+Regional execution uses `scripts/manage_grb_refresh.py` outside the request
+cycle. `stage` requires the exact official ISO edition, invokes the existing
+regional GRB operators with `--fetch-only`, validates all artifact/partition
+checksums and emits a SHA-256-bound plan. `apply` requires that exact plan hash,
+revalidates every staged byte and delegates persistence to DatasetService and
+VectorFeatureService. It creates new temporal snapshots and never deletes or
+overwrites an older Dataset.
+
 ### GET `/api/v1/projects/{project_id}/datasets/{dataset_id}`
 
 Return metadata.

docs/CODEX_EXECUTION_LOG.md

@@ -9339,3 +9339,36 @@ Next:
 - Extend the same probe pattern only where another official source publishes a
   stable machine-readable edition. Do not infer release versions from service
   protocol versions, ETags or HTTP modification dates.
+
+## Sprint 223 - Governed regional GRB refresh (2026-07-16)
+
+Implemented:
+- Added a canonical read-only refresh plan that maps the official dated GRB
+  edition onto the immutable regional buildings, roads, water and parcel
+  temporal series.
+- Added explicit per-layer current/update/review states, existing Dataset,
+  feature/storage impact and non-destructive action guidance.
+- Added `manage_grb_refresh.py` with separate plan, fetch-only stage and
+  checksum-confirmed apply phases. Stage validates all retained artifacts and
+  municipality partitions; apply revalidates the exact staged plan before
+  calling the existing regional DatasetService-based operators.
+- Added a compact Status surface after the explicit official catalog check.
+
+Boundaries:
+- No migration, scheduler, arbitrary provider URL, browser-triggered heavy
+  import, direct vector-feature write or automatic Dataset replacement was
+  introduced.
+- Existing GRB snapshots remain immutable historical observations. Exact
+  remote deltas are reported only after full staging, never guessed from
+  capabilities metadata.
+
+Validation so far:
+- Backend compile and frontend typecheck passed.
+- Source-catalog plus governed-refresh focused suite passed: 20 tests.
+- Full `scripts/run_readiness_check.sh` passed: 779 backend tests, 110
+  documented route contracts, one Alembic head, frontend typecheck/build and
+  all packaged smoke checks.
+
+Next:
+- Run the full release gate, deploy to Tower, execute a live read-only plan and
+  then stage/apply only the exact confirmed official edition.

docs/DATABASE_IMPLEMENTATION_PLAN.md

@@ -267,6 +267,13 @@ evidence audit. External catalogue checks and refresh jobs remain explicit
 operator actions; they may never overwrite a fixed edition or scenario in
 place.
 
+Regional GRB refreshes add no lifecycle table. A read-only plan compares the
+official dated edition with the latest Dataset in each governed temporal
+series. Staging is filesystem evidence only. Checksum-confirmed apply creates a
+new Dataset plus DatasetVersion and vector_features through the existing
+DatasetService/VectorFeatureService transaction; prior snapshots remain
+unchanged and queryable for temporal comparison.
+
 ## Geometry normalization
 
 - User-drawn polygons arrive as EPSG:4326.

docs/TODO.md

@@ -40,6 +40,7 @@
 - [x] Present municipal DHMV/VMM partitions as logical regional layers and analyse cross-boundary rectangles without a municipality prerequisite.
 - [x] Add a read-only source freshness/version audit with explicit publication policies, local integrity checks, operator CLI and compact Status workspace surface.
 - [x] Add explicit bounded GRB/orthophoto catalogue release probes using official capabilities and ISO metadata; keep every acquisition separate and operator-controlled.
+- [x] Add a governed regional GRB plan -> stage -> checksum-confirmed apply workflow that preserves every previous snapshot.
 - [ ] Extend catalogue probes only to additional sources that publish a stable machine-readable edition contract; do not add background polling or infer releases from HTTP dates alone.
 
 ## Governed source expansion backlog

frontend/README.md

@@ -13,6 +13,13 @@ replaces source data. A separate `Officiële edities controleren` action is the
 only trigger for bounded GRB/orthophoto catalog reads. It presents official and
 local editions, layer-contract coverage and honest non-comparable version
 markers without starting an import or background poll.
+After that explicit check, the same surface shows a compact GRB refresh plan
+for buildings, roads, water and parcels. It states whether each local regional
+snapshot is current, updateable or needs review and shows the current object
+count. The UI cannot start the million-feature import: staging and checksum-
+confirmed application remain an operator action so a browser request cannot
+silently replace regional reference data. Existing snapshots remain visible
+as historical observations.
 
 The user-facing shell is task based: `Kaart`, `Bronnen`, `Kwaliteit`, `Beeldanalyse`, `Downloads`, `Status` and `Beheer`. Internal benchmark projects, raw dataset metadata, provider capabilities, model registry details and QA evidence remain accessible through labelled advanced disclosures instead of competing with the normal workflow.
 

frontend/src/App.tsx

@@ -864,6 +864,9 @@ function App(): JSX.Element {
                 catalogLoading={sourceFreshness.catalogLoading}
                 catalogError={sourceFreshness.catalogError}
                 onProbeCatalogs={(force) => { void sourceFreshness.probeCatalogs(force) }}
+                grbRefreshPlan={sourceFreshness.grbRefreshPlan}
+                grbRefreshPlanLoading={sourceFreshness.grbRefreshPlanLoading}
+                grbRefreshPlanError={sourceFreshness.grbRefreshPlanError}
               />
               <details className="status-details-disclosure">
                 <summary>

frontend/src/components/status/SourceFreshnessPanel.tsx

@@ -1,4 +1,7 @@
 import type {
+  GrbRefreshLayerPlan,
+  GrbRefreshLayerStatus,
+  GrbRefreshPlan,
   SourceCatalogComparisonStatus,
   SourceCatalogProbeItem,
   SourceCatalogProbeReport,
@@ -17,6 +20,9 @@ interface SourceFreshnessPanelProps {
   catalogLoading: boolean
   catalogError: string | null
   onProbeCatalogs: (force: boolean) => void
+  grbRefreshPlan: GrbRefreshPlan | null
+  grbRefreshPlanLoading: boolean
+  grbRefreshPlanError: string | null
 }
 
 function statusLabel(status: SourceFreshnessStatus): string {
@@ -46,6 +52,35 @@ function formatDate(value?: string | null): string {
   return new Intl.DateTimeFormat('nl-BE', { dateStyle: 'medium' }).format(new Date(value))
 }
 
+function formatCount(value?: number | null): string {
+  return value == null ? 'nog niet bekend' : value.toLocaleString('nl-BE')
+}
+
+function grbLayerStatusLabel(status: GrbRefreshLayerStatus): string {
+  if (status === 'current') return 'actueel'
+  if (status === 'update_available') return 'update beschikbaar'
+  if (status === 'not_loaded') return 'nog niet ingeladen'
+  if (status === 'remote_unavailable') return 'bron niet bereikbaar'
+  return 'beoordeling nodig'
+}
+
+function GrbRefreshLayerRow({ item, remoteEdition }: { item: GrbRefreshLayerPlan; remoteEdition?: string | null }): JSX.Element {
+  return (
+    <div className={`grb-refresh-layer grb-refresh-layer-${item.status}`}>
+      <div>
+        <strong>{item.display_name}</strong>
+        <span>{grbLayerStatusLabel(item.status)}</span>
+      </div>
+      <p>{item.action_message}</p>
+      <div className="source-freshness-meta">
+        <span>Lokaal: {item.local_source_version ?? 'niet aanwezig'}</span>
+        <span>Officieel: {remoteEdition ?? 'onbekend'}</span>
+        <span>{formatCount(item.local_feature_count)} objecten lokaal</span>
+      </div>
+    </div>
+  )
+}
+
 function integrityIssueCount(item: SourceFreshnessItem): number {
   return Object.values(item.integrity).reduce((total, value) => total + value, 0)
 }
@@ -102,6 +137,9 @@ export function SourceFreshnessPanel({
   catalogLoading,
   catalogError,
   onProbeCatalogs,
+  grbRefreshPlan,
+  grbRefreshPlanLoading,
+  grbRefreshPlanError,
 }: SourceFreshnessPanelProps): JSX.Element {
   const attentionItems = report?.items.filter((item) => item.status === 'due' || item.status === 'review_required') ?? []
   const summary = report?.summary
@@ -180,6 +218,27 @@ export function SourceFreshnessPanel({
                 </p>
               </>
             ) : null}
+            {grbRefreshPlanLoading ? <p className="source-catalog-empty">Veilig vernieuwingsplan wordt opgebouwd...</p> : null}
+            {grbRefreshPlanError ? <p className="inline-error">{grbRefreshPlanError}</p> : null}
+            {grbRefreshPlan ? (
+              <div className="grb-refresh-plan">
+                <div className="grb-refresh-plan-heading">
+                  <div>
+                    <strong>Veilige GRB-vernieuwing</strong>
+                    <span>{grbRefreshPlan.message}</span>
+                  </div>
+                  <span>{grbRefreshPlan.summary.update_available_count} updates</span>
+                </div>
+                <div className="grb-refresh-layer-list">
+                  {grbRefreshPlan.layers.map((item) => (
+                    <GrbRefreshLayerRow item={item} remoteEdition={grbRefreshPlan.remote_edition_date} key={item.theme} />
+                  ))}
+                </div>
+                <p className="source-freshness-limitation">
+                  Een serveroperator controleert eerst exacte aantallen en checksums. Bestaande snapshots blijven altijd behouden.
+                </p>
+              </div>
+            ) : null}
           </div>
         </>
       ) : null}

frontend/src/hooks/useSourceFreshness.ts

@@ -1,7 +1,7 @@
 import { useCallback, useEffect, useRef, useState } from 'react'
 
 import { datasetsApi } from '../services/api'
-import type { SourceCatalogProbeReport, SourceFreshnessReport } from '../types'
+import type { GrbRefreshPlan, SourceCatalogProbeReport, SourceFreshnessReport } from '../types'
 
 export function useSourceFreshness(selectedProjectId: string | null) {
   const [report, setReport] = useState<SourceFreshnessReport | null>(null)
@@ -12,6 +12,9 @@ export function useSourceFreshness(selectedProjectId: string | null) {
   const [catalogReport, setCatalogReport] = useState<SourceCatalogProbeReport | null>(null)
   const [catalogLoading, setCatalogLoading] = useState(false)
   const [catalogError, setCatalogError] = useState<string | null>(null)
+  const [grbRefreshPlan, setGrbRefreshPlan] = useState<GrbRefreshPlan | null>(null)
+  const [grbRefreshPlanLoading, setGrbRefreshPlanLoading] = useState(false)
+  const [grbRefreshPlanError, setGrbRefreshPlanError] = useState<string | null>(null)
 
   const refresh = useCallback(async () => {
     const requestId = ++requestSequence.current
@@ -45,10 +48,15 @@ export function useSourceFreshness(selectedProjectId: string | null) {
       setCatalogReport(null)
       setCatalogError(null)
       setCatalogLoading(false)
+      setGrbRefreshPlan(null)
+      setGrbRefreshPlanError(null)
+      setGrbRefreshPlanLoading(false)
       return
     }
     setCatalogLoading(true)
     setCatalogError(null)
+    setGrbRefreshPlanLoading(true)
+    setGrbRefreshPlanError(null)
     try {
       const nextReport = await datasetsApi.sourceCatalogProbes(selectedProjectId, force)
       if (catalogRequestSequence.current === requestId) {
@@ -58,9 +66,25 @@ export function useSourceFreshness(selectedProjectId: string | null) {
       if (catalogRequestSequence.current === requestId) {
         setCatalogError(caught instanceof Error ? caught.message : 'De officiële broncatalogi konden niet worden gecontroleerd.')
       }
+      if (catalogRequestSequence.current === requestId) {
+        setCatalogLoading(false)
+        setGrbRefreshPlanLoading(false)
+      }
+      return
+    }
+    try {
+      const nextPlan = await datasetsApi.grbRefreshPlan(selectedProjectId)
+      if (catalogRequestSequence.current === requestId) {
+        setGrbRefreshPlan(nextPlan)
+      }
+    } catch (caught) {
+      if (catalogRequestSequence.current === requestId) {
+        setGrbRefreshPlanError(caught instanceof Error ? caught.message : 'Het veilige GRB-vernieuwingsplan kon niet worden opgebouwd.')
+      }
     } finally {
       if (catalogRequestSequence.current === requestId) {
         setCatalogLoading(false)
+        setGrbRefreshPlanLoading(false)
       }
     }
   }, [selectedProjectId])
@@ -77,6 +101,9 @@ export function useSourceFreshness(selectedProjectId: string | null) {
     setCatalogReport(null)
     setCatalogError(null)
     setCatalogLoading(false)
+    setGrbRefreshPlan(null)
+    setGrbRefreshPlanError(null)
+    setGrbRefreshPlanLoading(false)
   }, [selectedProjectId])
 
   return {
@@ -88,5 +115,8 @@ export function useSourceFreshness(selectedProjectId: string | null) {
     catalogLoading,
     catalogError,
     probeCatalogs,
+    grbRefreshPlan,
+    grbRefreshPlanLoading,
+    grbRefreshPlanError,
   }
 }

frontend/src/services/api/datasets.ts

@@ -29,6 +29,7 @@ import type {
   ThematicRasterSelectionResponse,
   SourceFreshnessReport,
   SourceCatalogProbeReport,
+  GrbRefreshPlan,
 } from '../../types'
 
 const DATASET_PAGE_SIZE = 200
@@ -66,6 +67,10 @@ export const datasetsApi = {
     apiGet<SourceCatalogProbeReport>(
       `/api/v1/projects/${projectId}/datasets/source-catalog-probes?refresh=${refresh ? 'true' : 'false'}`,
     ),
+  grbRefreshPlan: (projectId: string, refreshCatalog = false): Promise<GrbRefreshPlan> =>
+    apiGet<GrbRefreshPlan>(
+      `/api/v1/projects/${projectId}/datasets/grb-refresh-plan?refresh_catalog=${refreshCatalog ? 'true' : 'false'}`,
+    ),
   upload: (
     projectId: string,
     payload: {

frontend/src/styles/premium.css

@@ -1041,6 +1041,95 @@ details.ai-lab-model-surface > summary strong {
   font-size: 0.75rem;
 }
 
+.grb-refresh-plan {
+  border-top: 1px solid var(--line);
+  background: #ffffff;
+}
+
+.grb-refresh-plan-heading {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 1rem;
+  padding: 0.8rem 1rem;
+}
+
+.grb-refresh-plan-heading > div {
+  display: grid;
+  min-width: 0;
+  gap: 0.15rem;
+}
+
+.grb-refresh-plan-heading strong {
+  color: #23313b;
+  font-size: 0.84rem;
+}
+
+.grb-refresh-plan-heading span {
+  color: var(--muted);
+  font-size: 0.75rem;
+}
+
+.grb-refresh-plan-heading > span {
+  flex: 0 0 auto;
+  font-weight: 750;
+}
+
+.grb-refresh-layer-list {
+  display: grid;
+  grid-template-columns: repeat(2, minmax(0, 1fr));
+  border-top: 1px solid var(--line);
+}
+
+.grb-refresh-layer {
+  min-width: 0;
+  padding: 0.78rem 1rem;
+  border-right: 1px solid var(--line);
+  border-bottom: 1px solid var(--line);
+  box-shadow: inset 3px 0 0 #8a98a3;
+}
+
+.grb-refresh-layer:nth-child(2n) {
+  border-right: 0;
+}
+
+.grb-refresh-layer:nth-last-child(-n + 2) {
+  border-bottom: 0;
+}
+
+.grb-refresh-layer-update_available,
+.grb-refresh-layer-not_loaded {
+  box-shadow: inset 3px 0 0 #ca7a18;
+}
+
+.grb-refresh-layer-current {
+  box-shadow: inset 3px 0 0 #2d9272;
+}
+
+.grb-refresh-layer-remote_unavailable,
+.grb-refresh-layer-review_required {
+  box-shadow: inset 3px 0 0 #b84e4e;
+}
+
+.grb-refresh-layer > div:first-child {
+  display: flex;
+  align-items: center;
+  justify-content: space-between;
+  gap: 0.75rem;
+}
+
+.grb-refresh-layer > div:first-child span {
+  color: #4d5c66;
+  font-size: 0.72rem;
+  font-weight: 750;
+}
+
+.grb-refresh-layer p {
+  margin: 0.4rem 0;
+  color: var(--muted);
+  font-size: 0.75rem;
+}
+
 .source-freshness-details > summary {
   display: flex;
   cursor: pointer;
@@ -1097,6 +1186,26 @@ details.ai-lab-model-surface > summary strong {
   .source-catalog-row:last-child {
     border-bottom: 0;
   }
+
+  .grb-refresh-plan-heading {
+    align-items: flex-start;
+    flex-direction: column;
+  }
+
+  .grb-refresh-layer-list {
+    grid-template-columns: 1fr;
+  }
+
+  .grb-refresh-layer,
+  .grb-refresh-layer:nth-child(2n),
+  .grb-refresh-layer:nth-last-child(-n + 2) {
+    border-right: 0;
+    border-bottom: 1px solid var(--line);
+  }
+
+  .grb-refresh-layer:last-child {
+    border-bottom: 0;
+  }
 }
 
 .workflow-guidance-panel {

frontend/src/types.ts

@@ -755,6 +755,58 @@ export interface SourceCatalogProbeReport {
   limitations: string[]
 }
 
+export type GrbRefreshLayerStatus =
+  | 'current'
+  | 'update_available'
+  | 'not_loaded'
+  | 'review_required'
+  | 'remote_unavailable'
+
+export interface GrbRefreshLayerPlan {
+  theme: 'buildings' | 'roads' | 'water' | 'parcels'
+  display_name: string
+  collections: string[]
+  temporal_series_key: string
+  status: GrbRefreshLayerStatus
+  local_dataset_id?: string | null
+  local_source_version?: string | null
+  local_observed_at?: string | null
+  local_imported_at?: string | null
+  local_feature_count?: number | null
+  local_size_bytes?: number | null
+  retained_after_refresh: boolean
+  action_message: string
+}
+
+export interface GrbRefreshPlan {
+  project_id: string
+  scope: string
+  generated_at: string
+  remote_status: string
+  remote_version?: string | null
+  remote_edition_date?: string | null
+  catalog_checked_at?: string | null
+  summary: {
+    layer_count: number
+    current_count: number
+    update_available_count: number
+    not_loaded_count: number
+    review_required_count: number
+    remote_unavailable_count: number
+    new_dataset_count_if_applied: number
+    retained_dataset_count: number
+    current_feature_count: number
+    current_size_bytes: number
+  }
+  layers: GrbRefreshLayerPlan[]
+  execution_mode: 'operator_stage_then_apply'
+  staging_required: true
+  automatic_import: false
+  destructive_replacement: false
+  message: string
+  limitations: string[]
+}
+
 export interface GeojsonEnvelopeResponse {
   data: object
 }

scripts/README.md

@@ -1716,6 +1716,42 @@ JSON output contains `source_freshness` and `catalog_probes`; without it, the
 original local report shape is unchanged. No flag downloads provider features
 or imagery and no flag writes a Dataset.
 
+## Governed GRB refresh
+
+Use the refresh coordinator only inside the GeoIntel container. The default
+`plan` action is read-only and prints the canonical decision from the API:
+
+```bash
+docker exec geointel python /app/scripts/manage_grb_refresh.py plan \
+  --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --refresh-catalog
+```
+
+A refresh is deliberately split into two operator approvals:
+
+```bash
+docker exec geointel python /app/scripts/manage_grb_refresh.py stage \
+  --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --confirm-edition 2026-07-15
+
+docker exec geointel python /app/scripts/manage_grb_refresh.py apply \
+  --project-id 82a85913-c522-45d7-84a1-02b393d89e55 \
+  --api-url http://127.0.0.1:8000/api/v1 \
+  --confirm-edition 2026-07-15 \
+  --confirm-plan-sha256 SHA256_FROM_STAGE
+```
+
+`stage` delegates to the existing buildings/context operators in fetch-only
+mode, validates every municipality partition and writes
+`/app/storage/operator-evidence/grb-refresh/<scope>/<edition>/staged-plan.json`.
+The plan contains exact feature-count deltas and artifact sizes. `apply`
+revalidates the entire plan and refuses any mismatched byte before delegating
+to the DatasetService-based persistence path. Existing snapshots are retained.
+The coordinator accepts no provider URL, collection name, arbitrary process or
+automatic schedule.
+
 ## Tower deployment
 
 Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime:

scripts/manage_grb_refresh.py

@@ -0,0 +1,373 @@
+#!/usr/bin/env python3
+"""Plan, stage and explicitly apply a governed regional GRB refresh.
+
+Planning is read-only. Staging downloads resumable municipality partitions but
+does not persist application data. Applying requires the exact staged plan
+SHA-256 and delegates persistence to the existing DatasetService-based regional
+operators. Existing snapshots are immutable and remain available.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+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
+
+
+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-themes")
+DEFAULT_EVIDENCE_ROOT = Path("/app/storage/operator-evidence/grb-refresh")
+THEMES = ("buildings", "roads", "water", "parcels")
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Governed GRB refresh: plan, stage, then checksum-confirmed apply.")
+    parser.add_argument("action", choices=("plan", "stage", "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("--layers", nargs="+", default=list(THEMES), help="Subset: buildings roads water parcels")
+    parser.add_argument("--confirm-edition", help="Exact official ISO edition date required for stage/apply")
+    parser.add_argument("--confirm-plan-sha256", help="Exact staged plan hash required for apply")
+    parser.add_argument("--plan-path", type=Path, help="Staged plan path; defaults to the governed evidence location")
+    parser.add_argument("--output-root", type=Path, default=Path(os.environ.get("GEOINTEL_REGIONAL_THEME_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)))
+    parser.add_argument("--evidence-root", type=Path, default=Path(os.environ.get("GEOINTEL_OPERATOR_EVIDENCE_ROOT", DEFAULT_EVIDENCE_ROOT)))
+    parser.add_argument("--refresh-catalog", action="store_true", help="Bypass the short catalog probe cache")
+    parser.add_argument("--request-timeout", type=int, default=180)
+    parser.add_argument("--api-timeout", type=int, default=180)
+    parser.add_argument("--batch-size", type=int, default=1000)
+    parser.add_argument("--page-limit", type=int, default=1000)
+    parser.add_argument("--max-features-per-member", type=int, default=100000)
+    parser.add_argument("--max-total-features", type=int, default=1500000)
+    return parser.parse_args()
+
+
+def selected_themes(values: str | list[str]) -> list[str]:
+    raw = [values] if isinstance(values, str) else values
+    requested = {item.strip().lower() for value in raw for item in value.split(",") if item.strip()}
+    unknown = requested - set(THEMES)
+    if unknown or not requested:
+        raise ValueError(f"Unsupported GRB layers: {sorted(unknown)}")
+    return [theme for theme in THEMES if theme in requested]
+
+
+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-GRB-refresh/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 fetch_refresh_plan(args: argparse.Namespace, *, refresh: bool) -> dict[str, Any]:
+    query = "true" if refresh else "false"
+    return api_data(
+        args.api_url,
+        f"projects/{args.project_id}/datasets/grb-refresh-plan?scope={args.scope}&refresh_catalog={query}",
+        args.api_timeout,
+    )
+
+
+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 require_confirmed_edition(plan: dict[str, Any], confirmation: str | None) -> str:
+    edition = str(plan.get("remote_edition_date") or "")
+    if not edition:
+        raise RuntimeError("The official GRB catalog did not provide a safe ISO edition date")
+    if confirmation != edition:
+        raise RuntimeError(f"Explicit --confirm-edition {edition} is required")
+    return edition
+
+
+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 build_operator_commands(
+    args: argparse.Namespace,
+    themes: list[str],
+    edition: str,
+    *,
+    fetch_only: bool,
+) -> list[tuple[str, list[str]]]:
+    scripts_dir = Path(__file__).resolve().parent
+    common = [
+        "--scope", args.scope,
+        "--observed-date", edition,
+        "--base-url", internal_base_url(args.api_url),
+        "--output-root", str(args.output_root),
+        "--request-timeout", str(args.request_timeout),
+        "--api-timeout", str(args.api_timeout),
+        "--batch-size", str(args.batch_size),
+        "--page-limit", str(args.page_limit),
+        "--max-features-per-member", str(args.max_features_per_member),
+        "--max-total-features", str(args.max_total_features),
+    ]
+    if fetch_only:
+        common.append("--fetch-only")
+    commands: list[tuple[str, list[str]]] = []
+    if "buildings" in themes:
+        commands.append(("buildings", [sys.executable, str(scripts_dir / "provision_regional_grb_buildings.py"), *common]))
+    context = [theme for theme in themes if theme != "buildings"]
+    if context:
+        commands.append(
+            (
+                "context",
+                [
+                    sys.executable,
+                    str(scripts_dir / "provision_regional_grb_context.py"),
+                    "--layers",
+                    *context,
+                    *common,
+                ],
+            )
+        )
+    return commands
+
+
+def run_operator(label: str, command: list[str]) -> dict[str, Any]:
+    print(f"GRB {label}: {'staging' if '--fetch-only' in command else 'applying'}...", 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"GRB {label} failed with exit {completed.returncode}: {detail[-3000:]}")
+    try:
+        payload = json.loads(completed.stdout)
+    except json.JSONDecodeError as exc:
+        raise RuntimeError(f"GRB {label} returned invalid JSON: {completed.stdout[-1000:]}") from exc
+    if payload.get("status") != "ok":
+        raise RuntimeError(f"GRB {label} did not report success")
+    return payload
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(8 * 1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def canonical_plan_sha256(payload: dict[str, Any]) -> str:
+    content = {key: value for key, value in payload.items() if key != "plan_sha256"}
+    encoded = json.dumps(content, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    return hashlib.sha256(encoded).hexdigest()
+
+
+def manifest_path(output_root: Path, scope: str, theme: str, edition: str) -> Path:
+    name = "regional_buildings_manifest.json" if theme == "buildings" else f"regional_{theme}_manifest.json"
+    return output_root / scope / theme / edition / name
+
+
+def validate_manifest(path: Path, *, output_root: Path, scope: str, theme: str, edition: str) -> dict[str, Any]:
+    root = output_root.resolve()
+    resolved = path.resolve()
+    if not resolved.is_relative_to(root):
+        raise RuntimeError(f"Manifest is outside the governed output root: {path}")
+    if not path.is_file():
+        raise RuntimeError(f"Staged manifest is missing: {path}")
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    if payload.get("status") != "complete" or payload.get("scope") != scope or payload.get("theme") != theme:
+        raise RuntimeError(f"Staged manifest identity is invalid: {path}")
+    if payload.get("observed_at") != edition or payload.get("reference_truncated") is not False:
+        raise RuntimeError(f"Staged manifest edition or completeness is invalid: {path}")
+    artifact = path.parent / str(payload.get("artifact_filename") or "")
+    if not artifact.is_file() or sha256_file(artifact) != payload.get("artifact_sha256"):
+        raise RuntimeError(f"Staged artifact checksum is invalid: {artifact}")
+    partitions = payload.get("partitions")
+    if not isinstance(partitions, list) or len(partitions) != int(payload.get("member_count") or 0):
+        raise RuntimeError(f"Staged partition count is invalid: {path}")
+    partition_root = path.parent / "partitions"
+    for item in partitions:
+        partition = partition_root / str(item.get("filename") or "")
+        if not partition.is_file() or sha256_file(partition) != item.get("sha256"):
+            raise RuntimeError(f"Staged partition checksum is invalid: {partition}")
+    return payload
+
+
+def default_plan_path(args: argparse.Namespace, edition: str) -> Path:
+    return args.evidence_root / args.scope / edition / "staged-plan.json"
+
+
+def build_staged_plan(
+    args: argparse.Namespace,
+    remote_plan: dict[str, Any],
+    themes: list[str],
+    edition: str,
+) -> dict[str, Any]:
+    local_by_theme = {item["theme"]: item for item in remote_plan.get("layers", [])}
+    layers: list[dict[str, Any]] = []
+    for theme in themes:
+        path = manifest_path(args.output_root, args.scope, theme, edition)
+        manifest = validate_manifest(path, output_root=args.output_root, scope=args.scope, theme=theme, edition=edition)
+        local = local_by_theme.get(theme, {})
+        staged_count = int(manifest["feature_count"])
+        local_count = local.get("local_feature_count")
+        layers.append(
+            {
+                "theme": theme,
+                "collections": manifest.get("grb_collections") or (["GBG"] if theme == "buildings" else []),
+                "manifest_path": str(path),
+                "artifact_path": str(path.parent / manifest["artifact_filename"]),
+                "artifact_sha256": manifest["artifact_sha256"],
+                "artifact_size_bytes": int(manifest["artifact_size_bytes"]),
+                "partition_count": len(manifest["partitions"]),
+                "staged_feature_count": staged_count,
+                "local_dataset_id": local.get("local_dataset_id"),
+                "local_source_version": local.get("local_source_version"),
+                "local_feature_count": local_count,
+                "feature_count_delta": staged_count - int(local_count) if local_count is not None else None,
+                "existing_snapshot_retained": True,
+            }
+        )
+    payload: dict[str, Any] = {
+        "schema_version": 1,
+        "status": "staged",
+        "created_at": datetime.now(timezone.utc).isoformat(),
+        "project_id": args.project_id,
+        "scope": args.scope,
+        "official_remote_version": remote_plan.get("remote_version"),
+        "edition": edition,
+        "layers": layers,
+        "total_staged_feature_count": sum(item["staged_feature_count"] for item in layers),
+        "total_artifact_size_bytes": sum(item["artifact_size_bytes"] for item in layers),
+        "automatic_import": False,
+        "destructive_replacement": False,
+        "apply_requires_plan_sha256": True,
+    }
+    payload["plan_sha256"] = canonical_plan_sha256(payload)
+    return payload
+
+
+def write_json(path: Path, payload: dict[str, Any]) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(f"{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 load_and_validate_staged_plan(args: argparse.Namespace, edition: str) -> tuple[Path, dict[str, Any]]:
+    path = args.plan_path or default_plan_path(args, edition)
+    if not path.is_file():
+        raise RuntimeError(f"Staged refresh plan is missing: {path}")
+    payload = json.loads(path.read_text(encoding="utf-8"))
+    actual_sha = canonical_plan_sha256(payload)
+    if payload.get("plan_sha256") != actual_sha:
+        raise RuntimeError("Staged refresh 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:
+        raise RuntimeError("Staged refresh plan project or status is invalid")
+    if payload.get("scope") != args.scope or payload.get("edition") != edition:
+        raise RuntimeError("Staged refresh plan scope or edition is invalid")
+    for layer in payload.get("layers") or []:
+        manifest = validate_manifest(
+            Path(layer["manifest_path"]),
+            output_root=args.output_root,
+            scope=args.scope,
+            theme=layer["theme"],
+            edition=edition,
+        )
+        if manifest.get("artifact_sha256") != layer.get("artifact_sha256"):
+            raise RuntimeError(f"Staged plan no longer matches {layer['theme']} artifact")
+    return path, payload
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        themes = selected_themes(args.layers)
+        if min(args.request_timeout, args.api_timeout, args.batch_size, args.page_limit, args.max_features_per_member, args.max_total_features) <= 0:
+            raise ValueError("All timeout, paging, batching and feature safety limits must be positive")
+        validate_project_scope(args)
+        remote_plan = fetch_refresh_plan(args, refresh=args.refresh_catalog or args.action == "stage")
+        if args.action == "plan":
+            print(json.dumps(remote_plan, ensure_ascii=False, indent=2))
+            return 0
+
+        edition = require_confirmed_edition(remote_plan, args.confirm_edition)
+        if args.action == "stage":
+            actionable = {
+                item["theme"]
+                for item in remote_plan.get("layers", [])
+                if item.get("status") in {"update_available", "not_loaded"}
+            }
+            blocked = set(themes) - actionable
+            if blocked:
+                raise RuntimeError(f"Layers are not safely stageable according to the refresh plan: {sorted(blocked)}")
+            commands = build_operator_commands(args, themes, edition, fetch_only=True)
+            operator_results = {label: run_operator(label, command) for label, command in commands}
+            staged = build_staged_plan(args, remote_plan, themes, edition)
+            staged["operator_results"] = operator_results
+            staged["plan_sha256"] = canonical_plan_sha256(staged)
+            path = args.plan_path or default_plan_path(args, edition)
+            write_json(path, staged)
+            print(json.dumps({"status": "staged", "plan_path": str(path), **staged}, ensure_ascii=False, indent=2))
+            return 0
+
+        path, staged = load_and_validate_staged_plan(args, edition)
+        staged_themes = [item["theme"] for item in staged["layers"]]
+        commands = build_operator_commands(args, staged_themes, edition, fetch_only=False)
+        operator_results = {label: run_operator(label, command) for label, command in commands}
+        final_plan = fetch_refresh_plan(args, refresh=False)
+        by_theme = {item["theme"]: item for item in final_plan.get("layers", [])}
+        incomplete = [theme for theme in staged_themes if by_theme.get(theme, {}).get("status") != "current"]
+        if incomplete:
+            raise RuntimeError(f"Applied datasets did not become current: {incomplete}")
+        evidence = {
+            "schema_version": 1,
+            "status": "applied",
+            "applied_at": datetime.now(timezone.utc).isoformat(),
+            "project_id": args.project_id,
+            "scope": args.scope,
+            "edition": edition,
+            "staged_plan_path": str(path),
+            "staged_plan_sha256": staged["plan_sha256"],
+            "operator_results": operator_results,
+            "resulting_layers": [by_theme[theme] for theme in staged_themes],
+            "existing_snapshots_retained": True,
+        }
+        evidence_path = 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", "message": str(exc)}, ensure_ascii=False), file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/run_readiness_check.sh

@@ -66,6 +66,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_geographic_scope.py
 ${PYTHON_BIN} -m py_compile scripts/provision_regional_grb_buildings.py
 ${PYTHON_BIN} -m py_compile scripts/provision_regional_grb_context.py
 ${PYTHON_BIN} -m py_compile scripts/audit_source_freshness.py
+${PYTHON_BIN} -m py_compile scripts/manage_grb_refresh.py
 ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_dataset.py
 ${PYTHON_BIN} -m py_compile scripts/export_operator_yolo_tile_dataset.py
 ${PYTHON_BIN} -m py_compile scripts/audit_operator_yolo_dataset_quality.py