From bd440d75cd6b804838b85d2a70956013cf013d88 Mon Sep 17 00:00:00 2001 From: Polichinl Date: Fri, 24 Jul 2026 03:39:15 +0200 Subject: [PATCH] =?UTF-8?q?fix(wire):=20run-0=20pre-flight=20=E2=80=94=20d?= =?UTF-8?q?eclared-region=20curation=20at=20the=20delivery=20(C-30=20exclu?= =?UTF-8?q?sions=20applied=20to=20contract-mode=20frames)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The real run-0 payload carries the producer's full model grid (64,818 cells) incl. exactly the 76 declared land_gaul GAUL-uncovered exclusions; uncurated it fails loud at the sidecar builder. Curation is partner-product policy, so it lands at the anti-corruption layer: frame_extraction.drop_units applies the explicit coverage.excluded_for(region) frozenset (declared, never inferred) to assembled frames in the contract read; empty exclusion = identity. Dry-run proven on the real run-0 arrays: 64,742 curated cells, §6 gate passed at 128 draws, sidecar+manifest staged, zero store calls. ADR post-adoption entry added. Co-Authored-By: Claude Fable 5 --- .../013_sampled_forecast_wire_contract.md | 8 +++++++ tests/test_frame_extraction.py | 21 +++++++++++++++++++ .../unfao/frame_extraction.py | 21 +++++++++++++++++++ views_postprocessing/unfao/managers/unfao.py | 15 +++++++++++++ 4 files changed, 65 insertions(+) diff --git a/docs/ADRs/013_sampled_forecast_wire_contract.md b/docs/ADRs/013_sampled_forecast_wire_contract.md index 6cd015f..ada9e70 100644 --- a/docs/ADRs/013_sampled_forecast_wire_contract.md +++ b/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, diff --git a/tests/test_frame_extraction.py b/tests/test_frame_extraction.py index 066fae9..846837b 100644 --- a/tests/test_frame_extraction.py +++ b/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 diff --git a/views_postprocessing/unfao/frame_extraction.py b/views_postprocessing/unfao/frame_extraction.py index 08da2dd..c6aaaab 100644 --- a/views_postprocessing/unfao/frame_extraction.py +++ b/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]]: diff --git a/views_postprocessing/unfao/managers/unfao.py b/views_postprocessing/unfao/managers/unfao.py index 1259310..42ad944 100644 --- a/views_postprocessing/unfao/managers/unfao.py +++ b/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).",