From 125a087a8861b1efab1ba18cb7a818f5d520ac7c Mon Sep 17 00:00:00 2001 From: Polichinl Date: Tue, 25 Aug 2026 10:37:54 +0200 Subject: [PATCH] feat(delivery): stamp the observed-range boundary into provenance (#297) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit CRAF'd received July 2026 as history. #297 established the month was real but ~1% reported — six cells of 64,742, zero in `ged_sb` but non-zero in `ged_ns` and `ged_os` — so the producer's inferred boundary declared it observed and the clip kept it, correctly, by its own contract. Establishing that took a day, because the artifact could not answer the question a partner asks afterwards: observed through when, and decided against what? The boundary was recoverable only because those six cells happened to land in the two columns that leave a trace. In `ged_sb` the data would have been mute. `build_provenance` now requires `observed_through` and always emits it. Required rather than optional because a caller that forgets it is the failure being fixed; always emitted because an absent key is indistinguishable from an artifact built before the field existed. An explicit null carries the degrade-open case — the boundary could not be read, so this delivery was NOT clipped — which is the case that most needs recording and exactly the one an omit-when-absent field drops. A third state is kept distinct: `UNREAD`, the managers' initial value, refuses at build time. Letting it collapse to null would report "clip skipped" for a run whose clip in fact ran — the C-103 conflation, one layer down. `observed_through` joins the essential set in `compact_description`, or the 255-char fallback would drop it precisely when descriptions are long, and joins the redaction guard's declared keyset deliberately: a month_id integer, no PII, no credential, and what the partner needs to tell a sparse month from a fabricated one. Both partners, not just CRAF'd. The managers are deliberate clones and the UN-FAO side has the external partner; a fix in one is half a fix. A test asserts they do not diverge. Deferred with a trigger: the producer also publishes a per-source `last_valid_month_ids` map. `datafactory_query.defaults` exposes only the scalar, and a multi-source map would not fit the 255-char carrier. When C-15's structured metadata field lands upstream and the ceiling goes, stamp the map too. 12 tests: the rule on primitives, and the wiring as declaration checks. One asserts the assignment precedes the degrade-open early return — placed after it, the path that most needs recording would raise at provenance time instead. Co-Authored-By: Claude Opus 5 (1M context) --- tests/test_input_integrity_e2e.py | 2 + tests/test_observed_range_provenance.py | 188 +++++++++++++++++++ tests/test_provenance.py | 8 + tests/test_redaction_guard.py | 7 + tests/test_selection_guard.py | 1 + views_postprocessing/crafd/managers/crafd.py | 8 + views_postprocessing/delivery/provenance.py | 53 +++++- views_postprocessing/unfao/managers/unfao.py | 8 + 8 files changed, 274 insertions(+), 1 deletion(-) create mode 100644 tests/test_observed_range_provenance.py diff --git a/tests/test_input_integrity_e2e.py b/tests/test_input_integrity_e2e.py index 8e0a5a8..20a9361 100644 --- a/tests/test_input_integrity_e2e.py +++ b/tests/test_input_integrity_e2e.py @@ -100,6 +100,7 @@ def test_s5_upload_description_carries_structured_provenance(): expected_cell_count=coverage.expected_for("land_gaul"), actual_cell_count=3, unmapped_count=0, + observed_through=559, ) description = f"Enriched ... provenance={json.dumps(prov, separators=(',', ':'))}" @@ -120,5 +121,6 @@ def test_s5_provenance_records_unmapped_cells_when_present(): expected_cell_count=coverage.expected_for("land_gaul"), actual_cell_count=3, unmapped_count=1, + observed_through=559, ) assert prov["unmapped_count"] == 1 diff --git a/tests/test_observed_range_provenance.py b/tests/test_observed_range_provenance.py new file mode 100644 index 0000000..7fdc14f --- /dev/null +++ b/tests/test_observed_range_provenance.py @@ -0,0 +1,188 @@ +"""#297: the artifact must say what boundary it clipped against. + +CRAF'd received July 2026 as history. The month was real but ~1% reported — six +cells of 64,742 — and the producer's *inferred* boundary (a month counts as +observed once its slice sums above zero) therefore declared it observed. The clip +kept it, correctly, by its own contract. + +Establishing that took a day, because the delivered artifact could not answer the +one question a partner asks afterwards: *observed through when, and decided against +what?* The boundary was only recoverable at all because July's six cells happened to +land in ``ged_ns``/``ged_os`` rather than ``ged_sb``, leaving a non-zero trace. Had +they landed in ``ged_sb``, the data would have been mute. + +So the boundary is stamped, and stamped **unconditionally**. The degrade-open case — +boundary unreadable, clip skipped, unobserved months may be present — is the case that +most needs recording, and it is exactly the case an "omit when absent" field would drop. + +Two layers, as the repo tests every other delivery invariant: the rule on primitives, +and the wiring as declaration checks (constructing a manager needs pipeline-core, a +views-models path manager and a live Appwrite environment — C-40). +""" + +from __future__ import annotations + +import ast +import json +from pathlib import Path + +import pytest + +from tests.conftest import PARTNER_PACKAGES +from views_postprocessing.delivery.provenance import ( + DESCRIPTION_MAX, + UNREAD, + build_provenance, + compact_description, +) + +_REPO = Path(__file__).resolve().parent.parent + + +def _prov(**over): + base = dict( + lookup_version="gaul-2024a", + region="land_gaul", + expected_cell_count=64742, + actual_cell_count=64742, + unmapped_count=0, + observed_through=559, + ) + base.update(over) + return build_provenance(**base) + + +# ── the rule ─────────────────────────────────────────────────────────────── + + +def test_the_boundary_the_clip_used_is_stamped(): + assert _prov(observed_through=559)["observed_through"] == 559 + + +def test_a_skipped_clip_is_stamped_as_null_rather_than_omitted(): + """The #297 property. An absent key is indistinguishable from an artifact + built before this field existed; an explicit null says "the boundary could + not be read, so this delivery was NOT clipped".""" + prov = _prov(observed_through=None) + assert "observed_through" in prov + assert prov["observed_through"] is None + assert json.loads(compact_description(prov))["observed_through"] is None + + +def test_building_provenance_without_reading_the_boundary_refuses(): + """UNREAD is a call-order bug, not a delivery condition. It must not + silently become null — that would report "clip skipped" for a run whose + clip in fact ran.""" + with pytest.raises(ValueError, match="never read"): + _prov(observed_through=UNREAD) + + +def test_unread_is_distinct_from_none(): + assert UNREAD is not None + assert not isinstance(None, type(UNREAD)) + + +def test_the_boundary_survives_the_compaction_fallback(): + """``compact_description`` drops non-essential keys when the 255-char carrier + overflows. A boundary that vanishes precisely when the description is long is + a guard that disappears when it is needed, so it belongs in the essential set.""" + # The padding must overflow the full dict while leaving the essential set inside + # the limit. Measured 2026-08-25: the fallback triggers from 104 chars and the + # essential set still fits to ~117. Both assertions below fail loudly if that + # window ever moves, so the constant cannot drift silently into a vacuous test. + prov = _prov(lookup_version="g" * 110, fill_count=3) + text = compact_description(prov) + assert len(text) <= DESCRIPTION_MAX + round_trip = json.loads(text) + assert "fill_count" not in round_trip, "the fallback did not trigger; test is vacuous" + assert round_trip["observed_through"] == 559 + + +# ── the wiring ───────────────────────────────────────────────────────────── + + +def _manager_source(partner: str) -> str: + return (_REPO / "views_postprocessing" / partner / "managers" / f"{partner}.py").read_text() + + +def _func(source: str, name: str) -> ast.FunctionDef: + return next( + n for n in ast.walk(ast.parse(source)) + if isinstance(n, ast.FunctionDef) and n.name == name + ) + + +@pytest.mark.parametrize("partner", PARTNER_PACKAGES) +def test_the_manager_initialises_the_boundary_as_unread(partner): + init = _func(_manager_source(partner), "__init__") + assigns = [ + n for n in ast.walk(init) + if isinstance(n, ast.Assign) + and any( + isinstance(t, ast.Attribute) and t.attr == "_observed_through" + for t in n.targets + ) + ] + assert assigns, f"{partner}: __init__ does not initialise _observed_through" + src = ast.unparse(assigns[0].value) + assert "UNREAD" in src, ( + f"{partner}: _observed_through initialises to {src!r}, not UNREAD. " + "Initialising to None makes 'never read' indistinguishable from 'read and " + "unavailable' — the exact conflation #297 exists to prevent." + ) + + +@pytest.mark.parametrize("partner", PARTNER_PACKAGES) +def test_the_boundary_is_recorded_before_the_degrade_open_return(partner): + """The assignment must precede the ``if lv is None: return`` early exit. + + Placed after it, the degrade-open path — the one that most needs recording — + would leave the attribute UNREAD and the delivery would raise at provenance + time instead of reporting that it did not clip. + """ + read = _func(_manager_source(partner), "_read_historical_frame") + assigns = [ + n for n in ast.walk(read) + if isinstance(n, ast.Assign) + and any( + isinstance(t, ast.Attribute) and t.attr == "_observed_through" + for t in n.targets + ) + ] + assert assigns, f"{partner}: _read_historical_frame never records the boundary" + + returns = [n for n in ast.walk(read) if isinstance(n, ast.Return)] + assert returns, f"{partner}: expected an early return in _read_historical_frame" + first_return = min(n.lineno for n in returns) + assert min(a.lineno for a in assigns) < first_return, ( + f"{partner}: _observed_through is assigned at or after the early return, so " + "the degrade-open path would never record that the clip was skipped." + ) + + +@pytest.mark.parametrize("partner", PARTNER_PACKAGES) +def test_the_manager_passes_the_boundary_into_provenance(partner): + desc = _func(_manager_source(partner), "_historical_frame_description") + kwargs = { + kw.arg + for c in ast.walk(desc) + if isinstance(c, ast.Call) + for kw in c.keywords + if kw.arg + } + assert "observed_through" in kwargs, ( + f"{partner}: _historical_frame_description builds provenance without the " + "boundary. build_provenance requires it, so this would raise at delivery." + ) + + +def test_both_partners_carry_it_identically(): + """#297 was filed as a CRAF'd defect; the two managers are deliberate clones + and the UN-FAO side has an external partner. A fix in one only is half a fix.""" + counts = { + p: _manager_source(p).count("_observed_through") for p in PARTNER_PACKAGES + } + assert len(set(counts.values())) == 1, ( + f"the partners diverge on the boundary stamp: {counts}" + ) + assert all(v >= 3 for v in counts.values()), counts diff --git a/tests/test_provenance.py b/tests/test_provenance.py index 0ac3355..08e2b91 100644 --- a/tests/test_provenance.py +++ b/tests/test_provenance.py @@ -15,9 +15,13 @@ def test_carries_all_core_fields_from_primitives(): expected_cell_count=64_742, actual_cell_count=64_742, unmapped_count=0, + observed_through=559, ) assert prov == { "lookup_version": "v1.4.0", + # #297: always present, never omitted — an absent boundary is + # indistinguishable from an artifact built before the field existed. + "observed_through": 559, "region": "land_gaul", "expected_cell_count": 64_742, "actual_cell_count": 64_742, @@ -32,6 +36,7 @@ def test_fill_count_omitted_when_not_supplied(): expected_cell_count=1, actual_cell_count=1, unmapped_count=0, + observed_through=559, ) assert "fill_count" not in prov @@ -43,6 +48,7 @@ def test_fill_count_included_when_supplied(): expected_cell_count=1, actual_cell_count=1, unmapped_count=0, + observed_through=559, fill_count=7, ) assert prov["fill_count"] == 7 @@ -55,6 +61,7 @@ def test_unpinned_region_keeps_none_expected_count(): expected_cell_count=None, actual_cell_count=13_110, unmapped_count=0, + observed_through=559, ) assert prov["expected_cell_count"] is None assert prov["region"] == "africa_me_legacy" @@ -67,6 +74,7 @@ def test_result_is_json_serializable(): expected_cell_count=64_742, actual_cell_count=64_700, unmapped_count=0, + observed_through=559, fill_count=3, ) # round-trips cleanly — it must survive serialization into the upload description. diff --git a/tests/test_redaction_guard.py b/tests/test_redaction_guard.py index 62e2b26..7f163ad 100644 --- a/tests/test_redaction_guard.py +++ b/tests/test_redaction_guard.py @@ -141,10 +141,17 @@ def test_provenance_carries_only_the_declared_closed_keyset(): expected_cell_count=64742, actual_cell_count=64742, unmapped_count=0, + observed_through=559, fill_count=3, ) assert set(prov) == { "lookup_version", + # #297: the producer's observed-data frontier this run clipped against. A + # month_id integer — no PII, no credential, no internal path — and it is + # precisely what the partner needs to tell a sparsely-reported month from + # a fabricated one. Widening this keyset is a deliberate act; that is why + # this guard exists. + "observed_through", "region", "expected_cell_count", "actual_cell_count", diff --git a/tests/test_selection_guard.py b/tests/test_selection_guard.py index 1cb87bb..b113ba0 100644 --- a/tests/test_selection_guard.py +++ b/tests/test_selection_guard.py @@ -98,6 +98,7 @@ def test_compact_description_fits_the_store_limit(): expected_cell_count=64742, actual_cell_count=64742, unmapped_count=0, + observed_through=559, ) text = compact_description(prov) assert len(text) <= DESCRIPTION_MAX diff --git a/views_postprocessing/crafd/managers/crafd.py b/views_postprocessing/crafd/managers/crafd.py index 5201056..d5f9982 100644 --- a/views_postprocessing/crafd/managers/crafd.py +++ b/views_postprocessing/crafd/managers/crafd.py @@ -156,6 +156,9 @@ def __init__( logger.info(f"Initializing {self.__class__.__name__}") self._forecast_resolution = None # {target: TargetLease}, set by _read self._historical_frame = None # views_frames.FeatureFrame, set by _read + # int | None, set by _read_historical_frame. UNREAD until then, so that + # "never read" cannot be mistaken for "read and unavailable" (#297). + self._observed_through = provenance.UNREAD def _read_historical_frame(self): """#126: historical actuals as a views_frames.FeatureFrame — the first @@ -179,6 +182,10 @@ def _read_historical_frame(self): exc_info=True, ) lv = None + # Stamped into provenance either way: the boundary this run clipped against, + # or None meaning the clip was skipped. #297 cost a day of forensics because + # the artifact could not answer which. + self._observed_through = lv if lv is None: self._historical_frame = frame return @@ -417,6 +424,7 @@ def _historical_frame_description(self, table, timestamp: str) -> str: expected_cell_count=coverage.expected_for(region), actual_cell_count=len(frame_extraction.cells_of(self._historical_frame)), unmapped_count=historical.unmapped_cell_count(table), + observed_through=self._observed_through, ) return provenance.compact_description(prov) diff --git a/views_postprocessing/delivery/provenance.py b/views_postprocessing/delivery/provenance.py index 35adab8..a7cc049 100644 --- a/views_postprocessing/delivery/provenance.py +++ b/views_postprocessing/delivery/provenance.py @@ -10,11 +10,40 @@ a free-text ``description`` — so the manager serializes this dict into ``description`` as JSON for now. A dedicated field is requested upstream (see C-15); when it lands, only the manager's attach step changes, not this shape. + +Deferred deliberately (#297): the producer also publishes a **per-source** boundary map, +``last_valid_month_ids``, in the store attrs and consumer manifest. It is not stamped here +because ``datafactory_query.defaults`` exposes only the scalar ``get_last_valid_month_id``, +and a multi-source map would not fit the 255-char carrier below in any case. +**Trigger:** when C-15's structured metadata field lands upstream and the 255-char ceiling +goes with it, stamp the per-source map alongside the scalar. """ from __future__ import annotations +class _Unread: + """Sentinel for "the observed-range boundary has not been read yet". + + Distinct from ``None``, which means "read attempted and unavailable, so this + delivery was NOT clipped". Conflating the two is the C-103 mistake — a broken + read reported as a producer that publishes no boundary — and #297 is what it + costs: an artifact that cannot say what it clipped against, and a day of + forensics to recover a number the delivery already had. + """ + + __slots__ = () + + def __repr__(self) -> str: # pragma: no cover - diagnostic only + return "" + + +#: Managers initialise their boundary attribute to this; ``build_provenance`` +#: refuses it. Reaching provenance without having read the boundary is a bug in +#: the call order, not a delivery condition. +UNREAD = _Unread() + + def build_provenance( *, lookup_version: str, @@ -22,6 +51,7 @@ def build_provenance( expected_cell_count: int | None, actual_cell_count: int, unmapped_count: int, + observed_through: int | None | _Unread, fill_count: int | None = None, ) -> dict: """Assemble the structured provenance for one delivered file. @@ -37,13 +67,27 @@ def build_provenance( expected_cell_count: the region's pinned cell count, or None if unpinned. actual_cell_count: distinct cells actually delivered in this file. unmapped_count: delivered cells with missing metadata (0 once validation passes). + observed_through: the producer's ``last_valid_month_id`` this delivery clipped + against, or ``None`` if the boundary could not be read and the clip was + therefore **skipped** (degrade-open, C-26). Always emitted, never omitted: + an absent field is indistinguishable from an older artifact that never + stamped one, and that ambiguity is the whole of #297. Passing + :data:`UNREAD` raises. fill_count: optional count of fabricated/filled values, if known. Returns: A JSON-serializable dict of the provenance fields. """ + if isinstance(observed_through, _Unread): + raise ValueError( + "observed_through was never read — provenance is being built before the " + "observed-range boundary was fetched. This is a call-order bug, not a " + "delivery condition: pass the boundary the clip used, or None if the read " + "failed and the clip was skipped (C-26)." + ) provenance: dict = { "lookup_version": lookup_version, + "observed_through": observed_through, "region": region, "expected_cell_count": expected_cell_count, "actual_cell_count": actual_cell_count, @@ -70,7 +114,14 @@ def compact_description(prov: dict) -> str: if len(text) > DESCRIPTION_MAX: essential = { k: prov[k] - for k in ("lookup_version", "region", "expected_cell_count", "actual_cell_count", "unmapped_count") + for k in ( + "lookup_version", + "observed_through", + "region", + "expected_cell_count", + "actual_cell_count", + "unmapped_count", + ) if k in prov } text = json.dumps(essential, separators=(",", ":")) diff --git a/views_postprocessing/unfao/managers/unfao.py b/views_postprocessing/unfao/managers/unfao.py index 4d6ddf6..8720230 100644 --- a/views_postprocessing/unfao/managers/unfao.py +++ b/views_postprocessing/unfao/managers/unfao.py @@ -156,6 +156,9 @@ def __init__( logger.info(f"Initializing {self.__class__.__name__}") self._forecast_resolution = None # {target: TargetLease}, set by _read self._historical_frame = None # views_frames.FeatureFrame, set by _read + # int | None, set by _read_historical_frame. UNREAD until then, so that + # "never read" cannot be mistaken for "read and unavailable" (#297). + self._observed_through = provenance.UNREAD def _read_historical_frame(self): """#126: historical actuals as a views_frames.FeatureFrame — the first @@ -179,6 +182,10 @@ def _read_historical_frame(self): exc_info=True, ) lv = None + # Stamped into provenance either way: the boundary this run clipped against, + # or None meaning the clip was skipped. #297 cost a day of forensics because + # the artifact could not answer which. + self._observed_through = lv if lv is None: self._historical_frame = frame return @@ -417,6 +424,7 @@ def _historical_frame_description(self, table, timestamp: str) -> str: expected_cell_count=coverage.expected_for(region), actual_cell_count=len(frame_extraction.cells_of(self._historical_frame)), unmapped_count=historical.unmapped_cell_count(table), + observed_through=self._observed_through, ) return provenance.compact_description(prov)