feat: provision regional Kempen context layers

CHANGELOG.md

@@ -7,6 +7,16 @@
 
 # Changelog
 
+## Sprint 191 Regional Kempen GRB context foundation (2026-07-14)
+
+- Added one explicit regional operator for current GRB roads, water and parcels, with independent resumable municipality partitions and one normal PostGIS dataset per theme.
+- Preserved official source geometry dimensions across `Wegsegment`, `WTZ`, `WLAS`, `WGR` and `ADP`; collection-qualified source IDs prevent collisions inside the combined water layer.
+- Assigned cross-boundary polygons by maximum overlap area and lines by maximum overlap length, with deterministic NIS-code tie breaking and clipping only to the complete approved region.
+- Reused DatasetService, VectorFeatureService, immutable dataset versions, exact selection aggregation and existing Area/bbox APIs; no API contract, migration or direct vector-feature write was added.
+- Added truncation guards, atomic manifests, checksum reuse, duplicate rejection, Docker packaging, readiness compilation and focused geometry/persistence tests.
+- Documented source semantics honestly: road objects are not traffic data, heterogeneous water objects are not volume metrics and ADP is not a legal cadastral survey.
+- Kept all source access operator-only; the browser, startup path and public `not_configured` GRB provider perform no external fetch.
+
 ## Sprint 190 Regional Kempen GRB buildings (2026-07-14)
 
 - Added an explicit regional GRB building operator that fetches the approved Kempen scope in 28 resumable municipality partitions and follows every OGC API pagination link.

backend/README.md

@@ -972,6 +972,22 @@ exact manifest feature count or it rolls back and removes the managed copy.
 No public API contract or provider readiness claim is changed by this
 operator-only path.
 
+Provision the regional current road, water and parcel context through the
+same persistence boundary:
+
+```bash
+docker exec geointel python /app/scripts/provision_regional_grb_context.py \
+  --scope kempen-transport-region --layers roads water parcels
+```
+
+The operator keeps one resumable municipality partition set per theme and
+creates one regional reference Dataset per theme. Polygon ownership uses
+maximum overlap area; line ownership uses maximum overlap length. It preserves
+source geometry dimensions and collection-qualified source IDs, copies the
+combined artifact through StorageService and indexes bounded batches through
+DatasetService/VectorFeatureService. It does not add API routes, direct SQL or
+interactive provider downloads.
+
 ## Temporal Mol data and evolution
 
 Dataset uploads accept `temporal_series_key`, `observed_at`, `valid_from`,

backend/tests/test_sprint191_regional_grb_context.py

@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+from pathlib import Path
+import sys
+
+import pytest
+from shapely.geometry import LineString, Polygon, shape
+from shapely.ops import unary_union
+
+
+ROOT = Path(__file__).resolve().parents[2]
+SCRIPTS = ROOT / "scripts"
+
+
+def load_operator():
+    scripts_path = str(SCRIPTS)
+    if scripts_path not in sys.path:
+        sys.path.insert(0, scripts_path)
+    spec = importlib.util.spec_from_file_location(
+        "provision_regional_grb_context_test",
+        SCRIPTS / "provision_regional_grb_context.py",
+    )
+    assert spec is not None
+    assert spec.loader is not None
+    module = importlib.util.module_from_spec(spec)
+    sys.modules[spec.name] = module
+    spec.loader.exec_module(module)
+    return module
+
+
+def source_feature(feature_id: str, geometry) -> dict:
+    return {
+        "type": "Feature",
+        "id": feature_id,
+        "geometry": geometry.__geo_interface__,
+        "properties": {"UIDN": feature_id},
+    }
+
+
+def test_layer_registry_matches_verified_official_grb_collections() -> None:
+    operator = load_operator()
+
+    assert [item.key for item in operator.LAYERS] == ["roads", "water", "parcels"]
+    assert [(item.name, item.geometry_dimension) for item in operator.LAYER_BY_KEY["roads"].collections] == [
+        ("Wegsegment", 1)
+    ]
+    assert [(item.name, item.geometry_dimension) for item in operator.LAYER_BY_KEY["water"].collections] == [
+        ("WTZ", 2),
+        ("WLAS", 1),
+        ("WGR", 1),
+    ]
+    assert [(item.name, item.geometry_dimension) for item in operator.LAYER_BY_KEY["parcels"].collections] == [
+        ("ADP", 2)
+    ]
+    assert [item.key for item in operator.selected_definitions("parcels,roads")] == ["roads", "parcels"]
+    with pytest.raises(ValueError, match="Unsupported layers"):
+        operator.selected_definitions("buildings")
+
+
+def test_line_owner_uses_intersection_length_and_deterministic_tie_break() -> None:
+    operator = load_operator()
+    scopes = __import__("geographic_scopes")
+    alpha = scopes.ScopeMember("Alpha", "10001")
+    beta = scopes.ScopeMember("Beta", "10002")
+    alpha_boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
+    beta_boundary = Polygon([(1, 0), (2, 0), (2, 1), (1, 1)])
+    members = [(alpha, alpha_boundary), (beta, beta_boundary)]
+
+    assert operator.assign_owner_nis(
+        LineString([(0.7, 0.5), (1.1, 0.5)]),
+        members,
+        expected_dimension=1,
+    ) == "10001"
+    assert operator.assign_owner_nis(
+        LineString([(0.8, 0.5), (1.2, 0.5)]),
+        members,
+        expected_dimension=1,
+    ) == "10001"
+
+
+def test_mixed_water_partition_preserves_dimensions_and_source_identity() -> None:
+    operator = load_operator()
+    scopes = __import__("geographic_scopes")
+    member = scopes.ScopeMember("Alpha", "10001")
+    scope = scopes.GeographicScope(
+        key="test-region",
+        display_name="Test region",
+        project_name="Test project",
+        project_region="Test",
+        area_name="Test boundary",
+        authority_name="Test",
+        authority_url="https://example.test/scope",
+        scope_type="policy_region",
+        limitation_message="Test only.",
+        members=(member,),
+    )
+    boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
+    definition = operator.LAYER_BY_KEY["water"]
+    pages = [
+        (
+            definition.collections[0],
+            {"type": "FeatureCollection", "features": [source_feature("WTZ.1", Polygon([(0.1, 0.1), (0.3, 0.1), (0.3, 0.3), (0.1, 0.3)]))]},
+            "https://example.test/wtz",
+        ),
+        (
+            definition.collections[1],
+            {"type": "FeatureCollection", "features": [source_feature("WLAS.1", LineString([(-0.2, 0.5), (0.5, 0.5)]))]},
+            "https://example.test/wlas",
+        ),
+        (
+            definition.collections[2],
+            {"type": "FeatureCollection", "features": [source_feature("WGR.1", LineString([(0.4, 0.7), (0.8, 0.7)]))]},
+            "https://example.test/wgr",
+        ),
+    ]
+
+    features, summary = operator.build_partition_features(
+        pages,
+        definition=definition,
+        member=member,
+        members=[(member, boundary)],
+        regional_boundary=boundary,
+        scope=scope,
+        max_features=10,
+    )
+
+    assert [item["id"] for item in features] == ["WTZ:WTZ.1", "WLAS:WLAS.1", "WGR:WGR.1"]
+    assert [shape(item["geometry"]).geom_type for item in features] == ["Polygon", "LineString", "LineString"]
+    assert shape(features[1]["geometry"]).bounds == (0.0, 0.5, 0.5, 0.5)
+    assert features[1]["properties"]["clipped_to_regional_scope"] is True
+    assert summary["features_by_collection"] == {"WTZ": 1, "WLAS": 1, "WGR": 1}
+    assert summary["reference_truncated"] is False
+
+
+def test_polygon_partition_assignment_does_not_duplicate_cross_boundary_parcel() -> None:
+    operator = load_operator()
+    scopes = __import__("geographic_scopes")
+    alpha = scopes.ScopeMember("Alpha", "10001")
+    beta = scopes.ScopeMember("Beta", "10002")
+    scope = scopes.GeographicScope(
+        key="test-region",
+        display_name="Test region",
+        project_name="Test project",
+        project_region="Test",
+        area_name="Test boundary",
+        authority_name="Test",
+        authority_url="https://example.test/scope",
+        scope_type="policy_region",
+        limitation_message="Test only.",
+        members=(alpha, beta),
+    )
+    alpha_boundary = Polygon([(0, 0), (1, 0), (1, 1), (0, 1)])
+    beta_boundary = Polygon([(1, 0), (2, 0), (2, 1), (1, 1)])
+    members = [(alpha, alpha_boundary), (beta, beta_boundary)]
+    region = unary_union([alpha_boundary, beta_boundary])
+    definition = operator.LAYER_BY_KEY["parcels"]
+    crossing = source_feature("ADP.1", Polygon([(0.7, 0.2), (1.1, 0.2), (1.1, 0.6), (0.7, 0.6)]))
+    page = (definition.collections[0], {"type": "FeatureCollection", "features": [crossing]}, "https://example.test/adp")
+
+    alpha_features, _ = operator.build_partition_features(
+        [page], definition=definition, member=alpha, members=members, regional_boundary=region, scope=scope, max_features=10
+    )
+    with pytest.raises(RuntimeError, match="No GRB parcels"):
+        operator.build_partition_features(
+            [page], definition=definition, member=beta, members=members, regional_boundary=region, scope=scope, max_features=10
+        )
+
+    assert [item["id"] for item in alpha_features] == ["ADP:ADP.1"]
+
+
+def test_combined_context_artifact_rejects_duplicate_source_identity(tmp_path: Path) -> None:
+    operator = load_operator()
+    scope = __import__("geographic_scopes").KEMPEN_TRANSPORT_REGION_SCOPE
+    definition = operator.LAYER_BY_KEY["roads"]
+    feature = source_feature("Wegsegment:Wegsegment.1", LineString([(4.9, 51.1), (4.91, 51.11)]))
+    first = tmp_path / "first.geojson"
+    second = tmp_path / "second.geojson"
+    first.write_text(json.dumps({"type": "FeatureCollection", "features": [feature]}), encoding="utf-8")
+    second.write_text(json.dumps({"type": "FeatureCollection", "features": [source_feature("Wegsegment:Wegsegment.2", LineString([(5.0, 51.2), (5.01, 51.21)]))]}), encoding="utf-8")
+    combined = tmp_path / "combined.geojson"
+
+    summary = operator.write_combined_artifact(
+        combined,
+        definition=definition,
+        scope=scope,
+        observed_date=operator.date(2026, 7, 14),
+        partition_paths=[first, second],
+        expected_feature_count=2,
+    )
+    assert summary["feature_count"] == 2
+
+    second.write_text(first.read_text(encoding="utf-8"), encoding="utf-8")
+    with pytest.raises(RuntimeError, match="Duplicate regional source feature"):
+        operator.write_combined_artifact(
+            combined,
+            definition=definition,
+            scope=scope,
+            observed_date=operator.date(2026, 7, 14),
+            partition_paths=[first, second],
+            expected_feature_count=2,
+        )
+
+
+def test_context_operator_is_packaged_and_uses_existing_service_boundary() -> 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")
+    operator = (SCRIPTS / "provision_regional_grb_context.py").read_text(encoding="utf-8")
+
+    assert "provision_regional_grb_context.py" in dockerfile
+    assert "py_compile scripts/provision_regional_grb_context.py" in readiness
+    assert "DatasetService.import_partitioned_vector_artifact" in operator
+    assert "insert into vector_features" not in operator.lower()
+    assert "db.add(VectorFeature" not in operator

deploy/unraid/Dockerfile.all-in-one

@@ -80,6 +80,7 @@ COPY scripts/provision_official_landuse_timeseries.py /app/scripts/provision_off
 COPY scripts/geographic_scopes.py /app/scripts/geographic_scopes.py
 COPY scripts/provision_geographic_scope.py /app/scripts/provision_geographic_scope.py
 COPY scripts/provision_regional_grb_buildings.py /app/scripts/provision_regional_grb_buildings.py
+COPY scripts/provision_regional_grb_context.py /app/scripts/provision_regional_grb_context.py
 COPY scripts/export_operator_yolo_tile_dataset.py /app/scripts/export_operator_yolo_tile_dataset.py
 COPY scripts/audit_operator_yolo_dataset_quality.py /app/scripts/audit_operator_yolo_dataset_quality.py
 COPY scripts/render_operator_yolo_label_qa_contact_sheets.py /app/scripts/render_operator_yolo_label_qa_contact_sheets.py

docs/CODEX_EXECUTION_LOG.md

@@ -1,3 +1,19 @@
+## Sprint 191 Regional Kempen GRB context foundation (2026-07-14)
+
+Changed:
+- Added `scripts/provision_regional_grb_context.py` for bounded, independently resumable `roads`, `water` and `parcels` source snapshots across the approved 28-municipality scope.
+- Reused geographic-scope artifacts and the existing DatasetService/VectorFeatureService partition import boundary; no direct SQL, migration, API contract or interactive provider fetch was introduced.
+- Preserved line and polygon source dimensions and collection-qualified identities, with length-based or area-based deterministic ownership for boundary-crossing features.
+- Added Docker packaging, readiness compilation, focused source-geometry/service tests and operator/source documentation.
+
+Tested before deployment:
+- `python -m py_compile scripts/provision_regional_grb_context.py`.
+- `python -m pytest backend/tests/test_sprint191_regional_grb_context.py backend/tests/test_sprint190_regional_grb_buildings.py backend/tests/test_sprint106_map_bbox_extract.py -q` (`19 passed`).
+- `bash scripts/run_readiness_check.sh` (`552 passed`; one Alembic head; frontend typecheck/build and script syntax gates passed).
+
+Next:
+- Pass the complete release gate, deploy the operator and provision all three live regional snapshots before claiming runtime readiness.
+
 ## Sprint 169 Filtered YOLO candidate gate and operator hardening (2026-07-12)
 
 Changed:

docs/DATASET_STRATEGY.md

@@ -124,6 +124,28 @@ reference dataset uses `source_name=grb`, `reference_layer_name=buildings`,
 selection aggregation. Existing viewport and bbox selection APIs remain the
 only interactive delivery path.
 
+### Regional GRB context partition policy
+
+Roads, water and parcels reuse the building operator's 28 resumable source
+partitions and one-dataset-per-theme persistence model. Polygon collections
+use maximum intersection area for cross-boundary ownership; line collections
+use maximum intersection length. Exact ties use the lower NIS code. Source
+identities are collection-qualified because the water theme combines `WTZ`,
+`WLAS` and `WGR`.
+
+The user-facing datasets use `source_name=grb`, `dataset_role=reference`,
+`coverage_scope=kempen-transport-region` and respectively
+`reference_layer_name=roads`, `water` or `parcels`. They are current source
+snapshots with immutable observation dates, not temporal series inferred from
+successive operator runs. Feature counts retain source-object semantics and
+must not be relabelled as road length, water area or legal parcel counts.
+
+Only `scripts/provision_regional_grb_context.py` may fetch these bounded
+artifacts. Application startup, map requests and the public provider registry
+remain fetch-free. All combined artifacts enter PostGIS through the existing
+DatasetService/VectorFeatureService boundary and existing bbox/Area selection
+contracts.
+
 ## User-uploaded raster strategy
 V1 must support controlled local datasets because public raster access and model compatibility can be difficult.
 

docs/DATA_SOURCES.md

@@ -67,6 +67,27 @@ referentiedataset via DatasetService en VectorFeatureService. De publieke GRB
 provider blijft `not_configured`: dit is een bewuste operatorrun en geen live
 download vanuit een browseractie of applicatiestart.
 
+De overige actuele GRB-contextlagen volgen dezelfde begrensde aanpak via
+`scripts/provision_regional_grb_context.py`:
+
+- `roads`: collectie `Wegsegment` (lijngeometrie). Dit zijn wegsegmenten uit
+  de basiskaart, geen verkeersmetingen en geen routeringsnetwerk.
+- `water`: collecties `WTZ` (vlakken), `WLAS` en `WGR` (lijnen). De gemengde
+  objecttelling is geen maat voor wateroppervlakte, debiet of volume.
+- `parcels`: collectie `ADP` (vlakken). Dit is de grafische ligging van het
+  vermoedelijke kadastrale perceel en geen juridisch landmeetkundig bewijs.
+
+Lijnobjecten worden bij gemeentegrenzen toegewezen op basis van de grootste
+intersectielengte, vlakobjecten op basis van de grootste intersectieoppervlakte
+en exacte ties op de laagste NIS-code. De brongeometrie wordt alleen tegen de
+volledige regiogrens gesneden. Bron-ID's worden met hun collectienaam
+gekwalificeerd zodat de drie watercollecties nooit stilzwijgend botsen.
+
+Collectiecatalogus en contract:
+
+- https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections?f=text%2Fhtml
+- https://geo.api.vlaanderen.be/GRB/ogc/features/v1/openapi?f=text%2Fhtml
+
 ### Mol population history
 
 `scripts/provision_mol_population_history.py` imports official Statbel

scripts/README.md

@@ -1325,6 +1325,38 @@ buildings from 879 GRB source pages. Its 28 retained partitions produce one
 `e236fa03-7fce-4b37-bc7d-8bddd4d50346`; an immediate repeat run reused the
 same checksums and Dataset instead of fetching or inserting duplicates.
 
+### Regional Kempen GRB roads, water and parcels
+
+Provision the remaining current GRB context snapshots after the scope
+foundation is present:
+
+```bash
+docker exec -it geointel python3 /app/scripts/provision_regional_grb_context.py \
+  --scope kempen-transport-region \
+  --layers roads water parcels
+```
+
+Each theme has an independent manifest and can be resumed or rerun alone:
+
+```bash
+docker exec -it geointel python3 /app/scripts/provision_regional_grb_context.py \
+  --scope kempen-transport-region --layers roads
+```
+
+Use `--fetch-only` to prepare checksummed artifacts without database writes.
+Artifacts are retained below
+`/app/storage/operator-data/regional-themes/kempen-transport-region/<theme>/<date>`.
+The operator follows all OGC API pagination links, fails rather than hiding a
+safety-limit truncation and persists one normal PostGIS dataset per theme via
+DatasetService. Repeat runs reuse valid partitions, combined artifacts and
+datasets; `--force` refreshes the selected observation date.
+
+The semantic limits are deliberate: `Wegsegment` is road-map context rather
+than traffic or routing data; `WTZ`/`WLAS`/`WGR` are heterogeneous water
+objects rather than a volume metric; `ADP` is the graphical presumed cadastral
+parcel location rather than a legal survey. The public GRB provider remains
+`not_configured`, and no fetch runs from the browser or during startup.
+
 ## Tower deployment
 
 Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime:

scripts/provision_regional_grb_context.py

@@ -0,0 +1,808 @@
+"""Provision regional GRB roads, water and parcel datasets.
+
+Every source layer is fetched in resumable municipality partitions for an
+approved geographic scope. Source identities are assigned to exactly one
+partition, while retained geometries are clipped only to the complete region.
+The resulting artifacts are indexed through DatasetService and
+VectorFeatureService; this operator never writes directly to vector_features.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import os
+import sys
+from dataclasses import dataclass
+from datetime import date
+from pathlib import Path
+from typing import Any, Iterable
+from urllib.parse import urlparse
+from uuid import UUID
+
+import requests
+from shapely.geometry import LineString, MultiLineString, MultiPolygon, Polygon, mapping, shape
+from shapely.ops import unary_union
+from shapely.validation import make_valid
+
+from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope, ScopeMember
+from provision_geographic_scope import fetch_scope_members
+from provision_regional_grb_buildings import (
+    DEFAULT_API_URL,
+    DEFAULT_MAX_FEATURES_PER_MEMBER,
+    DEFAULT_MAX_TOTAL_FEATURES,
+    DEFAULT_OUTPUT_ROOT,
+    DEFAULT_PAGE_LIMIT,
+    DEFAULT_SCOPE_KEY,
+    GEOJSON_CRS,
+    GRB_ATTRIBUTION,
+    bounds_overlap,
+    build_member_geometries,
+    build_source_session,
+    ensure_backend_path,
+    list_paginated_items,
+    next_page_url,
+    observed_at,
+    response_data,
+    reusable_manifest,
+    safe_slug,
+    sha256_file,
+    utc_now,
+    write_json_atomic,
+)
+
+
+GRB_COLLECTION_URL = "https://geo.api.vlaanderen.be/GRB/ogc/features/v1/collections/{collection}/items"
+
+
+@dataclass(frozen=True)
+class CollectionDefinition:
+    name: str
+    geometry_dimension: int
+
+
+@dataclass(frozen=True)
+class LayerDefinition:
+    key: str
+    collections: tuple[CollectionDefinition, ...]
+    reference_layer_name: str
+    layer_type: str
+    geometry_types: tuple[str, ...]
+    metric_label: str
+    limitation_message: str
+
+
+LAYERS = (
+    LayerDefinition(
+        key="roads",
+        collections=(CollectionDefinition("Wegsegment", 1),),
+        reference_layer_name="roads",
+        layer_type="road",
+        geometry_types=("LineString", "MultiLineString"),
+        metric_label="Wegen",
+        limitation_message="GRB Wegsegment represents road-network line segments, not traffic volume or routing suitability.",
+    ),
+    LayerDefinition(
+        key="water",
+        collections=(
+            CollectionDefinition("WTZ", 2),
+            CollectionDefinition("WLAS", 1),
+            CollectionDefinition("WGR", 1),
+        ),
+        reference_layer_name="water",
+        layer_type="water",
+        geometry_types=("LineString", "MultiLineString", "Polygon", "MultiPolygon"),
+        metric_label="Waterobjecten",
+        limitation_message="GRB water combines surface-water polygons and water-related line collections; counts are object counts, not water volume.",
+    ),
+    LayerDefinition(
+        key="parcels",
+        collections=(CollectionDefinition("ADP", 2),),
+        reference_layer_name="parcels",
+        layer_type="parcel",
+        geometry_types=("Polygon", "MultiPolygon"),
+        metric_label="Percelen",
+        limitation_message="GRB ADP is a graphical representation of the presumed cadastral parcel location and is not a legal boundary survey.",
+    ),
+)
+LAYER_BY_KEY = {definition.key: definition for definition in LAYERS}
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Provision municipality-partitioned regional GRB context layers.")
+    parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY)
+    parser.add_argument("--layers", default=",".join(LAYER_BY_KEY), help="Comma-separated subset: roads,water,parcels")
+    parser.add_argument("--observed-date", type=date.fromisoformat, default=date.today())
+    parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL))
+    parser.add_argument(
+        "--output-root",
+        type=Path,
+        default=Path(os.environ.get("GEOINTEL_REGIONAL_THEME_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)),
+    )
+    parser.add_argument("--page-limit", type=int, default=DEFAULT_PAGE_LIMIT)
+    parser.add_argument("--max-features-per-member", type=int, default=DEFAULT_MAX_FEATURES_PER_MEMBER)
+    parser.add_argument("--max-total-features", type=int, default=DEFAULT_MAX_TOTAL_FEATURES)
+    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("--force", action="store_true", help="Refetch every selected municipality partition.")
+    parser.add_argument("--fetch-only", action="store_true", help="Build and validate artifacts without persistence.")
+    return parser.parse_args()
+
+
+def selected_definitions(raw_layers: str) -> list[LayerDefinition]:
+    requested = {item.strip().lower() for item in raw_layers.split(",") if item.strip()}
+    unknown = requested - set(LAYER_BY_KEY)
+    if unknown or not requested:
+        raise ValueError(f"Unsupported layers: {sorted(unknown)}")
+    return [definition for definition in LAYERS if definition.key in requested]
+
+
+def source_session() -> requests.Session:
+    session = build_source_session()
+    session.headers.update({"User-Agent": "GeoIntel-Regional-GRB-Context-Operator/1.0"})
+    return session
+
+
+def geometry_dimension(geometry) -> int:
+    if geometry is None or geometry.is_empty:
+        return -1
+    if "Polygon" in geometry.geom_type:
+        return 2
+    if "LineString" in geometry.geom_type or geometry.geom_type == "LinearRing":
+        return 1
+    if "Point" in geometry.geom_type:
+        return 0
+    if geometry.geom_type == "GeometryCollection":
+        return max((geometry_dimension(part) for part in geometry.geoms), default=-1)
+    return -1
+
+
+def extract_dimension(geometry, expected_dimension: int):
+    if geometry is None or geometry.is_empty:
+        return None
+    if not geometry.is_valid:
+        geometry = make_valid(geometry)
+
+    parts: list[Any] = []
+
+    def collect(candidate) -> None:
+        if candidate is None or candidate.is_empty:
+            return
+        if expected_dimension == 2:
+            if isinstance(candidate, Polygon):
+                parts.append(candidate)
+                return
+            if isinstance(candidate, MultiPolygon):
+                parts.extend(part for part in candidate.geoms if not part.is_empty)
+                return
+        if expected_dimension == 1:
+            if isinstance(candidate, LineString):
+                parts.append(candidate)
+                return
+            if isinstance(candidate, MultiLineString):
+                parts.extend(part for part in candidate.geoms if not part.is_empty)
+                return
+        if hasattr(candidate, "geoms"):
+            for part in candidate.geoms:
+                collect(part)
+
+    collect(geometry)
+    if not parts:
+        return None
+    normalized = unary_union(parts)
+    if normalized.is_empty:
+        return None
+    if not normalized.is_valid:
+        normalized = make_valid(normalized)
+    if normalized.is_empty or not normalized.is_valid or geometry_dimension(normalized) != expected_dimension:
+        return None
+    return normalized
+
+
+def normalized_geometry(payload: dict[str, Any] | None, expected_dimension: int):
+    if not payload:
+        return None
+    return extract_dimension(shape(payload), expected_dimension)
+
+
+def geometry_measure(geometry, expected_dimension: int) -> float:
+    candidate = extract_dimension(geometry, expected_dimension)
+    if candidate is None:
+        return 0.0
+    if expected_dimension == 2:
+        return float(candidate.area)
+    if expected_dimension == 1:
+        return float(candidate.length)
+    return 1.0
+
+
+def assign_owner_nis(
+    source_geometry,
+    members: list[tuple[ScopeMember, Any]],
+    *,
+    expected_dimension: int,
+) -> str | None:
+    candidates: list[tuple[float, str]] = []
+    source_bounds = source_geometry.bounds
+    for member, boundary in members:
+        if not bounds_overlap(source_bounds, boundary.bounds) or not source_geometry.intersects(boundary):
+            continue
+        score = geometry_measure(source_geometry.intersection(boundary), expected_dimension)
+        if score > 0:
+            candidates.append((score, member.nis_code))
+    if not candidates:
+        return None
+    candidates.sort(key=lambda item: (-item[0], item[1]))
+    return candidates[0][1]
+
+
+def iter_collection_pages(
+    session: requests.Session,
+    collection: CollectionDefinition,
+    bounds: tuple[float, float, float, float],
+    *,
+    page_limit: int,
+    timeout: int,
+) -> Iterable[tuple[CollectionDefinition, dict[str, Any], str]]:
+    params = {
+        "f": "application/geo+json",
+        "limit": str(page_limit),
+        "bbox": ",".join(f"{value:.8f}" for value in bounds),
+    }
+    url: str | None = GRB_COLLECTION_URL.format(collection=collection.name)
+    seen_urls: set[str] = set()
+    first_request = True
+    while url:
+        if url in seen_urls:
+            raise RuntimeError(f"GRB pagination loop detected for {collection.name}: {url}")
+        seen_urls.add(url)
+        response = session.get(url, params=params if first_request else None, timeout=timeout)
+        first_request = False
+        response.raise_for_status()
+        payload = response.json()
+        if payload.get("type") != "FeatureCollection":
+            raise RuntimeError(f"GRB {collection.name} returned a non-FeatureCollection response")
+        yield collection, payload, response.url
+        url = next_page_url(payload)
+
+
+def iter_layer_pages(
+    session: requests.Session,
+    definition: LayerDefinition,
+    bounds: tuple[float, float, float, float],
+    *,
+    page_limit: int,
+    timeout: int,
+) -> Iterable[tuple[CollectionDefinition, dict[str, Any], str]]:
+    for collection in definition.collections:
+        yield from iter_collection_pages(
+            session,
+            collection,
+            bounds,
+            page_limit=page_limit,
+            timeout=timeout,
+        )
+
+
+def build_partition_features(
+    pages: Iterable[tuple[CollectionDefinition, dict[str, Any], str]],
+    *,
+    definition: LayerDefinition,
+    member: ScopeMember,
+    members: list[tuple[ScopeMember, Any]],
+    regional_boundary,
+    scope: GeographicScope,
+    max_features: int,
+) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+    features: list[dict[str, Any]] = []
+    source_urls: list[str] = []
+    seen_ids: set[str] = set()
+    pages_by_collection: dict[str, int] = {collection.name: 0 for collection in definition.collections}
+    features_by_collection: dict[str, int] = {collection.name: 0 for collection in definition.collections}
+    geometry_types: dict[str, int] = {}
+    bbox_feature_count = 0
+    assigned_elsewhere_count = 0
+    outside_scope_count = 0
+    clipped_to_scope_count = 0
+    member_boundary = next(boundary for candidate, boundary in members if candidate.nis_code == member.nis_code)
+
+    for collection, payload, source_url in pages:
+        source_urls.append(source_url)
+        pages_by_collection[collection.name] += 1
+        for source_feature in payload.get("features") or []:
+            bbox_feature_count += 1
+            raw_id = str(source_feature.get("id") or "")
+            if not raw_id:
+                raw_id = hashlib.sha256(
+                    json.dumps(source_feature.get("geometry"), sort_keys=True).encode("utf-8")
+                ).hexdigest()
+            feature_id = f"{collection.name}:{raw_id}"
+            if feature_id in seen_ids:
+                continue
+            seen_ids.add(feature_id)
+
+            source_geometry = normalized_geometry(source_feature.get("geometry"), collection.geometry_dimension)
+            if source_geometry is None or not source_geometry.intersects(regional_boundary):
+                outside_scope_count += 1
+                continue
+            owner_nis = (
+                member.nis_code
+                if member_boundary.covers(source_geometry)
+                else assign_owner_nis(
+                    source_geometry,
+                    members,
+                    expected_dimension=collection.geometry_dimension,
+                )
+            )
+            if owner_nis != member.nis_code:
+                assigned_elsewhere_count += 1
+                continue
+
+            clipped = not regional_boundary.covers(source_geometry)
+            retained_geometry = source_geometry
+            if clipped:
+                retained_geometry = extract_dimension(
+                    source_geometry.intersection(regional_boundary),
+                    collection.geometry_dimension,
+                )
+                clipped_to_scope_count += 1
+            if retained_geometry is None:
+                outside_scope_count += 1
+                continue
+            if len(features) >= max_features:
+                raise RuntimeError(
+                    f"{member.name} {definition.key} exceeds --max-features-per-member={max_features}; "
+                    "refusing truncated output"
+                )
+
+            properties = dict(source_feature.get("properties") or {})
+            properties.update(
+                {
+                    "source_name": "grb",
+                    "source_collection": collection.name,
+                    "source_feature_id": feature_id,
+                    "reference_layer_name": definition.reference_layer_name,
+                    "layer_type": definition.layer_type,
+                    "theme": definition.key,
+                    "authority_level": "authoritative",
+                    "coverage_scope": scope.key,
+                    "scope_type": scope.scope_type,
+                    "partition_scope": "municipality",
+                    "partition_municipality": member.name,
+                    "partition_nis_code": member.nis_code,
+                    "partition_assignment": "maximum_same_dimension_intersection",
+                    "clipped_to_regional_scope": clipped,
+                    "attribution": GRB_ATTRIBUTION,
+                }
+            )
+            features.append(
+                {
+                    "type": "Feature",
+                    "id": feature_id,
+                    "geometry": mapping(retained_geometry),
+                    "properties": properties,
+                }
+            )
+            features_by_collection[collection.name] += 1
+            geometry_types[retained_geometry.geom_type] = geometry_types.get(retained_geometry.geom_type, 0) + 1
+
+    if not features:
+        raise RuntimeError(f"No GRB {definition.key} features were assigned to {member.name}")
+    return features, {
+        "municipality": member.name,
+        "nis_code": member.nis_code,
+        "pages_fetched": len(source_urls),
+        "pages_by_collection": pages_by_collection,
+        "source_urls": source_urls,
+        "bbox_feature_count": bbox_feature_count,
+        "feature_count": len(features),
+        "features_by_collection": features_by_collection,
+        "geometry_types": geometry_types,
+        "assigned_elsewhere_count": assigned_elsewhere_count,
+        "outside_scope_count": outside_scope_count,
+        "clipped_to_scope_count": clipped_to_scope_count,
+        "reference_truncated": False,
+    }
+
+
+def partition_filename(definition: LayerDefinition, member: ScopeMember, observed_date: date) -> str:
+    return f"{member.nis_code}_{safe_slug(member.name)}_grb_{definition.key}_{observed_date.isoformat()}.geojson"
+
+
+def combined_filename(definition: LayerDefinition, scope: GeographicScope, observed_date: date) -> str:
+    return f"grb_{definition.key}_{scope.key.replace('-', '_')}_{observed_date.isoformat()}.geojson"
+
+
+def write_partition(
+    path: Path,
+    *,
+    definition: LayerDefinition,
+    scope: GeographicScope,
+    member: ScopeMember,
+    features: list[dict[str, Any]],
+    generated_at: str,
+) -> None:
+    write_json_atomic(
+        path,
+        {
+            "type": "FeatureCollection",
+            "name": f"GRB {definition.key} - {member.name} partition of {scope.display_name}",
+            "crs": GEOJSON_CRS,
+            "features": features,
+            "source": f"Digitaal Vlaanderen GRB OGC API collections {', '.join(item.name for item in definition.collections)}",
+            "source_urls": [GRB_COLLECTION_URL.format(collection=item.name) for item in definition.collections],
+            "attribution": GRB_ATTRIBUTION,
+            "coverage_scope": scope.key,
+            "partition_municipality": member.name,
+            "partition_nis_code": member.nis_code,
+            "generated_at": generated_at,
+        },
+    )
+
+
+def write_combined_artifact(
+    path: Path,
+    *,
+    definition: LayerDefinition,
+    scope: GeographicScope,
+    observed_date: date,
+    partition_paths: list[Path],
+    expected_feature_count: int,
+) -> dict[str, Any]:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(f"{path.suffix}.partial")
+    header = {
+        "type": "FeatureCollection",
+        "name": f"GRB {definition.key} - {scope.display_name}",
+        "crs": GEOJSON_CRS,
+        "source": f"Digitaal Vlaanderen GRB OGC API collections {', '.join(item.name for item in definition.collections)}",
+        "source_urls": [GRB_COLLECTION_URL.format(collection=item.name) for item in definition.collections],
+        "attribution": GRB_ATTRIBUTION,
+        "coverage_scope": scope.key,
+        "scope_type": scope.scope_type,
+        "partition_count": len(partition_paths),
+        "partition_strategy": "municipality_bbox_maximum_same_dimension_intersection",
+        "observed_at": observed_date.isoformat(),
+    }
+    encoded_header = json.dumps(header, ensure_ascii=False, separators=(",", ":"))
+    seen_ids: set[str] = set()
+    written = 0
+    first = True
+    with temporary.open("w", encoding="utf-8", newline="") as output:
+        output.write(encoded_header[:-1])
+        output.write(',"features":[')
+        for partition_path in partition_paths:
+            payload = json.loads(partition_path.read_text(encoding="utf-8"))
+            if payload.get("type") != "FeatureCollection" or not isinstance(payload.get("features"), list):
+                raise RuntimeError(f"Invalid partition artifact: {partition_path}")
+            for feature in payload["features"]:
+                feature_id = str(feature.get("id") or (feature.get("properties") or {}).get("source_feature_id") or "")
+                if not feature_id:
+                    raise RuntimeError(f"Partition feature without source identity in {partition_path.name}")
+                if feature_id in seen_ids:
+                    raise RuntimeError(f"Duplicate regional source feature {feature_id} in {partition_path.name}")
+                seen_ids.add(feature_id)
+                if not first:
+                    output.write(",")
+                output.write(json.dumps(feature, ensure_ascii=False, separators=(",", ":")))
+                first = False
+                written += 1
+        output.write("]}")
+    if written != expected_feature_count:
+        temporary.unlink(missing_ok=True)
+        raise RuntimeError(f"Expected {expected_feature_count} combined features, wrote {written}")
+    temporary.replace(path)
+    return {"feature_count": written, "size_bytes": path.stat().st_size, "sha256": sha256_file(path)}
+
+
+def prepare_layer_artifacts(
+    args: argparse.Namespace,
+    scope: GeographicScope,
+    definition: LayerDefinition,
+) -> tuple[Path, list[Path], Path, dict[str, Any]]:
+    observation_dir = args.output_root / scope.key / definition.key / args.observed_date.isoformat()
+    partition_dir = observation_dir / "partitions"
+    artifact_path = observation_dir / combined_filename(definition, scope, args.observed_date)
+    manifest_path = observation_dir / f"regional_{definition.key}_manifest.json"
+    partition_dir.mkdir(parents=True, exist_ok=True)
+
+    if not args.force:
+        existing = reusable_manifest(manifest_path, artifact_path, partition_dir, len(scope.members))
+        if existing:
+            paths = [partition_dir / summary["filename"] for summary in existing["partitions"]]
+            return artifact_path, paths, manifest_path, existing
+
+    existing_manifest: dict[str, Any] = {}
+    if manifest_path.is_file() and not args.force:
+        existing_manifest = json.loads(manifest_path.read_text(encoding="utf-8"))
+    existing_by_nis = {
+        str(item.get("nis_code")): item
+        for item in existing_manifest.get("partitions") or []
+        if isinstance(item, dict)
+    }
+    generated_at = utc_now()
+    with source_session() as session:
+        source_features, vrbg_source_url = fetch_scope_members(session, scope, args.request_timeout)
+        members, regional_boundary = build_member_geometries(scope, source_features)
+        summaries: list[dict[str, Any]] = []
+        for member, boundary in members:
+            path = partition_dir / partition_filename(definition, member, args.observed_date)
+            reusable = existing_by_nis.get(member.nis_code)
+            if (
+                reusable
+                and path.is_file()
+                and reusable.get("filename") == path.name
+                and reusable.get("sha256") == sha256_file(path)
+            ):
+                summaries.append(reusable)
+                continue
+            features, summary = build_partition_features(
+                iter_layer_pages(
+                    session,
+                    definition,
+                    boundary.bounds,
+                    page_limit=args.page_limit,
+                    timeout=args.request_timeout,
+                ),
+                definition=definition,
+                member=member,
+                members=members,
+                regional_boundary=regional_boundary,
+                scope=scope,
+                max_features=args.max_features_per_member,
+            )
+            write_partition(
+                path,
+                definition=definition,
+                scope=scope,
+                member=member,
+                features=features,
+                generated_at=generated_at,
+            )
+            summary.update({"filename": path.name, "size_bytes": path.stat().st_size, "sha256": sha256_file(path)})
+            summaries.append(summary)
+            write_json_atomic(
+                manifest_path,
+                {
+                    "schema_version": 1,
+                    "status": "in_progress",
+                    "scope": scope.key,
+                    "theme": definition.key,
+                    "observed_at": args.observed_date.isoformat(),
+                    "generated_at": generated_at,
+                    "vrbg_source_url": vrbg_source_url,
+                    "grb_collections": [item.name for item in definition.collections],
+                    "partitions": summaries,
+                },
+                pretty=True,
+            )
+
+    total_features = sum(int(summary["feature_count"]) for summary in summaries)
+    if total_features > args.max_total_features:
+        raise RuntimeError(
+            f"Regional {definition.key} count {total_features} exceeds --max-total-features={args.max_total_features}"
+        )
+    partition_paths = [partition_dir / summary["filename"] for summary in summaries]
+    artifact = write_combined_artifact(
+        artifact_path,
+        definition=definition,
+        scope=scope,
+        observed_date=args.observed_date,
+        partition_paths=partition_paths,
+        expected_feature_count=total_features,
+    )
+    aggregate_geometry_types: dict[str, int] = {}
+    for summary in summaries:
+        for geometry_type, count in (summary.get("geometry_types") or {}).items():
+            aggregate_geometry_types[geometry_type] = aggregate_geometry_types.get(geometry_type, 0) + int(count)
+    manifest = {
+        "schema_version": 1,
+        "status": "complete",
+        "scope": scope.key,
+        "scope_type": scope.scope_type,
+        "scope_authority_url": scope.authority_url,
+        "scope_limitation": scope.limitation_message,
+        "theme": definition.key,
+        "reference_layer_name": definition.reference_layer_name,
+        "observed_at": args.observed_date.isoformat(),
+        "generated_at": generated_at,
+        "member_count": len(scope.members),
+        "feature_count": total_features,
+        "geometry_types": aggregate_geometry_types,
+        "reference_truncated": False,
+        "partition_strategy": "municipality_bbox_maximum_same_dimension_intersection",
+        "partition_assignment_rule": "largest same-dimension intersection measure; NIS code resolves exact ties",
+        "vrbg_source_url": vrbg_source_url,
+        "grb_collections": [item.name for item in definition.collections],
+        "grb_source_urls": [GRB_COLLECTION_URL.format(collection=item.name) for item in definition.collections],
+        "artifact_filename": artifact_path.name,
+        "artifact_size_bytes": artifact["size_bytes"],
+        "artifact_sha256": artifact["sha256"],
+        "bounds_json": {
+            "min_x": float(regional_boundary.bounds[0]),
+            "min_y": float(regional_boundary.bounds[1]),
+            "max_x": float(regional_boundary.bounds[2]),
+            "max_y": float(regional_boundary.bounds[3]),
+        },
+        "limitation_message": definition.limitation_message,
+        "partitions": summaries,
+        "attribution": GRB_ATTRIBUTION,
+    }
+    write_json_atomic(manifest_path, manifest, pretty=True)
+    return artifact_path, partition_paths, manifest_path, manifest
+
+
+def provision_dataset(
+    args: argparse.Namespace,
+    scope: GeographicScope,
+    definition: LayerDefinition,
+    artifact_path: Path,
+    partition_paths: list[Path],
+    manifest_path: Path,
+    manifest: dict[str, Any],
+) -> dict[str, Any]:
+    parsed_base = urlparse(args.base_url)
+    if parsed_base.hostname not in {"127.0.0.1", "localhost", "::1"}:
+        raise RuntimeError("Partitioned service import must run inside the GeoIntel container against its local backend")
+    with requests.Session() as session:
+        projects = list_paginated_items(session, f"{args.base_url.rstrip('/')}/api/v1/projects", args.api_timeout)
+        project = next((item for item in projects if item.get("name") == scope.project_name), None)
+        if not project:
+            raise RuntimeError("Regional scope project is missing; run provision_geographic_scope.py first")
+        project_id = str(project["id"])
+        areas = list_paginated_items(
+            session,
+            f"{args.base_url.rstrip('/')}/api/v1/projects/{project_id}/areas",
+            args.api_timeout,
+        )
+        area = next((item for item in areas if item.get("name") == scope.area_name), None)
+        if not area:
+            raise RuntimeError("Regional scope Area is missing; run provision_geographic_scope.py first")
+        datasets = list_paginated_items(
+            session,
+            f"{args.base_url.rstrip('/')}/api/v1/projects/{project_id}/datasets",
+            args.api_timeout,
+        )
+    existing = next((item for item in datasets if item.get("original_filename") == artifact_path.name), None)
+    if existing:
+        persisted_checksum = (existing.get("provenance_metadata") or {}).get("artifact_sha256")
+        if persisted_checksum and persisted_checksum != manifest["artifact_sha256"]:
+            raise RuntimeError(
+                f"Immutable dataset {artifact_path.name} checksum changed; use a new --observed-date for refreshed GRB data"
+            )
+        return {"dataset_id": str(existing["id"]), "feature_count": existing.get("feature_count"), "reused": True}
+
+    ensure_backend_path()
+    from app.db.session import SessionLocal
+    from app.services.dataset_service import DatasetService
+
+    metadata_json = {
+        "feature_count": manifest["feature_count"],
+        "feature_geometry_count": manifest["feature_count"],
+        "geometry_types": list(manifest.get("geometry_types") or definition.geometry_types),
+        "bounds_json": manifest["bounds_json"],
+        "approximate_area_m2": None,
+        "invalid_features": 0,
+        "z_dimension_feature_count": 0,
+        "canonical_storage_dimension": "2D",
+        "crs": "EPSG:4326",
+        "crs_assumed": False,
+        "extracted_at": manifest["generated_at"],
+    }
+    source_metadata = {
+        "provider": "Digitaal Vlaanderen",
+        "collections": [item.name for item in definition.collections],
+        "authority_level": "authoritative",
+        "theme": definition.key,
+        "layer_type": f"regional_{definition.key}",
+        "coverage_scope": scope.key,
+        "scope_type": scope.scope_type,
+        "scope_authority": scope.authority_name,
+        "scope_authority_url": scope.authority_url,
+        "scope_limitation": scope.limitation_message,
+        "layer_limitation": definition.limitation_message,
+        "member_count": len(scope.members),
+        "member_nis_codes": list(scope.nis_codes),
+        "feature_count": manifest["feature_count"],
+        "partition_count": len(partition_paths),
+        "partition_strategy": manifest["partition_strategy"],
+        "selection_aggregation": {
+            "method": "feature_count",
+            "label": definition.metric_label,
+            "unit": "objecten",
+            "is_estimate": False,
+        },
+        "attribution": GRB_ATTRIBUTION,
+    }
+    provenance_metadata = {
+        "operator_tool": "provision_regional_grb_context.py",
+        "operator_explicit_fetch": True,
+        "manifest_path": str(manifest_path),
+        "source_urls": manifest["grb_source_urls"],
+        "artifact_sha256": manifest["artifact_sha256"],
+        "artifact_size_bytes": manifest["artifact_size_bytes"],
+        "partition_checksums": {summary["nis_code"]: summary["sha256"] for summary in manifest["partitions"]},
+        "partition_assignment_rule": manifest["partition_assignment_rule"],
+        "reference_truncated": False,
+    }
+    with SessionLocal() as db:
+        dataset = DatasetService.import_partitioned_vector_artifact(
+            db,
+            project_id=UUID(project_id),
+            area_id=UUID(str(area["id"])),
+            artifact_path=artifact_path,
+            partition_paths=partition_paths,
+            original_filename=artifact_path.name,
+            source="operator_official_import",
+            dataset_role="reference",
+            source_name="grb",
+            reference_layer_name=definition.reference_layer_name,
+            metadata_json=metadata_json,
+            source_metadata=source_metadata,
+            provenance_metadata=provenance_metadata,
+            temporal_series_key=f"grb:{definition.key}:{scope.key}",
+            observed_at=observed_at(args.observed_date),
+            temporal_granularity="snapshot",
+            source_version=args.observed_date.isoformat(),
+            batch_size=args.batch_size,
+        )
+    if dataset.checksum_sha256 != manifest["artifact_sha256"]:
+        raise RuntimeError("Managed dataset checksum differs from the retained regional artifact")
+    return {"dataset_id": str(dataset.id), "feature_count": dataset.feature_count, "reused": False}
+
+
+def main() -> int:
+    args = parse_args()
+    scope = GEOGRAPHIC_SCOPES[args.scope]
+    try:
+        definitions = selected_definitions(args.layers)
+        if args.page_limit <= 0 or args.max_features_per_member <= 0 or args.max_total_features <= 0:
+            raise ValueError("Page and feature limits must be positive")
+        results: list[dict[str, Any]] = []
+        for definition in definitions:
+            artifact_path, partition_paths, manifest_path, manifest = prepare_layer_artifacts(args, scope, definition)
+            persistence = None if args.fetch_only else provision_dataset(
+                args,
+                scope,
+                definition,
+                artifact_path,
+                partition_paths,
+                manifest_path,
+                manifest,
+            )
+            results.append(
+                {
+                    "theme": definition.key,
+                    "collections": [item.name for item in definition.collections],
+                    "feature_count": manifest["feature_count"],
+                    "artifact_size_bytes": manifest["artifact_size_bytes"],
+                    "artifact_path": str(artifact_path),
+                    "manifest_path": str(manifest_path),
+                    "reference_truncated": manifest["reference_truncated"],
+                    "persistence": persistence,
+                }
+            )
+    except (OSError, RuntimeError, ValueError, KeyError, requests.RequestException) as exc:
+        print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr)
+        return 1
+    print(
+        json.dumps(
+            {
+                "status": "ok",
+                "mode": "fetch_only" if args.fetch_only else "provisioned",
+                "scope": scope.key,
+                "observed_at": args.observed_date.isoformat(),
+                "member_count": len(scope.members),
+                "layers": results,
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/run_readiness_check.sh

@@ -50,6 +50,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_official_landuse_timeseries.py
 ${PYTHON_BIN} -m py_compile scripts/geographic_scopes.py
 ${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/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