fix(wire): run-0 pre-flight — declared-region curation at the delivery

docs/ADRs/013_sampled_forecast_wire_contract.md

@@ -1044,6 +1044,14 @@ record execution progress against it.
   (maintainer: "fix this through the pipeline"): lifting the `<17` ceiling at its
   source is filed as pipeline-core#280; when it lifts, the fixture gets one
   planned re-baseline, coordinated, never a drive-by.**
+- **2026-07-24 — run-0 pre-flight: declared-region curation added at the
+  anti-corruption layer.** The real run-0 payload carries the full model grid
+  (64,818 cells) including exactly the 76 declared GAUL-uncovered exclusions
+  (C-30); the producer is right not to know partner curation, so the delivery now
+  applies `coverage.excluded_for(region)` — the explicit pinned frozenset, never
+  inference — to the assembled frames before the gate (contract-mode read).
+  Proven by a full dry run on the real run-0 data: curated 64,742 cells, gate
+  passed at 128 draws, sidecar + manifest staged, zero store calls.
 - **2026-07-20 — HOP-B SINK LEG SHIPPED (epic #105 complete; upload-disabled).**
   The contract's missing middle exists in fixture-proven code:
   `unfao/wire/` (naming, header, shard, sidecar, run_manifest, source_selection,

tests/test_frame_extraction.py

@@ -45,3 +45,24 @@ def test_months_of_is_int64_ascending():
     out = frame_extraction.months_of(pf)
     assert out.dtype == np.int64
     assert list(out) == sorted(out)
+
+
+def test_drop_units_removes_only_declared_cells():
+    import numpy as np
+
+    from views_postprocessing.unfao.frame_extraction import drop_units
+    from views_postprocessing.unfao.frames import build_prediction_frame
+
+    values = np.arange(12, dtype=np.float32).reshape(6, 2)
+    time = np.full(6, 543, dtype=np.int64)
+    unit = np.array([1, 2, 3, 4, 5, 6], dtype=np.int64)
+    frame = build_prediction_frame(values, time, unit)
+
+    curated = drop_units(frame, frozenset({2, 5}))
+    assert list(np.asarray(curated.index.unit)) == [1, 3, 4, 6]
+    np.testing.assert_array_equal(curated.values, values[[0, 2, 3, 5]])
+    # rows survive intact and aligned; empty exclusion is the identity
+    assert drop_units(frame, frozenset()) is frame
+    # excluded cells absent from the frame are a no-op, not an error (declared
+    # exclusions describe the region, not this payload)
+    assert drop_units(frame, frozenset({99})).n_rows == 6

views_postprocessing/unfao/frame_extraction.py

@@ -39,6 +39,27 @@ def months_of(frame: PredictionFrame) -> NDArray[np.int64]:
     return np.unique(np.asarray(frame.index.time, dtype=np.int64))
 
 
+def drop_units(frame: PredictionFrame, excluded: frozenset) -> PredictionFrame:
+    """The frame without the DECLARED excluded cells (row filter on ``unit``).
+
+    Product curation as a seam concern (ADR-013 anti-corruption role): the producer
+    publishes its full model grid; the delivery restricts it to the declared
+    partner region using an explicit exclusion set (e.g.
+    ``delivery.coverage.excluded_for(region)``) — never inferred. Empty exclusion
+    returns the frame unchanged.
+    """
+    if not excluded:
+        return frame
+    unit = np.asarray(frame.index.unit, dtype=np.int64)
+    keep = ~np.isin(unit, np.fromiter(excluded, dtype=np.int64))
+    if keep.all():
+        return frame
+    time = np.asarray(frame.index.time, dtype=np.int64)
+    from views_postprocessing.unfao.frames import build_prediction_frame
+
+    return build_prediction_frame(frame.values[keep], time[keep], unit[keep])
+
+
 def month_slice(
     frame: PredictionFrame, month_id: int
 ) -> tuple[NDArray[np.float32], NDArray[np.int64], NDArray[np.int64]]:

views_postprocessing/unfao/managers/unfao.py

@@ -152,6 +152,21 @@ def _read_forecast_data_contract(self):
             expected_targets=product.TARGETS,
             expected_ensemble=self.configs["ensemble"],
         )
+        # Declared-region curation (C-30/S4, at the anti-corruption layer): the
+        # producer publishes its full model grid; the FAO product excludes the
+        # declared GAUL-uncovered cells. Explicit frozenset, never inference.
+        excluded = coverage.excluded_for(self.configs.get("region"))
+        if excluded:
+            self._forecast_run = {
+                target: (frame_extraction.drop_units(frame, excluded), headers)
+                for target, (frame, headers) in self._forecast_run.items()
+            }
+            logger.info(
+                "Declared-region curation applied: %d excluded cells dropped per "
+                "target (region=%r).",
+                len(excluded),
+                self.configs.get("region"),
+            )
         run_id = next(iter(self._forecast_run.values()))[1][0]["run_id"]
         logger.info(
             "Contract inbound: run %s assembled (%d targets).",