feat: add governed agricultural parcel history

CHANGELOG.md

@@ -7,6 +7,20 @@
 
 # Changelog
 
+## Sprint 205 Governed agricultural-use parcel history (2026-07-15)
+
+- Added an explicit ALZ operator for the definitive 2008-2025 annual
+  agricultural-use parcel archives, with a fixed source allowlist, streamed
+  size limit, ZIP safety checks and EPSG:31370 schema validation.
+- Retained official archives, annual crop-code lists and checksum manifests;
+  exact scope clipping produces ordinary Dataset/vector_feature persistence
+  only through the existing upload service.
+- Added a separate Agriculture map theme with exact declared-use hectares and
+  server-owned official main-crop-group hectare metrics.
+- Added a scope-specific 18-edition temporal series while explicitly disabling
+  parcel-level lineage and excluding the provisional current campaign.
+- Added focused source, clipping, metric, pagination, packaging and UI tests.
+
 ## Sprint 204 Governed BWK and Natura 2000 for Mol (2026-07-15)
 
 - Added an explicit operator for the official INBO BWK/Natura 2000 state 2025

backend/README.md

@@ -1112,6 +1112,26 @@ the assistant cannot turn 2D water geometry into volume. GeoIntel rejects an
 answer when Ollama reports `done_reason=length`, so a visibly truncated sentence
 is never presented as a complete result.
 
+## Agricultural-use parcel history
+
+Prepare all definitive 2008-2025 regional editions without database writes:
+
+```bash
+docker exec geointel python /app/scripts/provision_agricultural_parcel_history.py --fetch-only
+```
+
+Import the checked artifacts through the canonical Dataset upload route:
+
+```bash
+docker exec geointel python /app/scripts/provision_agricultural_parcel_history.py
+```
+
+Use `--scope mol`, `--years 2008,2019,2025` or `--force` only as explicit
+operator choices. The default scope is the persisted 28-municipality Kempen
+transport region. Every annual source ZIP and crop code list remains under the
+storage volume. PostGIS computes exact hectares for drawn rectangles and
+persisted Areas; parcel identities are deliberately unavailable for lineage.
+
 ## Helpful repository scripts
 
 - `bash scripts/backend_install.sh`

backend/app/services/vector_feature_service.py

@@ -25,6 +25,7 @@ FULL_AREA_CLIPPED_OPERATOR_TOOLS = {
     "provision_regional_grb_context.py",
     "provision_waterinfo_station_history.py",
     "provision_mol_bwk_natura2000.py",
+    "provision_agricultural_parcel_history.py",
 }
 
 
@@ -86,6 +87,7 @@ SEMANTIC_SELECTION_METRICS: dict[str, tuple[dict[str, Any], ...]] = {
         },
     ),
     "nature_value": (),
+    "agriculture": (),
 }
 
 SEMANTIC_COUNT_LABELS = {
@@ -96,6 +98,7 @@ SEMANTIC_COUNT_LABELS = {
     "roads": "Wegsegmenten",
     "parcels": "Percelen",
     "nature_value": "BWK-kaartvlakken",
+    "agriculture": "Landbouwgebruikspercelen",
 }
 
 
@@ -121,6 +124,9 @@ class VectorFeatureService:
             "biodiversity": "nature_value",
             "bwk": "nature_value",
             "natura2000": "nature_value",
+            "agricultural": "agriculture",
+            "landbouw": "agriculture",
+            "landbouwgebruik": "agriculture",
         }
         for candidate in candidates:
             if not isinstance(candidate, str) or not candidate.strip():

backend/tests/test_sprint205_agricultural_parcel_history.py

@@ -0,0 +1,275 @@
+from __future__ import annotations
+
+import importlib.util
+import json
+import sys
+import zipfile
+from pathlib import Path
+from uuid import uuid4
+
+import pytest
+from shapely.geometry import box, shape
+from shapely.ops import transform as transform_geometry
+
+from app.models import Dataset
+from app.schemas.operations import VectorSelectionSummary
+from app.services.vector_feature_service import VectorFeatureService
+
+
+ROOT = Path(__file__).resolve().parents[2]
+BBOX = {"min_x": 5.0, "min_y": 51.1, "max_x": 5.3, "max_y": 51.4, "crs": "EPSG:4326"}
+
+
+def load_operator():
+    script_path = ROOT / "scripts" / "provision_agricultural_parcel_history.py"
+    spec = importlib.util.spec_from_file_location("agricultural_parcel_history_operator", script_path)
+    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
+
+
+class FakeRow:
+    def __init__(self, geometry, values: dict):  # noqa: ANN001
+        self.geometry = geometry
+        self.values = values
+
+    def __getitem__(self, key):  # noqa: ANN001
+        return self.values[key]
+
+
+class FakeFrame:
+    def __init__(self, rows: list[FakeRow], columns: list[str]):
+        self.rows = rows
+        self.columns = columns
+
+    def iterrows(self):
+        return iter(enumerate(self.rows))
+
+
+class ScalarQuery:
+    def __init__(self, value: float):
+        self.value = value
+
+    def filter(self, *args):  # noqa: ANN002, ARG002
+        return self
+
+    def scalar(self):
+        return self.value
+
+
+class SequenceScalarSession:
+    def __init__(self, values: list[float]):
+        self.values = iter(values)
+
+    def query(self, *args):  # noqa: ANN002, ARG002
+        return ScalarQuery(next(self.values))
+
+
+class ApiResponse:
+    ok = True
+    status_code = 200
+    text = ""
+
+    def __init__(self, data: dict):
+        self.data = data
+
+    def json(self):
+        return {"data": self.data}
+
+
+class PaginatedApiSession:
+    def __init__(self):
+        self.offsets: list[int] = []
+
+    def get(self, url, *, params, timeout):  # noqa: ANN001, ARG002
+        self.offsets.append(params["offset"])
+        if params["offset"] == 0:
+            return ApiResponse({"items": [{"id": index} for index in range(200)], "total": 201})
+        return ApiResponse({"items": [{"id": 200}], "total": 201})
+
+
+def test_only_definitive_2008_through_2025_archives_are_allowed() -> None:
+    module = load_operator()
+
+    assert module.SUPPORTED_YEARS == tuple(range(2008, 2026))
+    assert 2026 not in module.ARCHIVE_URLS
+    assert module.ARCHIVE_URLS[2025].endswith("agpa_2025_2026-05-13_public.zip")
+    assert all(url.startswith("https://www.landbouwvlaanderen.be/bestanden/gis/agpa_") for url in module.ARCHIVE_URLS.values())
+    with pytest.raises(ValueError, match="Supported definitive years"):
+        module.parse_years("2025,2026")
+
+
+def test_canonical_api_collection_reader_respects_200_item_limit_and_paginates() -> None:
+    module = load_operator()
+    session = PaginatedApiSession()
+
+    items = module.api_items(session, "http://geointel/api/v1/projects/project-id/datasets", 30)
+
+    assert len(items) == 201
+    assert session.offsets == [0, 200]
+
+
+def test_archive_requires_exactly_one_safe_geopackage(tmp_path: Path) -> None:
+    module = load_operator()
+    valid = tmp_path / "valid.zip"
+    with zipfile.ZipFile(valid, "w") as archive:
+        archive.writestr("agpa_2025.gpkg", b"source")
+        archive.writestr("metadata.pdf", b"metadata")
+    assert module.archive_geopackage_member(valid) == "agpa_2025.gpkg"
+
+    unsafe = tmp_path / "unsafe.zip"
+    with zipfile.ZipFile(unsafe, "w") as archive:
+        archive.writestr("nested/agpa_2025.gpkg", b"source")
+    with pytest.raises(RuntimeError, match="unsafe"):
+        module.archive_geopackage_member(unsafe)
+
+    ambiguous = tmp_path / "ambiguous.zip"
+    with zipfile.ZipFile(ambiguous, "w") as archive:
+        archive.writestr("one.gpkg", b"one")
+        archive.writestr("two.gpkg", b"two")
+    with pytest.raises(RuntimeError, match="exactly one"):
+        module.archive_geopackage_member(ambiguous)
+
+
+def test_crop_code_list_preserves_year_specific_titles_and_reports_conflicts() -> None:
+    module = load_operator()
+    result = module.build_crop_code_list(
+        [
+            {"maincrop_code": "201", "maincrop_title": "Mais", "maincropgroup_title": "Mais"},
+            {"maincrop_code": "201", "maincrop_title": "Korrelmais", "maincropgroup_title": "Mais"},
+            {"maincrop_code": "901", "maincrop_title": "Grasland", "maincropgroup_title": "Grasland"},
+        ],
+        year=2025,
+    )
+
+    assert result["year"] == 2025
+    assert len(result["crop_entries"]) == 3
+    assert result["code_title_conflicts"] == {"201": ["Korrelmais", "Mais"]}
+    assert "maincropgroup_title" in result["historical_comparison_rule"]
+    assert module.normalized_group_title("Maïs") == "maize"
+
+
+def test_features_are_exactly_clipped_in_lambert72_and_keep_source_fields() -> None:
+    module = load_operator()
+    boundary_wgs84 = box(5.10, 51.20, 5.11, 51.21)
+    boundary_lambert72 = transform_geometry(module.TO_LAMBERT72.transform, boundary_wgs84)
+    source_geometry = transform_geometry(module.TO_LAMBERT72.transform, box(5.095, 51.195, 5.105, 51.205))
+    values = {
+        "agpakey": "2025-42",
+        "parcelnumber": "42",
+        "area_ha": 1.25,
+        "maincrop_code": "201",
+        "maincrop_title": "Korrelmais",
+        "maincropgroup_title": "Maïs",
+        "geometry": source_geometry,
+    }
+    frame = FakeFrame([FakeRow(source_geometry, values)], list(values))
+
+    features, summary = module.normalize_frame(
+        frame,
+        year=2025,
+        boundary_lambert72=boundary_lambert72,
+        max_features=10,
+    )
+
+    assert summary["feature_count"] == 1
+    assert summary["clipped_feature_count"] == 1
+    feature = features[0]
+    assert feature["id"] == "alz:2025:2025-42"
+    assert shape(feature["geometry"]).difference(boundary_wgs84.buffer(1e-7)).area < 1e-12
+    properties = feature["properties"]
+    assert properties["maincrop_title"] == "Korrelmais"
+    assert properties["main_crop_group_key"] == "maize"
+    assert properties["geometry_was_clipped"] is True
+    assert properties["historical_parcel_identity_stable"] is False
+    assert properties["clipped_area_ha"] < properties["source_geometry_area_ha"]
+
+
+def test_duplicate_annual_source_identity_fails_closed() -> None:
+    module = load_operator()
+    boundary = box(100_000, 200_000, 101_000, 201_000)
+    values = {"agpakey": "same", "maincropgroup_title": "Grasland", "geometry": boundary}
+    frame = FakeFrame([FakeRow(boundary, values), FakeRow(boundary, values)], list(values))
+
+    with pytest.raises(RuntimeError, match="duplicate agpakey"):
+        module.normalize_frame(frame, year=2025, boundary_lambert72=boundary, max_features=10)
+
+
+def test_agriculture_summary_returns_grouped_hectares_without_parcel_lineage_claim() -> None:
+    module = load_operator()
+    dataset = Dataset(
+        id=uuid4(),
+        project_id=uuid4(),
+        name="agricultural_use_parcels_2025.geojson",
+        dataset_type="vector",
+        dataset_role="reference",
+        source_name=module.SOURCE_NAME,
+        reference_layer_name="agriculture",
+        source_metadata={
+            "theme": "agriculture",
+            "semantic_metrics": False,
+            "selection_aggregation": {
+                "metric_key": "declared_agricultural_use_area",
+                "method": "intersection_area",
+                "label": "Aangegeven gebruiksoppervlakte",
+                "unit": "ha",
+                "geometry_dimension": 2,
+            },
+            "selection_metrics": module.selection_metrics(),
+        },
+    )
+    square_metres = [150_000.0, 40_000.0, 30_000.0, 20_000.0, 10_000.0, 5_000.0, 4_000.0, 3_000.0, 2_000.0, 1_000.0, 500.0, 250.0]
+
+    result = VectorFeatureService.summarize_features_by_bbox(
+        SequenceScalarSession(square_metres),
+        dataset=dataset,
+        bbox=BBOX,
+        total_feature_count=321,
+        full_dataset_area=True,
+    )
+
+    assert result["primary_metric_key"] == "declared_agricultural_use_area"
+    assert result["metric_value"] == 15.0
+    metrics = {item["metric_key"]: item for item in result["metrics"]}
+    assert metrics["grassland_area"]["metric_value"] == 4.0
+    assert metrics["maize_area"]["metric_value"] == 3.0
+    assert metrics["agricultural_water_area"]["metric_value"] == 0.025
+    assert metrics["feature_count"]["metric_value"] == 321
+    assert "perceelidentiteiten" in metrics["grassland_area"]["warning"]
+    VectorSelectionSummary(**result)
+
+
+def test_operator_uses_canonical_upload_and_is_packaged_for_runtime() -> None:
+    operator = (ROOT / "scripts" / "provision_agricultural_parcel_history.py").read_text(encoding="utf-8")
+    service = (ROOT / "backend" / "app" / "services" / "vector_feature_service.py").read_text(encoding="utf-8")
+    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")
+    map_workspace = (ROOT / "frontend" / "src" / "components" / "map" / "MapWorkspace.tsx").read_text(encoding="utf-8")
+    source_catalog = (ROOT / "frontend" / "src" / "components" / "datasets" / "SourceCatalogPanel.tsx").read_text(encoding="utf-8")
+
+    assert "/datasets/upload" in operator
+    assert "VectorFeature" not in operator
+    assert "INSERT INTO vector_features" not in operator
+    assert "geo.api.vlaanderen.be/Landbgebrperc" not in operator
+    assert '"provision_agricultural_parcel_history.py"' in service
+    assert "COPY scripts/provision_agricultural_parcel_history.py" in dockerfile
+    assert "py_compile scripts/provision_agricultural_parcel_history.py" in readiness
+    assert "id: 'agriculture'" in map_workspace
+    assert "Landbouwgebruikspercelen" in source_catalog
+
+
+def test_upload_contract_is_annual_definitive_and_scope_specific(tmp_path: Path) -> None:
+    module = load_operator()
+    scope = module.GEOGRAPHIC_SCOPES["mol"]
+    assert module.series_key(scope) == "alz:agricultural-use-parcels:mol"
+    metrics = module.selection_metrics()
+    assert {item["metric_key"] for item in metrics} >= {"grassland_area", "maize_area", "agricultural_water_area"}
+    assert all(item["method"] == "intersection_area" for item in metrics)
+    assert all(item["filter_property"] == "main_crop_group_key" for item in metrics)
+
+    paths = module.artifact_paths(tmp_path, scope.key, 2025)
+    assert paths["archive"].name == "agpa_2025_2026-05-13_public.zip"
+    assert paths["artifact"].name == "agricultural_use_parcels_2025_mol.geojson"

deploy/unraid/Dockerfile.all-in-one

@@ -79,6 +79,7 @@ COPY scripts/provision_mol_historical_landuse.py /app/scripts/provision_mol_hist
 COPY scripts/provision_official_landuse_timeseries.py /app/scripts/provision_official_landuse_timeseries.py
 COPY scripts/provision_waterinfo_station_history.py /app/scripts/provision_waterinfo_station_history.py
 COPY scripts/provision_mol_bwk_natura2000.py /app/scripts/provision_mol_bwk_natura2000.py
+COPY scripts/provision_agricultural_parcel_history.py /app/scripts/provision_agricultural_parcel_history.py
 COPY scripts/provision_regional_timeseries.py /app/scripts/provision_regional_timeseries.py
 COPY scripts/geographic_scopes.py /app/scripts/geographic_scopes.py
 COPY scripts/provision_geographic_scope.py /app/scripts/provision_geographic_scope.py

docs/API_CONTRACTS.md

@@ -1632,6 +1632,13 @@ Added/removed/modified object changes are calculated only when source
 provenance declares stable feature identities; otherwise
 `object_changes.available=false` and no object history is inferred.
 
+For `reference_layer_name=agriculture`, the primary map metric is exact
+intersected declared-use area in hectares. Supplemental metrics use
+server-owned filters on the normalized official main-crop group. Annual ALZ
+Datasets share one scope-specific temporal series, but declare
+`identity_stable=false`; their temporal response compares area totals and
+returns no parcel-level added/removed/modified claims.
+
 ## Local GeoIntel assistant
 
 The assistant is an optional read-only language interface over persisted

docs/DATA_SOURCES.md

@@ -217,9 +217,6 @@ until a governed operator import, provenance record and validation pass exist.
 - Biologische Waarderingskaart / Natura 2000 (INBO), state 2025: suitable for
   habitat, biotope and ecological-value analysis, not a continuous annual
   series.
-- Agricultural-use parcels (Agentschap Landbouw en Zeevisserij): annual files
-  suitable for crop and agricultural-surface evolution after schema/version
-  harmonization.
 - Buildings and Addresses Register (Digitaal Vlaanderen): continuously updated
   building status, life cycle and address linkage; complementary to GRB
   geometry and not yet imported.
@@ -248,6 +245,32 @@ especially outside Habitats Directive areas, can still be based on older
 mapping. GeoIntel therefore preserves origin fields and never presents this
 single edition as a historical trend.
 
+## Governed annual agricultural-use parcels
+
+`scripts/provision_agricultural_parcel_history.py` uses the official annual
+download archives published by Agentschap Landbouw en Zeevisserij. Only the
+definitive editions 2008-2025 are allowed. The provisional current-campaign
+snapshot is deliberately excluded, and the retiring Digitaal Vlaanderen
+`Landbgebrperc` WFS is not used as the long-term source path.
+
+Each official ZIP, SHA256 checksum, full-edition crop code list and manifest is
+retained. The temporary GeoPackage is validated as EPSG:31370 with the stable
+annual field set, read through the GIS optional dependencies and deleted after
+the normalized GeoJSON has been built. Geometry is clipped exactly against the
+persisted scope Area in Lambert 72 and imported only through DatasetService.
+
+The regional default creates the series
+`alz:agricultural-use-parcels:kempen-transport-region`; `--scope mol` creates an
+independent Mol series. Selection and evolution expose exact intersected
+hectares for total declared use and official main-crop groups. Individual
+parcel additions/removals are unavailable because parcel identity is not
+stable across campaign years.
+
+Official catalogues:
+
+- https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen
+- https://www.vlaanderen.be/datavindplaats/catalogus/open-geodata-landbouwgebruikspercelen
+
 ## Governed Waterinfo station history
 
 `scripts/provision_waterinfo_station_history.py` uses the public Waterinfo

docs/DATA_SPECIFICATION.md

@@ -228,6 +228,27 @@ an estimate warning, especially for partial rectangle intersections. State
 2025 is a product edition and must not be presented as a uniform 2025 field
 survey or an annual time series.
 
+### Annual agricultural-use parcels
+
+Definitive Agentschap Landbouw en Zeevisserij editions from 2008 through 2025
+are stored as separate annual vector Datasets. The retained source is the
+official ZIP archive containing a Belgian Lambert 72 GeoPackage. Queryable
+geometry is clipped against the persisted Area in EPSG:31370 and normalized to
+EPSG:4326 before canonical `vector_features` persistence.
+
+Stable source fields include `agpakey`, parcel number, declared source area,
+reference id, spring crop, main crop, official main-crop group, production
+method and source municipality. All annual source properties remain available,
+but only `maincropgroup_title` is normalized to a controlled comparison key.
+Detailed crop codes and titles can change meaning or wording between editions;
+their complete annual code list is therefore retained with the source archive.
+
+Historical analysis compares intersected hectares for the complete declaration
+and official main-crop groups. It must not compare individual parcel lineage:
+identifiers, boundaries and declarations are not stable between campaign years.
+The declaration can include water, hedges, buildings and infrastructure, so its
+total is not a cadastral ownership area or cultivated-crop area.
+
 ### Vector
 
 Ondersteund:

docs/STORAGE_ARCHITECTURE.md

@@ -116,6 +116,23 @@ checksums, boundary checksum, page/feature counts, clipping diagnostics and
 official class totals. The final GeoJSON is the retained source artifact while
 `datasets` and `vector_features` remain the canonical queryable PostGIS state.
 
+Definitive agricultural-use parcel evidence lives under:
+
+```text
+storage/operator-evidence/agricultural-use-parcels/{scope}/{year}/
+  agpa_{year}_*_public.zip
+  agricultural_use_parcels_{year}_{scope}.geojson
+  agricultural_use_parcels_{year}_crop_codes.json
+  agricultural_use_parcels_{year}_{scope}.manifest.json
+```
+
+The official ZIP is immutable source evidence and is never removed by normal
+cache cleanup. The extracted GeoPackage is temporary to avoid retaining a
+second full source copy. The normalized GeoJSON is the canonical upload
+artifact; Dataset and vector_feature rows remain the queryable PostGIS state.
+The manifest binds source, crop-code list and upload artifact checksums. A
+checksum conflict with an existing annual Dataset fails closed.
+
 Offline demo export artifacts can be inspected and cleaned with:
 
 ```bash

frontend/README.md

@@ -476,6 +476,14 @@ separate official value classes, PHAB-derived Natura 2000 and regional-biotope
 hectares, uncertain habitat knowledge gaps and source provenance. This single
 edition is not exposed as a fabricated annual series in Evolution.
 
+When definitive ALZ datasets are loaded, the Map explorer adds `Landbouw` as a
+separate theme rather than mixing it with cadastral parcels. Current mode shows
+declared-use hectares and official main-crop group hectares. Evolution mode
+compares any two persisted annual editions and charts the complete series.
+Object additions/removals stay hidden because annual parcel identity is not
+stable. The Sources inventory only labels the series available after real
+Datasets exist.
+
 ## Useful repository scripts
 
 - `bash scripts/frontend_install.sh`

frontend/src/components/datasets/SourceCatalogPanel.tsx

@@ -9,6 +9,7 @@ const THEME_LABELS: Record<string, string> = {
   population: 'Bevolking',
   forest: 'Bos',
   nature_value: 'Natuurwaarde',
+  agriculture: 'Landbouw',
   water: 'Water',
   roads: 'Wegen en transport',
   parcels: 'Percelen',
@@ -105,10 +106,14 @@ export function SourceCatalogPanel({ datasets }: SourceCatalogPanelProps): JSX.E
       String(dataset.source_metadata?.['product_key'] ?? 'most_recent') !== 'most_recent',
   )
   const bwkDatasets = ready.filter((dataset) => dataset.source_name === 'inbo_bwk_natura2000')
+  const agricultureDatasets = ready.filter((dataset) => dataset.source_name === 'agentschap_landbouw_zeevisserij_agricultural_parcels')
+  const agricultureYears = agricultureDatasets
+    .flatMap((dataset) => dataset.observed_at ? [new Date(dataset.observed_at).getUTCFullYear()] : [])
   const pendingSources = AVAILABLE_SOURCES.filter((source) => {
     if (source.key === 'waterinfo') return waterinfoDatasets.length === 0
     if (source.key === 'historical_orthophoto') return historicalOrthophotos.length === 0
     if (source.key === 'bwk') return bwkDatasets.length === 0
+    if (source.key === 'agriculture') return agricultureDatasets.length === 0
     return true
   })
   const themes = Object.keys(THEME_LABELS).map((theme) => {
@@ -168,7 +173,7 @@ export function SourceCatalogPanel({ datasets }: SourceCatalogPanelProps): JSX.E
         ))}
       </div>
 
-      {waterinfoDatasets.length > 0 || historicalOrthophotos.length > 0 || bwkDatasets.length > 0 ? (
+      {waterinfoDatasets.length > 0 || historicalOrthophotos.length > 0 || bwkDatasets.length > 0 || agricultureDatasets.length > 0 ? (
         <div className="source-catalog-loaded" aria-label="Aanvullende ingeladen bronnen">
           {waterinfoDatasets.length > 0 ? (
             <article>
@@ -191,6 +196,13 @@ export function SourceCatalogPanel({ datasets }: SourceCatalogPanelProps): JSX.E
               <p>Biologische waardering en habitataandelen blijven afzonderlijke, brongetrouwe metingen met herkomstinformatie.</p>
             </article>
           ) : null}
+          {agricultureDatasets.length > 0 ? (
+            <article>
+              <strong>Landbouwgebruikspercelen</strong>
+              <span>{agricultureDatasets.length} definitieve jaargangen{agricultureYears.length ? ` · ${Math.min(...agricultureYears)}-${Math.max(...agricultureYears)}` : ''}</span>
+              <p>Oppervlakte en officiele hoofdteeltgroepen zijn historisch vergelijkbaar; perceelidentiteiten blijven bewust niet gekoppeld tussen jaren.</p>
+            </article>
+          ) : null}
         </div>
       ) : null}
 

frontend/src/components/map/MapWorkspace.tsx

@@ -13,7 +13,7 @@ const EMPTY_TEMPORAL_SERIES: DatasetCreateResponse[] = []
 const MOL_PROJECT_NAME = 'Mol Municipality Workbench'
 const KEMPEN_PROJECT_NAME = 'Kempen Regional Workbench'
 
-type DataThemeId = 'buildings' | 'population' | 'forest' | 'nature_value' | 'water' | 'roads' | 'parcels'
+type DataThemeId = 'buildings' | 'population' | 'forest' | 'nature_value' | 'agriculture' | 'water' | 'roads' | 'parcels'
 
 interface DataTheme {
   id: DataThemeId
@@ -58,6 +58,13 @@ const DATA_THEMES: DataTheme[] = [
     description: 'Biologische waardering, Natura 2000-habitat en regionaal belangrijke biotopen uit de BWK.',
     tokens: ['nature_value', 'nature value', 'natuurwaarde', 'bwk', 'natura2000', 'natura 2000', 'biodiversity'],
   },
+  {
+    id: 'agriculture',
+    label: 'Landbouw',
+    shortLabel: 'Landbouwgebruik',
+    description: 'Jaarlijkse officiele landbouwgebruikspercelen en hoofdteeltgroepen.',
+    tokens: ['agriculture', 'agricultural', 'landbouw', 'landbouwgebruik', 'agpa'],
+  },
   {
     id: 'water',
     label: 'Water',
@@ -86,6 +93,7 @@ const DATA_THEME_MAP_STYLES: Record<DataThemeId, { fill: string; line: string }>
   population: { fill: '#7559a6', line: '#5b3f88' },
   forest: { fill: '#347950', line: '#225f3b' },
   nature_value: { fill: '#9a4f64', line: '#74364a' },
+  agriculture: { fill: '#7b8f32', line: '#53671d' },
   water: { fill: '#2676a8', line: '#155b85' },
   roads: { fill: '#6b7280', line: '#4b5563' },
   parcels: { fill: '#a7792f', line: '#7d571f' },
@@ -120,6 +128,7 @@ function pickThemeDataset(datasets: DatasetCreateResponse[], theme: DataTheme):
       (dataset.source_name === 'grb' ? 100_000 : 0) +
       (dataset.source_name === 'department_omgeving_land_use' ? 90_000 : 0) +
       (dataset.source_name === 'inbo_bwk_natura2000' ? 95_000 : 0) +
+      (dataset.source_name === 'agentschap_landbouw_zeevisserij_agricultural_parcels' ? 98_000 : 0) +
       (dataset.dataset_role === 'reference' ? 10_000 : 0) +
       (dataset.observed_at ? new Date(dataset.observed_at).getTime() / 100_000_000 : 0) +
       (dataset.feature_count ?? dataset.vector_summary?.feature_count ?? 0)

frontend/src/styles/app.css

@@ -5754,6 +5754,7 @@ section {
 .geo-theme-symbol-population { background: #7559a6; }
 .geo-theme-symbol-forest { background: #347950; }
 .geo-theme-symbol-nature_value { background: #9a4f64; }
+.geo-theme-symbol-agriculture { background: #7b8f32; }
 .geo-theme-symbol-water { background: #2676a8; }
 .geo-theme-symbol-roads { background: #6b7280; }
 .geo-theme-symbol-parcels { background: #a7792f; }

scripts/README.md

@@ -1457,6 +1457,30 @@ pagination, retains raw JSON/checksums, clips polygon geometry in EPSG:31370
 and writes only through the Dataset upload flow. `--force` refetches source
 evidence but never silently replaces a conflicting persisted 2025 artifact.
 
+## Definitive agricultural-use parcel history
+
+Prepare a bounded subset of years for source review:
+
+```bash
+docker exec geointel python /app/scripts/provision_agricultural_parcel_history.py \
+  --years 2008,2019,2025 --fetch-only
+```
+
+Import all definitive annual editions for the regional workspace:
+
+```bash
+docker exec geointel python /app/scripts/provision_agricultural_parcel_history.py
+```
+
+The operator requires the optional GIS runtime already included in the Unraid
+image. It paginates GeoIntel API collections within the canonical 200-item
+limit, downloads only the fixed official archive allowlist, enforces a 250 MiB
+per-archive ceiling, validates one EPSG:31370 polygon GeoPackage and clips
+exactly to the persisted scope. Repeat runs reuse matching manifests and
+Datasets. `--force` refreshes retained evidence but cannot silently replace a
+conflicting persisted annual checksum. Use `--scope mol` for an independent
+municipal series.
+
 ## Tower deployment
 
 Push the local branch to Gitea, then rebuild the Unraid/Tower Docker runtime:

scripts/provision_agricultural_parcel_history.py

@@ -0,0 +1,750 @@
+"""Provision definitive Flemish agricultural-use parcels for an explicit scope.
+
+The operator downloads only the official annual ALZ archives, retains those
+archives as checksummed evidence, validates the GeoPackage schema and CRS,
+clips parcel geometry to the persisted GeoIntel Area in EPSG:31370 and uploads
+one canonical vector Dataset per year. It never writes to vector_features
+directly and never uses the provisional current-campaign snapshot.
+"""
+
+from __future__ import annotations
+
+import argparse
+import hashlib
+import json
+import math
+import os
+import shutil
+import sys
+import tempfile
+import zipfile
+from datetime import date, datetime, timezone
+from pathlib import Path
+from typing import Any, Iterable
+
+import requests
+from pyproj import Transformer
+from requests.adapters import HTTPAdapter
+from shapely.geometry import GeometryCollection, MultiPolygon, Polygon, mapping, shape
+from shapely.ops import transform as transform_geometry
+from shapely.ops import unary_union
+from shapely.validation import make_valid
+from urllib3.util.retry import Retry
+
+
+SCRIPT_DIR = Path(__file__).resolve().parent
+if str(SCRIPT_DIR) not in sys.path:
+    sys.path.insert(0, str(SCRIPT_DIR))
+
+from geographic_scopes import GEOGRAPHIC_SCOPES, GeographicScope  # noqa: E402
+
+
+DEFAULT_API_URL = "http://127.0.0.1:8000"
+DEFAULT_SCOPE_KEY = "kempen-transport-region"
+DEFAULT_OUTPUT_ROOT = Path("/app/storage/operator-evidence/agricultural-use-parcels")
+CATALOG_URL = "https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen"
+DATA_CATALOG_URL = "https://www.vlaanderen.be/datavindplaats/catalogus/open-geodata-landbouwgebruikspercelen"
+ATTRIBUTION = "Agentschap Landbouw en Zeevisserij - Landbouwcijfers"
+SOURCE_NAME = "agentschap_landbouw_zeevisserij_agricultural_parcels"
+REFERENCE_LAYER_NAME = "agriculture"
+SOURCE_CRS = "EPSG:31370"
+OUTPUT_CRS = "EPSG:4326"
+SCHEMA_VERSION = 1
+MAX_ARCHIVE_BYTES = 250 * 1024 * 1024
+
+ARCHIVE_URLS = {
+    2008: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2008_2022-03-23_public.zip",
+    2009: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2009_2022-03-23_public.zip",
+    2010: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2010_2022-03-23_public.zip",
+    2011: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2011_2022-03-23_public.zip",
+    2012: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2012_2022-03-23_public.zip",
+    2013: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2013_2022-03-23_public.zip",
+    2014: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2014_2022-03-23_public.zip",
+    2015: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2015_2022-03-23_public.zip",
+    2016: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2016_2022-03-23_public.zip",
+    2017: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2017_2022-03-23_public.zip",
+    2018: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2018_2022-03-23_public.zip",
+    2019: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2019_2020-03-20_public.zip",
+    2020: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2020_2021-03-19_public.zip",
+    2021: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2021_2022-03-15_public.zip",
+    2022: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2022_2023-06-26_public.zip",
+    2023: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2023_2024-03-28_public.zip",
+    2024: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2024_2025-03-27_public.zip",
+    2025: "https://www.landbouwvlaanderen.be/bestanden/gis/agpa_2025_2026-05-13_public.zip",
+}
+SUPPORTED_YEARS = tuple(ARCHIVE_URLS)
+
+STABLE_REQUIRED_FIELDS = {
+    "agpakey",
+    "parcelnumber",
+    "area_ha",
+    "reference_id",
+    "springcrop_code",
+    "springcrop_title",
+    "maincrop_code",
+    "maincrop_title",
+    "maincropgroup_title",
+    "productionmethod_code",
+    "productionmethod_title",
+    "municipality_code",
+    "municipality_title",
+}
+
+GROUP_ALIASES = {
+    "grasland": "grassland",
+    "mais": "maize",
+    "granen/zaden/peulvruchten": "grains_seeds_legumes",
+    "aardappelen": "potatoes",
+    "groenten/kruiden/sierplanten": "horticulture",
+    "suikerbieten": "sugar_beets",
+    "voedergewassen": "fodder_crops",
+    "fruit en noten": "fruit_nuts",
+    "overige gewassen": "other_crops",
+    "vlas en hennep": "flax_hemp",
+    "houtachtige gewassen": "woody_crops",
+    "landbouwinfrastructuur": "agricultural_infrastructure",
+    "water": "water",
+}
+
+METRIC_GROUPS = (
+    ("grassland_area", "Grasland", ("grassland",)),
+    ("maize_area", "Mais", ("maize",)),
+    ("grains_seeds_legumes_area", "Granen, zaden en peulvruchten", ("grains_seeds_legumes",)),
+    ("potatoes_area", "Aardappelen", ("potatoes",)),
+    ("horticulture_area", "Groenten, kruiden en sierplanten", ("horticulture",)),
+    ("sugar_beets_area", "Suikerbieten", ("sugar_beets",)),
+    ("fodder_crops_area", "Voedergewassen", ("fodder_crops",)),
+    ("fruit_nuts_area", "Fruit en noten", ("fruit_nuts",)),
+    ("other_crops_area", "Overige gewassen", ("other_crops", "flax_hemp", "woody_crops")),
+    ("agricultural_infrastructure_area", "Landbouwinfrastructuur", ("agricultural_infrastructure",)),
+    ("agricultural_water_area", "Water binnen de aangifte", ("water",)),
+)
+
+TO_LAMBERT72 = Transformer.from_crs(OUTPUT_CRS, SOURCE_CRS, always_xy=True)
+TO_WGS84 = Transformer.from_crs(SOURCE_CRS, OUTPUT_CRS, always_xy=True)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Provision definitive ALZ agricultural-use parcel history.")
+    parser.add_argument("--base-url", default=os.environ.get("GEOINTEL_INTERNAL_API_URL", DEFAULT_API_URL))
+    parser.add_argument("--scope", choices=sorted(GEOGRAPHIC_SCOPES), default=DEFAULT_SCOPE_KEY)
+    parser.add_argument("--years", default=",".join(str(year) for year in SUPPORTED_YEARS))
+    parser.add_argument("--output-root", type=Path, default=Path(os.environ.get("GEOINTEL_AGRICULTURE_OUTPUT_ROOT", DEFAULT_OUTPUT_ROOT)))
+    parser.add_argument("--request-timeout", type=int, default=900)
+    parser.add_argument("--import-timeout", type=int, default=3600)
+    parser.add_argument("--max-features", type=int, default=250_000)
+    parser.add_argument("--max-archive-mb", type=int, default=250)
+    parser.add_argument("--force", action="store_true", help="Redownload and rebuild retained evidence artifacts.")
+    parser.add_argument("--fetch-only", action="store_true", help="Prepare evidence without importing Datasets.")
+    return parser.parse_args()
+
+
+def utc_now() -> str:
+    return datetime.now(timezone.utc).isoformat()
+
+
+def sha256_file(path: Path) -> str:
+    digest = hashlib.sha256()
+    with path.open("rb") as handle:
+        for chunk in iter(lambda: handle.read(1024 * 1024), b""):
+            digest.update(chunk)
+    return digest.hexdigest()
+
+
+def write_json_atomic(path: Path, payload: Any, *, pretty: bool = False) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    temporary = path.with_suffix(path.suffix + ".partial")
+    temporary.write_text(
+        json.dumps(payload, ensure_ascii=False, indent=2 if pretty else None, sort_keys=pretty, separators=None if pretty else (",", ":")),
+        encoding="utf-8",
+    )
+    temporary.replace(path)
+
+
+def build_session() -> requests.Session:
+    retry = Retry(
+        total=5,
+        connect=5,
+        read=5,
+        status=5,
+        backoff_factor=1.0,
+        status_forcelist=(429, 500, 502, 503, 504),
+        allowed_methods=frozenset({"GET"}),
+        raise_on_status=True,
+    )
+    session = requests.Session()
+    session.headers.update({"User-Agent": "GeoIntel-ALZ-Agricultural-Parcels-Operator/1.0"})
+    adapter = HTTPAdapter(max_retries=retry)
+    session.mount("https://", adapter)
+    session.mount("http://", adapter)
+    return session
+
+
+def response_data(response: requests.Response) -> Any:
+    try:
+        payload = response.json()
+    except requests.JSONDecodeError as exc:
+        raise RuntimeError(f"GeoIntel API returned non-JSON content ({response.status_code})") from exc
+    if not response.ok:
+        if isinstance(payload, dict):
+            message = payload.get("message") or payload.get("error") or response.text
+        else:
+            message = response.text
+        raise RuntimeError(f"GeoIntel API request failed ({response.status_code}): {message}")
+    if isinstance(payload, dict) and "data" in payload:
+        return payload["data"]
+    return payload
+
+
+def api_items(session: requests.Session, url: str, timeout: int) -> list[dict[str, Any]]:
+    items: list[dict[str, Any]] = []
+    offset = 0
+    while True:
+        response = session.get(url, params={"limit": 200, "offset": offset}, timeout=timeout)
+        data = response_data(response)
+        if isinstance(data, dict):
+            page = data.get("items") or data.get("results") or []
+            total = int(data.get("total") or len(page))
+        else:
+            page = data
+            total = len(page) if isinstance(page, list) else 0
+        if not isinstance(page, list):
+            raise RuntimeError(f"Expected a list response from {url}")
+        items.extend(item for item in page if isinstance(item, dict))
+        if not page or len(items) >= total:
+            return items
+        offset += len(page)
+
+
+def locate_workspace(
+    session: requests.Session,
+    base_url: str,
+    scope: GeographicScope,
+    timeout: int,
+) -> tuple[str, str, Any, list[dict[str, Any]]]:
+    projects = api_items(session, f"{base_url}/api/v1/projects", timeout)
+    project = next((item for item in projects if item.get("name") == scope.project_name), None)
+    if project is None:
+        raise RuntimeError(f"Project {scope.project_name!r} is missing; provision the geographic scope first")
+    project_id = str(project["id"])
+    areas = api_items(session, f"{base_url}/api/v1/projects/{project_id}/areas", timeout)
+    area = next((item for item in areas if item.get("name") == scope.area_name), None)
+    if area is None:
+        raise RuntimeError(f"Area {scope.area_name!r} is missing from project {scope.project_name!r}")
+    boundary = polygonal_geometry(shape(area.get("geometry")))
+    if boundary is None:
+        raise RuntimeError(f"Area {scope.area_name!r} has no valid polygon geometry")
+    datasets = api_items(session, f"{base_url}/api/v1/projects/{project_id}/datasets", timeout)
+    return project_id, str(area["id"]), boundary, datasets
+
+
+def parse_years(raw: str) -> list[int]:
+    try:
+        years = sorted({int(value.strip()) for value in raw.split(",") if value.strip()})
+    except ValueError as exc:
+        raise ValueError("Years must be a comma-separated list of integers") from exc
+    unsupported = [year for year in years if year not in ARCHIVE_URLS]
+    if not years or unsupported:
+        raise ValueError(f"Supported definitive years are {SUPPORTED_YEARS}; unsupported: {unsupported}")
+    return years
+
+
+def polygonal_geometry(geometry):
+    if geometry is None or geometry.is_empty:
+        return None
+    if not geometry.is_valid:
+        geometry = make_valid(geometry)
+    if isinstance(geometry, (Polygon, MultiPolygon)):
+        return geometry
+    if isinstance(geometry, GeometryCollection):
+        polygons = [part for part in geometry.geoms if isinstance(part, (Polygon, MultiPolygon)) and not part.is_empty]
+        if not polygons:
+            return None
+        merged = unary_union(polygons)
+        return merged if isinstance(merged, (Polygon, MultiPolygon)) and not merged.is_empty else None
+    return None
+
+
+def normalized_group_title(value: Any) -> str | None:
+    if value is None:
+        return None
+    normalized = str(value).strip().casefold().replace("ï", "i")
+    return GROUP_ALIASES.get(normalized, normalized.replace(" ", "_")) if normalized else None
+
+
+def json_value(value: Any) -> Any:
+    if value is None:
+        return None
+    if isinstance(value, (datetime, date)):
+        return value.isoformat()
+    if isinstance(value, float) and not math.isfinite(value):
+        return None
+    if isinstance(value, (str, int, float, bool)):
+        return value
+    if hasattr(value, "item"):
+        try:
+            return json_value(value.item())
+        except (TypeError, ValueError):
+            pass
+    return str(value)
+
+
+def download_archive(
+    session: requests.Session,
+    url: str,
+    destination: Path,
+    *,
+    timeout: int,
+    max_bytes: int,
+    force: bool,
+) -> dict[str, Any]:
+    destination.parent.mkdir(parents=True, exist_ok=True)
+    if destination.is_file() and not force:
+        validate_archive(destination)
+        return {"status": "reused", "sha256": sha256_file(destination), "size_bytes": destination.stat().st_size}
+    temporary = destination.with_suffix(destination.suffix + ".partial")
+    temporary.unlink(missing_ok=True)
+    response = session.get(url, timeout=timeout, stream=True)
+    response.raise_for_status()
+    content_length = int(response.headers.get("content-length") or 0)
+    if content_length > max_bytes:
+        raise RuntimeError(f"Official archive exceeds the configured {max_bytes // (1024 * 1024)} MiB limit")
+    size = 0
+    try:
+        with temporary.open("wb") as handle:
+            for chunk in response.iter_content(chunk_size=1024 * 1024):
+                if not chunk:
+                    continue
+                size += len(chunk)
+                if size > max_bytes:
+                    raise RuntimeError(f"Official archive exceeded the configured {max_bytes // (1024 * 1024)} MiB limit while streaming")
+                handle.write(chunk)
+        validate_archive(temporary)
+        temporary.replace(destination)
+    except Exception:
+        temporary.unlink(missing_ok=True)
+        raise
+    return {"status": "downloaded", "sha256": sha256_file(destination), "size_bytes": destination.stat().st_size}
+
+
+def archive_geopackage_member(path: Path) -> str:
+    with zipfile.ZipFile(path) as archive:
+        members = [item.filename for item in archive.infolist() if not item.is_dir() and item.filename.lower().endswith(".gpkg")]
+    if len(members) != 1:
+        raise RuntimeError(f"Official archive must contain exactly one GeoPackage; found {len(members)}")
+    member = members[0]
+    if Path(member).name != member or ".." in Path(member).parts:
+        raise RuntimeError("Official archive contains an unsafe GeoPackage path")
+    return member
+
+
+def validate_archive(path: Path) -> str:
+    try:
+        return archive_geopackage_member(path)
+    except zipfile.BadZipFile as exc:
+        raise RuntimeError(f"Official source archive {path.name} is not a valid ZIP") from exc
+
+
+def extract_geopackage(archive_path: Path, destination_dir: Path) -> Path:
+    member = validate_archive(archive_path)
+    destination = destination_dir / Path(member).name
+    with zipfile.ZipFile(archive_path) as archive, archive.open(member) as source, destination.open("wb") as target:
+        shutil.copyfileobj(source, target, length=1024 * 1024)
+    if destination.stat().st_size == 0:
+        raise RuntimeError("Extracted official GeoPackage is empty")
+    return destination
+
+
+def load_pyogrio():
+    try:
+        import pyogrio  # type: ignore[import-not-found]
+    except ImportError as exc:
+        raise RuntimeError("Agricultural parcel preparation requires the optional GeoIntel GIS dependencies (pyogrio/geopandas)") from exc
+    return pyogrio
+
+
+def inspect_geopackage(pyogrio, path: Path) -> tuple[str, set[str], int]:
+    layers = pyogrio.list_layers(path)
+    polygon_layers = [str(row[0]) for row in layers if "Polygon" in str(row[1])]
+    if len(polygon_layers) != 1:
+        raise RuntimeError(f"Expected one polygon layer in {path.name}; found {polygon_layers}")
+    layer = polygon_layers[0]
+    info = pyogrio.read_info(path, layer=layer)
+    crs = str(info.get("crs") or "")
+    if "31370" not in crs:
+        raise RuntimeError(f"Expected EPSG:31370 agricultural parcels; received {crs or 'no CRS'}")
+    info_fields = info.get("fields")
+    fields = {str(value).lower() for value in info_fields} if info_fields is not None else set()
+    missing = sorted(STABLE_REQUIRED_FIELDS - fields)
+    if missing:
+        raise RuntimeError(f"Agricultural parcel schema is missing stable fields: {', '.join(missing)}")
+    return layer, fields, int(info.get("features") or 0)
+
+
+def build_crop_code_list(records: Iterable[dict[str, Any]], *, year: int) -> dict[str, Any]:
+    crops: dict[tuple[str, str, str], dict[str, Any]] = {}
+    groups: set[str] = set()
+    title_by_code: dict[str, set[str]] = {}
+    for record in records:
+        code = str(json_value(record.get("maincrop_code")) or "").strip()
+        title = str(json_value(record.get("maincrop_title")) or "").strip()
+        group_title = str(json_value(record.get("maincropgroup_title")) or "").strip()
+        if not code and not title and not group_title:
+            continue
+        crops[(code, title, group_title)] = {"code": code or None, "title": title or None, "group_title": group_title or None}
+        if group_title:
+            groups.add(group_title)
+        if code and title:
+            title_by_code.setdefault(code, set()).add(title)
+    conflicts = {code: sorted(titles) for code, titles in title_by_code.items() if len(titles) > 1}
+    return {
+        "year": year,
+        "source": ATTRIBUTION,
+        "generated_from": "full official annual GeoPackage attributes",
+        "crop_entries": sorted(crops.values(), key=lambda item: (str(item["code"]), str(item["title"]), str(item["group_title"]))),
+        "main_crop_groups": sorted(groups),
+        "code_title_conflicts": conflicts,
+        "historical_comparison_rule": "Use maincropgroup_title for comparable grouped hectares; detailed crop code/title remains source-faithful per year.",
+    }
+
+
+def dataframe_records(frame, columns: Iterable[str]) -> Iterable[dict[str, Any]]:
+    for row in frame.itertuples(index=False, name=None):
+        yield {column: json_value(value) for column, value in zip(columns, row)}
+
+
+def normalize_frame(frame, *, year: int, boundary_lambert72, max_features: int) -> tuple[list[dict[str, Any]], dict[str, Any]]:
+    features: list[dict[str, Any]] = []
+    seen_ids: set[str] = set()
+    clipped_count = 0
+    source_area_ha = 0.0
+    exact_area_ha = 0.0
+    for _, row in frame.iterrows():
+        source_geometry = polygonal_geometry(row.geometry)
+        if source_geometry is None or not source_geometry.intersects(boundary_lambert72):
+            continue
+        clipped = polygonal_geometry(source_geometry.intersection(boundary_lambert72))
+        if clipped is None or clipped.area <= 0:
+            continue
+        properties = {str(column).lower(): json_value(row[column]) for column in frame.columns if str(column).lower() != "geometry"}
+        agpa_key = str(properties.get("agpakey") or "").strip()
+        if not agpa_key:
+            raise RuntimeError(f"Agricultural parcel {year} contains a feature without agpakey")
+        source_feature_id = f"alz:{year}:{agpa_key}"
+        if source_feature_id in seen_ids:
+            raise RuntimeError(f"Agricultural parcel {year} contains duplicate agpakey {agpa_key}")
+        seen_ids.add(source_feature_id)
+        clipped_wgs84 = polygonal_geometry(transform_geometry(TO_WGS84.transform, clipped))
+        if clipped_wgs84 is None:
+            raise RuntimeError(f"Agricultural parcel {agpa_key} could not be transformed to EPSG:4326")
+        source_area = float(source_geometry.area) / 10_000.0
+        exact_area = float(clipped.area) / 10_000.0
+        source_area_ha += source_area
+        exact_area_ha += exact_area
+        was_clipped = not source_geometry.within(boundary_lambert72)
+        clipped_count += int(was_clipped)
+        properties.update(
+            {
+                "source_feature_id": source_feature_id,
+                "source_year": year,
+                "source_agpa_key": agpa_key,
+                "main_crop_group_key": normalized_group_title(properties.get("maincropgroup_title")),
+                "source_geometry_area_ha": round(source_area, 8),
+                "clipped_area_ha": round(exact_area, 8),
+                "geometry_was_clipped": was_clipped,
+                "historical_parcel_identity_stable": False,
+            }
+        )
+        features.append({"type": "Feature", "id": source_feature_id, "geometry": mapping(clipped_wgs84), "properties": properties})
+        if len(features) > max_features:
+            raise RuntimeError(f"Clipped agricultural parcel count exceeds the configured limit of {max_features}")
+    if not features:
+        raise RuntimeError(f"Official agricultural parcel archive {year} has no features inside the persisted scope")
+    return features, {
+        "feature_count": len(features),
+        "clipped_feature_count": clipped_count,
+        "source_geometry_area_ha": round(source_area_ha, 4),
+        "clipped_area_ha": round(exact_area_ha, 4),
+    }
+
+
+def artifact_paths(output_root: Path, scope_key: str, year: int) -> dict[str, Path]:
+    directory = output_root / scope_key / str(year)
+    return {
+        "directory": directory,
+        "archive": directory / Path(ARCHIVE_URLS[year]).name,
+        "artifact": directory / f"agricultural_use_parcels_{year}_{scope_key}.geojson",
+        "codelist": directory / f"agricultural_use_parcels_{year}_crop_codes.json",
+        "manifest": directory / f"agricultural_use_parcels_{year}_{scope_key}.manifest.json",
+    }
+
+
+def reusable_artifact(paths: dict[str, Path], *, year: int, scope_key: str) -> dict[str, Any] | None:
+    if not all(paths[key].is_file() for key in ("archive", "artifact", "codelist", "manifest")):
+        return None
+    manifest = json.loads(paths["manifest"].read_text(encoding="utf-8"))
+    if manifest.get("schema_version") != SCHEMA_VERSION or manifest.get("year") != year or manifest.get("scope_key") != scope_key:
+        return None
+    if manifest.get("source_archive_sha256") != sha256_file(paths["archive"]):
+        return None
+    if manifest.get("artifact_sha256") != sha256_file(paths["artifact"]):
+        return None
+    if manifest.get("crop_code_list_sha256") != sha256_file(paths["codelist"]):
+        return None
+    validate_archive(paths["archive"])
+    return manifest
+
+
+def prepare_year(
+    session: requests.Session,
+    *,
+    year: int,
+    scope: GeographicScope,
+    boundary_wgs84,
+    output_root: Path,
+    request_timeout: int,
+    max_archive_bytes: int,
+    max_features: int,
+    force: bool,
+) -> tuple[dict[str, Path], dict[str, Any]]:
+    paths = artifact_paths(output_root, scope.key, year)
+    paths["directory"].mkdir(parents=True, exist_ok=True)
+    if not force:
+        reused = reusable_artifact(paths, year=year, scope_key=scope.key)
+        if reused is not None:
+            return paths, reused
+    download = download_archive(
+        session,
+        ARCHIVE_URLS[year],
+        paths["archive"],
+        timeout=request_timeout,
+        max_bytes=max_archive_bytes,
+        force=force,
+    )
+    boundary_lambert72 = polygonal_geometry(transform_geometry(TO_LAMBERT72.transform, boundary_wgs84))
+    if boundary_lambert72 is None:
+        raise RuntimeError(f"Scope {scope.key} could not be transformed to EPSG:31370")
+    pyogrio = load_pyogrio()
+    with tempfile.TemporaryDirectory(prefix=f"agpa-{year}-", dir=paths["directory"]) as temporary:
+        gpkg_path = extract_geopackage(paths["archive"], Path(temporary))
+        layer, fields, source_feature_count = inspect_geopackage(pyogrio, gpkg_path)
+        code_columns = [field for field in ("maincrop_code", "maincrop_title", "maincropgroup_title") if field in fields]
+        code_frame = pyogrio.read_dataframe(gpkg_path, layer=layer, columns=code_columns, read_geometry=False)
+        code_list = build_crop_code_list(dataframe_records(code_frame, code_columns), year=year)
+        write_json_atomic(paths["codelist"], code_list, pretty=True)
+        frame = pyogrio.read_dataframe(gpkg_path, layer=layer, bbox=boundary_lambert72.bounds)
+        features, summary = normalize_frame(frame, year=year, boundary_lambert72=boundary_lambert72, max_features=max_features)
+    artifact = {"type": "FeatureCollection", "name": paths["artifact"].stem, "features": features}
+    write_json_atomic(paths["artifact"], artifact)
+    manifest = {
+        "schema_version": SCHEMA_VERSION,
+        "generated_at": utc_now(),
+        "year": year,
+        "scope_key": scope.key,
+        "scope_name": scope.display_name,
+        "scope_type": scope.scope_type,
+        "member_nis_codes": list(scope.nis_codes),
+        "source_url": ARCHIVE_URLS[year],
+        "catalog_url": CATALOG_URL,
+        "data_catalog_url": DATA_CATALOG_URL,
+        "attribution": ATTRIBUTION,
+        "source_crs": SOURCE_CRS,
+        "output_crs": OUTPUT_CRS,
+        "source_archive_path": str(paths["archive"]),
+        "source_archive_sha256": download["sha256"],
+        "source_archive_size_bytes": download["size_bytes"],
+        "source_feature_count": source_feature_count,
+        "source_fields": sorted(fields),
+        "crop_code_list_path": str(paths["codelist"]),
+        "crop_code_list_sha256": sha256_file(paths["codelist"]),
+        "artifact_path": str(paths["artifact"]),
+        "artifact_sha256": sha256_file(paths["artifact"]),
+        **summary,
+        "limitations": [
+            "Parcel identities are not stable across campaign years; evolution compares grouped area totals, not parcel lineage.",
+            "The layer describes declared agricultural use at the annual reference deadline and can include water, hedges, buildings and infrastructure.",
+            "Detailed crop codes and titles remain source-faithful per year; only official main-crop groups are used for comparable historical metrics.",
+            "The current-campaign provisional snapshot is intentionally excluded.",
+        ],
+    }
+    write_json_atomic(paths["manifest"], manifest, pretty=True)
+    return paths, manifest
+
+
+def selection_metrics() -> list[dict[str, Any]]:
+    warning = "Historische evolutie vergelijkt officiele hoofdteeltgroepen; individuele perceelidentiteiten zijn niet stabiel tussen campagnejaren."
+    return [
+        {
+            "metric_key": metric_key,
+            "method": "intersection_area",
+            "label": label,
+            "unit": "ha",
+            "geometry_dimension": 2,
+            "filter_property": "main_crop_group_key",
+            "filter_values": list(group_keys),
+            "warning": warning,
+        }
+        for metric_key, label, group_keys in METRIC_GROUPS
+    ]
+
+
+def series_key(scope: GeographicScope) -> str:
+    return f"alz:agricultural-use-parcels:{scope.key}"
+
+
+def upload_artifact(
+    session: requests.Session,
+    *,
+    base_url: str,
+    project_id: str,
+    area_id: str,
+    scope: GeographicScope,
+    year: int,
+    paths: dict[str, Path],
+    manifest: dict[str, Any],
+    timeout: int,
+) -> dict[str, Any]:
+    source_metadata = {
+        "provider": "Agentschap Landbouw en Zeevisserij",
+        "theme": "agriculture",
+        "layer_name": f"Landbouwgebruikspercelen {year}",
+        "authority_level": "authoritative",
+        "coverage_scope": scope.scope_type,
+        "scope_key": scope.key,
+        "scope_name": scope.display_name,
+        "member_nis_codes": list(scope.nis_codes),
+        "feature_count": manifest["feature_count"],
+        "geometry_clipped_to_area": True,
+        "identity_stable": False,
+        "semantic_metrics": False,
+        "attribution": ATTRIBUTION,
+        "catalog_url": CATALOG_URL,
+        "selection_aggregation": {
+            "metric_key": "declared_agricultural_use_area",
+            "method": "intersection_area",
+            "label": "Aangegeven gebruiksoppervlakte",
+            "unit": "ha",
+            "geometry_dimension": 2,
+            "warning": "De aangifte bevat naast teelten ook onder meer water, hagen, gebouwen en landbouwinfrastructuur; dit is geen eigendoms- of kadastrale oppervlakte.",
+        },
+        "selection_metrics": selection_metrics(),
+    }
+    provenance_metadata = {
+        "operator_tool": "provision_agricultural_parcel_history.py",
+        "operator_explicit_fetch": True,
+        "geometry_clipped_to_area": True,
+        "source_archive_url": ARCHIVE_URLS[year],
+        "source_archive_path": str(paths["archive"]),
+        "source_archive_sha256": manifest["source_archive_sha256"],
+        "crop_code_list_path": str(paths["codelist"]),
+        "crop_code_list_sha256": manifest["crop_code_list_sha256"],
+        "manifest_path": str(paths["manifest"]),
+        "artifact_sha256": manifest["artifact_sha256"],
+        "catalog_url": CATALOG_URL,
+        "limitations": manifest["limitations"],
+    }
+    with paths["artifact"].open("rb") as handle:
+        response = session.post(
+            f"{base_url}/api/v1/projects/{project_id}/datasets/upload",
+            data={
+                "dataset_type": "vector",
+                "source": "operator_official_import",
+                "dataset_role": "reference",
+                "source_name": SOURCE_NAME,
+                "reference_layer_name": REFERENCE_LAYER_NAME,
+                "source_metadata_json": json.dumps(source_metadata, ensure_ascii=False),
+                "provenance_metadata_json": json.dumps(provenance_metadata, ensure_ascii=False),
+                "area_id": area_id,
+                "temporal_series_key": series_key(scope),
+                "observed_at": f"{year}-01-01T00:00:00Z",
+                "temporal_granularity": "year",
+                "source_version": f"{year}-definitive",
+            },
+            files={"file": (paths["artifact"].name, handle, "application/geo+json")},
+            timeout=timeout,
+        )
+    return response_data(response)
+
+
+def existing_dataset_for_year(datasets: list[dict[str, Any]], *, area_id: str, year: int) -> dict[str, Any] | None:
+    return next(
+        (
+            dataset
+            for dataset in datasets
+            if dataset.get("source_name") == SOURCE_NAME
+            and dataset.get("source_version") == f"{year}-definitive"
+            and str(dataset.get("area_id") or "") == area_id
+        ),
+        None,
+    )
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        years = parse_years(args.years)
+        if args.max_features < 1 or args.max_archive_mb < 1:
+            raise ValueError("Feature and archive safety limits must be positive")
+        scope = GEOGRAPHIC_SCOPES[args.scope]
+        base_url = args.base_url.rstrip("/")
+        with requests.Session() as api_session:
+            project_id, area_id, boundary, datasets = locate_workspace(api_session, base_url, scope, args.import_timeout)
+            results: list[dict[str, Any]] = []
+            with build_session() as official_session:
+                for year in years:
+                    paths, manifest = prepare_year(
+                        official_session,
+                        year=year,
+                        scope=scope,
+                        boundary_wgs84=boundary,
+                        output_root=args.output_root,
+                        request_timeout=args.request_timeout,
+                        max_archive_bytes=min(args.max_archive_mb * 1024 * 1024, MAX_ARCHIVE_BYTES),
+                        max_features=args.max_features,
+                        force=args.force,
+                    )
+                    existing = existing_dataset_for_year(datasets, area_id=area_id, year=year)
+                    if existing is not None:
+                        persisted_checksum = str(existing.get("checksum_sha256") or "")
+                        if persisted_checksum and persisted_checksum != manifest["artifact_sha256"]:
+                            raise RuntimeError(f"A different {year} agricultural parcel artifact is already persisted; refusing silent replacement")
+                        result = {"year": year, "status": "existing", "dataset_id": existing["id"], "feature_count": existing.get("feature_count")}
+                    elif args.fetch_only:
+                        result = {"year": year, "status": "prepared", "artifact_path": str(paths["artifact"]), "feature_count": manifest["feature_count"]}
+                    else:
+                        dataset = upload_artifact(
+                            api_session,
+                            base_url=base_url,
+                            project_id=project_id,
+                            area_id=area_id,
+                            scope=scope,
+                            year=year,
+                            paths=paths,
+                            manifest=manifest,
+                            timeout=args.import_timeout,
+                        )
+                        result = {"year": year, "status": "imported", "dataset_id": dataset["id"], "feature_count": dataset.get("feature_count")}
+                    results.append({**result, "area_ha": manifest["clipped_area_ha"], "manifest_path": str(paths["manifest"])})
+    except (KeyError, OSError, RuntimeError, ValueError, zipfile.BadZipFile, 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",
+                "scope": scope.key,
+                "project": scope.project_name,
+                "area": scope.area_name,
+                "series_key": series_key(scope),
+                "years": results,
+                "historical_identity_stable": False,
+            },
+            ensure_ascii=False,
+            indent=2,
+        )
+    )
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

scripts/run_readiness_check.sh

@@ -49,6 +49,7 @@ ${PYTHON_BIN} -m py_compile scripts/provision_mol_historical_landuse.py
 ${PYTHON_BIN} -m py_compile scripts/provision_official_landuse_timeseries.py
 ${PYTHON_BIN} -m py_compile scripts/provision_waterinfo_station_history.py
 ${PYTHON_BIN} -m py_compile scripts/provision_mol_bwk_natura2000.py
+${PYTHON_BIN} -m py_compile scripts/provision_agricultural_parcel_history.py
 ${PYTHON_BIN} -m py_compile scripts/provision_regional_timeseries.py
 ${PYTHON_BIN} -m py_compile scripts/geographic_scopes.py
 ${PYTHON_BIN} -m py_compile scripts/provision_geographic_scope.py