Add governed ALZ edition probe

backend/README.md

@@ -1444,7 +1444,8 @@ docker exec geointel python /app/scripts/audit_source_freshness.py \
 Use `--fail-on due` to make a planned review date fail automation, or
 `--fail-on never` for reporting only. The command never starts a refresh.
 
-An operator can explicitly add the official GRB and orthophoto edition check:
+An operator can explicitly add the official GRB, orthophoto and ALZ edition
+check:
 
 ```bash
 docker exec geointel python /app/scripts/audit_source_freshness.py \
@@ -1458,17 +1459,20 @@ Use `--refresh-catalogs` to bypass the 15-minute in-memory cache and
 `--fail-on-catalog` only when temporary official-provider unavailability must
 fail an operator job. This path reads bounded WFS/WMS capabilities and their
 fixed ISO 19139 metadata records. It confirms GRB `GBG`, `WBN`, `WGO`, `ADP`
-and orthophoto `Ortho`, `Vliegdagcontour`; it never requests feature or raster
-content. The public Datavindplaats API requires an access token, so release
-evidence comes from the public metadata links published by the official OGC
-services rather than HTML scraping.
+and orthophoto `Ortho`, `Vliegdagcontour`. It also reads the exact official ALZ
+publication page and validates only allowlisted archive-link identities. It
+never requests feature, raster or ALZ ZIP content. ALZ v1/v2 campaign snapshots
+remain provisional; only a v3 publication is compared with a local definitive
+historical edition.
 
 Runtime controls are `SOURCE_CATALOG_PROBE_ENABLED`,
-`SOURCE_CATALOG_GRB_WFS_URL`, `SOURCE_CATALOG_PROBE_TIMEOUT_SECONDS`,
-`SOURCE_CATALOG_PROBE_MAX_RESPONSE_MB` and
+`SOURCE_CATALOG_GRB_WFS_URL`, `SOURCE_CATALOG_ALZ_RELEASE_URL`,
+`SOURCE_CATALOG_PROBE_TIMEOUT_SECONDS`, `SOURCE_CATALOG_PROBE_MAX_RESPONSE_MB` and
 `SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS`. Metadata links remain restricted to
 the official HTTPS CSW path on `metadata.vlaanderen.be` even when an operator
-overrides the capabilities endpoint.
+overrides the capabilities endpoint. The ALZ release URL is fail-closed to the
+exact HTTPS host/path and cannot be redirected to another page or download
+host.
 
 ## Governed regional GRB refresh
 

backend/app/core/config.py

@@ -36,6 +36,10 @@ class Settings(BaseSettings):
         default="https://geo.api.vlaanderen.be/GRB/wfs",
         validation_alias="SOURCE_CATALOG_GRB_WFS_URL",
     )
+    source_catalog_alz_release_url: str = Field(
+        default="https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen",
+        validation_alias="SOURCE_CATALOG_ALZ_RELEASE_URL",
+    )
     source_catalog_probe_timeout_seconds: int = Field(
         default=10,
         ge=1,

backend/app/schemas/source_catalog.py

@@ -14,7 +14,7 @@ SourceCatalogComparisonStatus = Literal["same", "different", "not_comparable", "
 class SourceCatalogProbeItem(BaseModel):
     source_name: str
     display_name: str
-    service_type: Literal["WFS", "WMS"]
+    service_type: Literal["WFS", "WMS", "HTML"]
     endpoint_url: str
     status: SourceCatalogProbeStatus
     reachable: bool

backend/app/services/source_catalog_probe_service.py

@@ -4,6 +4,7 @@ from dataclasses import dataclass, replace
 from datetime import datetime, timedelta, timezone
 from email.utils import parsedate_to_datetime
 from hashlib import sha256
+from html.parser import HTMLParser
 import re
 from threading import Lock
 from typing import Any, Callable
@@ -32,6 +33,17 @@ _XLINK = "http://www.w3.org/1999/xlink"
 _METADATA_HOST = "metadata.vlaanderen.be"
 _VERSION_DATE = re.compile(r"^(?:toestand\s+)?(\d{4}-\d{2}-\d{2})$", re.IGNORECASE)
 _ORTHOPHOTO_EDITION = re.compile(r"^\d{4}\.\d{2}$")
+_ALZ_SOURCE_NAME = "agentschap_landbouw_zeevisserij_agricultural_parcels"
+_ALZ_RELEASE_HOST = "landbouwcijfers.vlaanderen.be"
+_ALZ_RELEASE_PATH = "/open-geodata-landbouwgebruikspercelen"
+_ALZ_DOWNLOAD_HOST = "www.landbouwvlaanderen.be"
+_ALZ_DOWNLOAD_PATH = re.compile(r"^/bestanden/gis/agpa_(20\d{2})_(\d{4}-\d{2}-\d{2})_public\.zip$")
+_ALZ_SNAPSHOT = re.compile(
+    r"^Landbouwgebruikspercelen\s+(20\d{2})\s*-\s*(\d+)e\s+snapshot\s*"
+    r"\(extractie\s+(\d{2}-\d{2}-\d{4})\)(?:\s*-\s*GPKG)?$",
+    re.IGNORECASE,
+)
+_ALZ_EDITION = re.compile(r"^(20\d{2})-(?:definitive|v3)$", re.IGNORECASE)
 
 
 class CatalogProbeFailure(RuntimeError):
@@ -88,6 +100,32 @@ class CacheEntry:
     probe: RemoteProbe
 
 
+class _AnchorParser(HTMLParser):
+    def __init__(self) -> None:
+        super().__init__(convert_charrefs=True)
+        self.anchors: list[tuple[str, str]] = []
+        self._href: str | None = None
+        self._text: list[str] = []
+
+    def handle_starttag(self, tag: str, attrs: list[tuple[str, str | None]]) -> None:
+        if tag.lower() != "a" or self._href is not None:
+            return
+        href = dict(attrs).get("href")
+        if href:
+            self._href = href.strip()
+            self._text = []
+
+    def handle_data(self, data: str) -> None:
+        if self._href is not None:
+            self._text.append(data)
+
+    def handle_endtag(self, tag: str) -> None:
+        if tag.lower() == "a" and self._href is not None:
+            self.anchors.append((self._href, "".join(self._text)))
+            self._href = None
+            self._text = []
+
+
 _REMOTE_CACHE: dict[str, CacheEntry] = {}
 _CACHE_LOCK = Lock()
 
@@ -124,7 +162,7 @@ def _with_capabilities_query(base_url: str, service_type: str) -> str:
 def _bounded_fetch(url: str, settings: Settings, opener: Callable[..., Any] | None = None) -> FetchResult:
     request = Request(
         url,
-        headers={"Accept": "application/xml,text/xml,application/json", "User-Agent": "GeoIntel/0.1 source-catalog-probe"},
+        headers={"Accept": "application/xml,text/xml,text/html,application/json", "User-Agent": "GeoIntel/0.1 source-catalog-probe"},
     )
     max_bytes = settings.source_catalog_probe_max_response_mb * 1024 * 1024
     try:
@@ -226,6 +264,127 @@ def _validate_capabilities_url(url: str) -> str:
     return url
 
 
+def _validate_alz_release_url(url: str) -> str:
+    parsed = urlsplit(url)
+    if (
+        parsed.scheme != "https"
+        or parsed.hostname != _ALZ_RELEASE_HOST
+        or parsed.port not in {None, 443}
+        or parsed.username
+        or parsed.password
+        or parsed.path.rstrip("/") != _ALZ_RELEASE_PATH
+        or parsed.query
+        or parsed.fragment
+    ):
+        raise CatalogProbeFailure(
+            "CATALOG_ALZ_RELEASE_URL_REJECTED",
+            "De ingestelde ALZ-publicatiepagina valt buiten de toegestane officiële URL.",
+        )
+    return url
+
+
+def _validate_alz_download_url(url: str) -> tuple[int, datetime]:
+    parsed = urlsplit(url)
+    match = _ALZ_DOWNLOAD_PATH.fullmatch(parsed.path)
+    if (
+        parsed.scheme != "https"
+        or parsed.hostname != _ALZ_DOWNLOAD_HOST
+        or parsed.port not in {None, 443}
+        or parsed.username
+        or parsed.password
+        or parsed.query
+        or parsed.fragment
+        or not match
+    ):
+        raise CatalogProbeFailure(
+            "CATALOG_ALZ_DOWNLOAD_URL_REJECTED",
+            "De ALZ-publicatiepagina bevat een datasetlink buiten de toegestane officiële URL-structuur.",
+        )
+    try:
+        published_at = datetime.strptime(match.group(2), "%Y-%m-%d").replace(tzinfo=timezone.utc)
+    except ValueError as exc:
+        raise CatalogProbeFailure(
+            "CATALOG_ALZ_DOWNLOAD_URL_REJECTED",
+            "De ALZ-datasetlink bevat geen geldige publicatiedatum.",
+        ) from exc
+    return int(match.group(1)), published_at
+
+
+def _normalized_html_text(value: str) -> str:
+    return re.sub(r"\s+", " ", value.replace("\xad", "").replace("–", "-").replace("—", "-")).strip()
+
+
+def _parse_alz_release_page(content: bytes) -> dict[str, Any]:
+    try:
+        html = content.decode("utf-8")
+    except UnicodeDecodeError as exc:
+        raise CatalogProbeFailure(
+            "CATALOG_ALZ_INVALID_HTML",
+            "De officiële ALZ-publicatiepagina is niet geldige UTF-8 HTML.",
+        ) from exc
+    parser = _AnchorParser()
+    try:
+        parser.feed(html)
+        parser.close()
+    except Exception as exc:
+        raise CatalogProbeFailure(
+            "CATALOG_ALZ_INVALID_HTML",
+            "De officiële ALZ-publicatiepagina kon niet veilig worden ontleed.",
+        ) from exc
+
+    definitive: list[tuple[int, datetime]] = []
+    snapshots: list[tuple[int, int, datetime]] = []
+    for href, raw_text in parser.anchors:
+        text = _normalized_html_text(raw_text)
+        looks_like_alz_release = (
+            text.casefold() == "downloaden"
+            or text.casefold().startswith("landbouwgebruikspercelen ")
+            or "agpa_" in href.casefold()
+        )
+        if not looks_like_alz_release:
+            continue
+        year, file_date = _validate_alz_download_url(href)
+        snapshot = _ALZ_SNAPSHOT.fullmatch(text)
+        if snapshot:
+            snapshot_year = int(snapshot.group(1))
+            snapshot_number = int(snapshot.group(2))
+            try:
+                extraction_date = datetime.strptime(snapshot.group(3), "%d-%m-%Y").replace(tzinfo=timezone.utc)
+            except ValueError as exc:
+                raise CatalogProbeFailure(
+                    "CATALOG_ALZ_SNAPSHOT_INVALID",
+                    "De actuele ALZ-snapshot bevat geen geldige extractiedatum.",
+                ) from exc
+            if snapshot_year != year or extraction_date != file_date or snapshot_number not in {1, 2, 3}:
+                raise CatalogProbeFailure(
+                    "CATALOG_ALZ_SNAPSHOT_INVALID",
+                    "De actuele ALZ-snapshot is niet consistent met de officiële datasetlink.",
+                )
+            snapshots.append((year, snapshot_number, extraction_date))
+        elif text.casefold() == "downloaden":
+            definitive.append((year, file_date))
+        else:
+            raise CatalogProbeFailure(
+                "CATALOG_ALZ_RELEASE_UNRECOGNIZED",
+                "De ALZ-publicatiepagina bevat een niet-herkende landbouwdatasetpublicatie.",
+            )
+
+    if not definitive:
+        raise CatalogProbeFailure(
+            "CATALOG_ALZ_DEFINITIVE_MISSING",
+            "De officiële ALZ-publicatiepagina bevat geen herkenbare definitieve landbouwperceeleditie.",
+        )
+    latest_definitive = max(definitive, key=lambda item: (item[0], item[1]))
+    latest_snapshot = max(snapshots, key=lambda item: (item[0], item[1], item[2])) if snapshots else None
+    if latest_snapshot and latest_snapshot[1] == 3 and latest_snapshot[0] >= latest_definitive[0]:
+        latest_definitive = (latest_snapshot[0], latest_snapshot[2])
+    return {
+        "definitive": latest_definitive,
+        "definitive_count": len(definitive),
+        "snapshot": latest_snapshot,
+    }
+
+
 def _node_text(node: ElementTree.Element | None) -> str | None:
     if node is None:
         return None
@@ -282,6 +441,74 @@ def _parse_metadata(content: bytes) -> dict[str, Any]:
     }
 
 
+def _probe_alz_remote(
+    contract: ProbeContract,
+    settings: Settings,
+    *,
+    opener: Callable[..., Any] | None,
+    now: datetime,
+) -> RemoteProbe:
+    release_url = _validate_alz_release_url(contract.endpoint_url)
+    response = _bounded_fetch(release_url, settings, opener)
+    _validate_alz_release_url(response.final_url)
+    content_type = response.content_type.lower()
+    if content_type and "html" not in content_type and "text" not in content_type:
+        raise CatalogProbeFailure(
+            "CATALOG_ALZ_INVALID_CONTENT_TYPE",
+            "De officiële ALZ-publicatiepagina is geen HTML-respons.",
+        )
+    release = _parse_alz_release_page(response.content)
+    definitive_year, definitive_date = release["definitive"]
+    snapshot = release["snapshot"]
+    matched = ["definitive_archive"]
+    missing: list[str] = []
+    if snapshot:
+        matched.append("current_snapshot")
+    else:
+        missing.append("current_snapshot")
+
+    definitive_version = f"{definitive_year}-v3"
+    if snapshot:
+        snapshot_year, snapshot_number, snapshot_date = snapshot
+        snapshot_version = f"{snapshot_year}-v{snapshot_number}"
+        if snapshot_number < 3:
+            message = (
+                f"De officiële ALZ-publicatiepagina bevestigt definitieve editie {definitive_version}. "
+                f"De actuele publicatie {snapshot_version} van {snapshot_date.date().isoformat()} is voorlopig "
+                "en wordt niet als historische vervanging aangemerkt."
+            )
+        else:
+            message = f"De officiële ALZ-publicatiepagina bevestigt definitieve editie {definitive_version}."
+        remote_title = f"Landbouwgebruikspercelen {definitive_version}; actuele publicatie {snapshot_version}"
+    else:
+        message = (
+            f"De officiële ALZ-publicatiepagina bevestigt definitieve editie {definitive_version}, "
+            "maar bevat geen herkenbare actuele snapshot."
+        )
+        remote_title = f"Landbouwgebruikspercelen {definitive_version}"
+
+    return RemoteProbe(
+        status="available" if snapshot else "degraded",
+        reachable=True,
+        checked_at=now,
+        expected_layers=contract.expected_layers,
+        matched_layers=tuple(matched),
+        missing_layers=tuple(missing),
+        advertised_layer_count=release["definitive_count"] + (1 if snapshot else 0),
+        metadata_url=response.final_url,
+        metadata_identifier="alz-agricultural-use-parcels",
+        remote_title=remote_title,
+        remote_version=definitive_version,
+        remote_modified_at=response.last_modified_at,
+        remote_published_at=definitive_date,
+        capabilities_sha256=sha256(response.content).hexdigest(),
+        capabilities_etag=response.etag,
+        capabilities_last_modified_at=response.last_modified_at,
+        message=message,
+        error_code="CATALOG_ALZ_CURRENT_SNAPSHOT_MISSING" if not snapshot else None,
+    )
+
+
 def _probe_remote(
     contract: ProbeContract,
     settings: Settings,
@@ -290,6 +517,8 @@ def _probe_remote(
     now: datetime,
 ) -> RemoteProbe:
     try:
+        if contract.service_type == "HTML":
+            return _probe_alz_remote(contract, settings, opener=opener, now=now)
         capabilities = _bounded_fetch(_validate_capabilities_url(contract.endpoint_url), settings, opener)
         _validate_capabilities_url(capabilities.final_url)
         content_type = capabilities.content_type.lower()
@@ -394,6 +623,18 @@ def _latest_local_version(source_name: str, datasets: list[Dataset]) -> str | No
         ]
         if explicit_current:
             candidates = explicit_current
+    elif source_name == _ALZ_SOURCE_NAME:
+        definitive = [item for item in candidates if _ALZ_EDITION.fullmatch((item.source_version or "").strip())]
+        if definitive:
+            latest = max(
+                definitive,
+                key=lambda item: (
+                    int(_ALZ_EDITION.fullmatch((item.source_version or "").strip()).group(1)),
+                    _utc(item.observed_at) if item.observed_at else datetime.min.replace(tzinfo=timezone.utc),
+                    str(item.id),
+                ),
+            )
+            return latest.source_version
     if not candidates:
         return None
     latest = max(
@@ -416,6 +657,9 @@ def _normalized_version(source_name: str, version: str | None) -> str | None:
         return match.group(1) if match else None
     if source_name == "digitaal_vlaanderen_orthophoto" and _ORTHOPHOTO_EDITION.fullmatch(value):
         return value
+    if source_name == _ALZ_SOURCE_NAME:
+        match = _ALZ_EDITION.fullmatch(value)
+        return f"{match.group(1)}-v3" if match else None
     return None
 
 
@@ -467,6 +711,13 @@ class SourceCatalogProbeService:
                 endpoint_url=_with_capabilities_query(active_settings.orthophoto_wms_url, "WMS"),
                 expected_layers=(active_settings.orthophoto_wms_layer, "Vliegdagcontour"),
             ),
+            ProbeContract(
+                source_name=_ALZ_SOURCE_NAME,
+                display_name="Landbouwgebruikspercelen (ALZ)",
+                service_type="HTML",
+                endpoint_url=active_settings.source_catalog_alz_release_url,
+                expected_layers=("definitive_archive", "current_snapshot"),
+            ),
         )
         items: list[SourceCatalogProbeItem] = []
         for contract in contracts:
@@ -538,8 +789,9 @@ class SourceCatalogProbeService:
             summary=summary,
             items=items,
             limitations=[
-                "Deze expliciete controle leest alleen allowlisted WFS/WMS-capabilities en gekoppelde ISO 19139 metadata.",
+                "Deze expliciete controle leest alleen allowlisted WFS/WMS-capabilities, gekoppelde ISO 19139 metadata en de officiële ALZ-publicatiepagina.",
                 "Er worden geen features, rasters of modelbestanden opgehaald en geen datasets aangemaakt of overschreven.",
+                "Voor ALZ is alleen de nieuwste definitieve v3-editie vergelijkbaar; voorlopige v1/v2-snapshots zijn uitsluitend informatief.",
                 "Een versieverschil is controlesignaal, geen bewijs dat een lokale dataset onbruikbaar is en geen automatische importopdracht.",
             ],
         )

backend/tests/test_sprint222_source_catalog_probes.py

@@ -21,6 +21,7 @@ ORTHO_METADATA_URL = (
     "https://metadata.vlaanderen.be/srv/dut/csw?request=GetRecordById&service=CSW&"
     "id=f5304d6d-0dd4-43fd-a726-427af31e8d61&OUTPUTSCHEMA=http://www.isotc211.org/2005/gmd"
 )
+ALZ_RELEASE_URL = "https://landbouwcijfers.vlaanderen.be/open-geodata-landbouwgebruikspercelen"
 
 
 def _wfs_capabilities(metadata_url: str = GRB_METADATA_URL) -> bytes:
@@ -77,6 +78,23 @@ def _metadata(identifier: str, title: str, edition: str, modified: str, publishe
     """.encode()
 
 
+def _alz_release_page(*, include_snapshot: bool = True, download_host: str = "www.landbouwvlaanderen.be") -> bytes:
+    snapshot = (
+        f'<a href="https://{download_host}/bestanden/gis/agpa_2026_2026-06-02_public.zip">'
+        "Landbouwgebruikspercelen 2026 – 1e snapshot (extractie 02-06-2026) - GPKG</a>"
+        if include_snapshot
+        else ""
+    )
+    return f"""
+    <html><body>
+      {snapshot}
+      <p>Definitieve datasets</p>
+      <a href="https://{download_host}/bestanden/gis/agpa_2025_2026-05-13_public.zip">Downloaden</a>
+      <a href="https://{download_host}/bestanden/gis/agpa_2024_2025-03-27_public.zip">Downloaden</a>
+    </body></html>
+    """.encode()
+
+
 class _Response:
     def __init__(self, content: bytes, content_type: str = "text/xml", *, content_length: int | None = None) -> None:
         self.content = content
@@ -97,8 +115,8 @@ class _Response:
 
 
 class _RedirectedResponse(_Response):
-    def __init__(self, content: bytes, final_url: str) -> None:
-        super().__init__(content)
+    def __init__(self, content: bytes, final_url: str, content_type: str = "text/xml") -> None:
+        super().__init__(content, content_type)
         self.final_url = final_url
 
     def geturl(self) -> str:
@@ -144,6 +162,7 @@ def _settings(**overrides) -> Settings:
     values = {
         "SOURCE_CATALOG_PROBE_ENABLED": True,
         "SOURCE_CATALOG_GRB_WFS_URL": "https://geo.api.vlaanderen.be/GRB/wfs",
+        "SOURCE_CATALOG_ALZ_RELEASE_URL": ALZ_RELEASE_URL,
         "SOURCE_CATALOG_PROBE_TIMEOUT_SECONDS": 3,
         "SOURCE_CATALOG_PROBE_MAX_RESPONSE_MB": 1,
         "SOURCE_CATALOG_PROBE_CACHE_TTL_SECONDS": 0,
@@ -157,6 +176,8 @@ def _settings(**overrides) -> Settings:
 def _opener(request, timeout):
     assert timeout == 3
     url = request.full_url
+    if url == ALZ_RELEASE_URL:
+        return _Response(_alz_release_page(), "text/html; charset=utf-8")
     if "metadata.vlaanderen.be" in url:
         if "f5304d6d" in url:
             return _Response(_metadata("f5304d6d", "Orthofoto meest recent, 2025.04", "2025.04", "2026-04-27", "2025-12-11"))
@@ -170,7 +191,13 @@ def test_catalog_probe_reads_real_editions_and_compares_only_compatible_versions
     SourceCatalogProbeService.clear_cache()
     project_id = uuid.uuid4()
     report = SourceCatalogProbeService.audit_project(
-        _Db([_dataset("grb", "2026-07-14"), _dataset("digitaal_vlaanderen_orthophoto", "most_recent_at_2026-07-14")]),
+        _Db(
+            [
+                _dataset("grb", "2026-07-14"),
+                _dataset("digitaal_vlaanderen_orthophoto", "most_recent_at_2026-07-14"),
+                _dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2025-definitive"),
+            ]
+        ),
         project_id,
         settings=_settings(),
         opener=_opener,
@@ -178,7 +205,7 @@ def test_catalog_probe_reads_real_editions_and_compares_only_compatible_versions
     )
     by_source = {item.source_name: item for item in report.items}
 
-    assert report.summary.available_count == 2
+    assert report.summary.available_count == 3
     assert report.summary.different_version_count == 1
     assert by_source["grb"].remote_version == "Toestand 2026-07-15"
     assert by_source["grb"].comparison_status == "different"
@@ -187,6 +214,13 @@ def test_catalog_probe_reads_real_editions_and_compares_only_compatible_versions
     assert by_source["digitaal_vlaanderen_orthophoto"].remote_version == "2025.04"
     assert by_source["digitaal_vlaanderen_orthophoto"].comparison_status == "not_comparable"
     assert by_source["digitaal_vlaanderen_orthophoto"].remote_published_at.year == 2025
+    alz = by_source["agentschap_landbouw_zeevisserij_agricultural_parcels"]
+    assert alz.service_type == "HTML"
+    assert alz.remote_version == "2025-v3"
+    assert alz.comparison_status == "same"
+    assert alz.matched_layers == ["definitive_archive", "current_snapshot"]
+    assert "2026-v1" in alz.message
+    assert "voorlopig" in alz.message
     assert all(item.capabilities_sha256 for item in report.items)
 
 
@@ -275,10 +309,10 @@ def test_catalog_probe_cache_is_explicitly_bypassable() -> None:
     second = SourceCatalogProbeService.audit_project(db, project_id, settings=settings, opener=counting_opener, now=NOW)
     SourceCatalogProbeService.audit_project(db, project_id, settings=settings, opener=counting_opener, now=NOW, force=True)
 
-    assert first_call_count == 4
+    assert first_call_count == 5
     assert all(item.cached is False for item in first.items)
     assert all(item.cached is True for item in second.items)
-    assert len(calls) == 8
+    assert len(calls) == 10
 
 
 def test_catalog_probe_can_be_disabled_without_network_access() -> None:
@@ -293,7 +327,7 @@ def test_catalog_probe_can_be_disabled_without_network_access() -> None:
         now=NOW,
     )
 
-    assert report.summary.disabled_count == 2
+    assert report.summary.disabled_count == 3
     assert all(item.status == "disabled" for item in report.items)
 
 
@@ -316,7 +350,7 @@ def test_catalog_probe_route_returns_canonical_envelope(monkeypatch) -> None:
 
     assert list(response) == ["data"]
     assert response["data"]["project_id"] == project_id
-    assert response["data"]["summary"]["provider_count"] == 2
+    assert response["data"]["summary"]["provider_count"] == 3
 
 
 def test_catalog_probe_remains_explicit_and_never_imports_provider_data() -> None:
@@ -330,3 +364,75 @@ def test_catalog_probe_remains_explicit_and_never_imports_provider_data() -> Non
     assert "VectorFeatureService" not in service
     assert "--probe-catalogs" in operator
     assert "/datasets/source-catalog-probes" in operator
+
+
+def test_alz_catalog_probe_rejects_release_page_redirect_outside_allowlist() -> None:
+    def redirected_opener(request, timeout):
+        if request.full_url == ALZ_RELEASE_URL:
+            return _RedirectedResponse(
+                _alz_release_page(),
+                "https://example.com/open-geodata-landbouwgebruikspercelen",
+                "text/html",
+            )
+        return _opener(request, timeout)
+
+    report = SourceCatalogProbeService.audit_project(
+        _Db([]), uuid.uuid4(), settings=_settings(), opener=redirected_opener, now=NOW
+    )
+    alz = next(item for item in report.items if item.source_name == "agentschap_landbouw_zeevisserij_agricultural_parcels")
+
+    assert alz.status == "unavailable"
+    assert alz.error_code == "CATALOG_ALZ_RELEASE_URL_REJECTED"
+
+
+def test_alz_catalog_probe_rejects_untrusted_download_host() -> None:
+    def malicious_opener(request, timeout):
+        if request.full_url == ALZ_RELEASE_URL:
+            return _Response(_alz_release_page(download_host="example.com"), "text/html")
+        return _opener(request, timeout)
+
+    report = SourceCatalogProbeService.audit_project(
+        _Db([]), uuid.uuid4(), settings=_settings(), opener=malicious_opener, now=NOW
+    )
+    alz = next(item for item in report.items if item.source_name == "agentschap_landbouw_zeevisserij_agricultural_parcels")
+
+    assert alz.status == "unavailable"
+    assert alz.error_code == "CATALOG_ALZ_DOWNLOAD_URL_REJECTED"
+
+
+def test_alz_catalog_probe_degrades_without_current_snapshot_but_keeps_definitive_evidence() -> None:
+    def archive_only_opener(request, timeout):
+        if request.full_url == ALZ_RELEASE_URL:
+            return _Response(_alz_release_page(include_snapshot=False), "text/html")
+        return _opener(request, timeout)
+
+    report = SourceCatalogProbeService.audit_project(
+        _Db([_dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2024-definitive")]),
+        uuid.uuid4(),
+        settings=_settings(),
+        opener=archive_only_opener,
+        now=NOW,
+    )
+    alz = next(item for item in report.items if item.source_name == "agentschap_landbouw_zeevisserij_agricultural_parcels")
+
+    assert alz.status == "degraded"
+    assert alz.remote_version == "2025-v3"
+    assert alz.comparison_status == "different"
+    assert alz.matched_layers == ["definitive_archive"]
+    assert alz.missing_layers == ["current_snapshot"]
+    assert alz.error_code == "CATALOG_ALZ_CURRENT_SNAPSHOT_MISSING"
+
+
+def test_alz_catalog_probe_compares_latest_definitive_year_not_latest_import_order() -> None:
+    older = _dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2024-definitive")
+    newer = _dataset("agentschap_landbouw_zeevisserij_agricultural_parcels", "2025-definitive")
+    older.imported_at = NOW
+    newer.imported_at = NOW.replace(year=2025)
+
+    report = SourceCatalogProbeService.audit_project(
+        _Db([older, newer]), uuid.uuid4(), settings=_settings(), opener=_opener, now=NOW
+    )
+    alz = next(item for item in report.items if item.source_name == "agentschap_landbouw_zeevisserij_agricultural_parcels")
+
+    assert alz.local_source_version == "2025-definitive"
+    assert alz.comparison_status == "same"