feat(delivery): stamp the observed-range boundary into provenance (#297)

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

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

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.

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",

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

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)
 

views_postprocessing/delivery/provenance.py

@@ -10,18 +10,48 @@
 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 "<boundary-unread>"
+
+
+#: 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,
     region: str | None,
     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=(",", ":"))