diff --git a/docs/ADRs/013_sampled_forecast_wire_contract.md b/docs/ADRs/013_sampled_forecast_wire_contract.md index ce9c3a3..0715da7 100644 --- a/docs/ADRs/013_sampled_forecast_wire_contract.md +++ b/docs/ADRs/013_sampled_forecast_wire_contract.md @@ -1116,6 +1116,34 @@ record execution progress against it. and the provenance keyset closed (PR #136). **P3/#138** — §7(d) added: this ADR now cross-references PLATFORM-001 by URL, never by copy. +- **2026-07-31 — the Hop-A LEGACY READER IS RETIRED; §11.4's transition is over on + this side (#149, epic #148).** §11.4 required a type-aware guard live at each + store **before the first contract artifact was uploaded there**. Both guards + merged 2026-07-15 and run-0 uploaded 2026-07-27, so the sequencing constraint was + satisfied. The legacy reader it protected has now been **deleted**: this repo + reads forecasts only through the contract path (manifest selection, + `unfao/wire/source_selection.py`) and historical actuals only as a + `views_frames.FeatureFrame` (#126). + + **The retired guard, recorded here because the code that carried it is gone.** + The legacy reader selected `{"category": "forecast", "type": "ensemble"}` — + pinning the legacy `type` was what made pipeline-core#269's contract uploads + (`sampled_forecast_shard` / `_manifest` / `_sidecar`) unselectable by it. That + disjointness is the fact §11.4 turned on, and it held for the whole transition. + + **Consequence for §11.4:** its Hop-A clause is now historical. **Its Hop-B clause + still binds** — views-faoapi's deployed selector is a separate deployment on its + own schedule (views-faoapi C-161). Nothing in this entry relaxes the consumer + side. + + **The delivery mode is now DECLARED, not inferred.** Previously + `configs.get("wire_contract")` returning `None` because a launcher never mentioned + the key was indistinguishable from a deliberate `False`, and either quietly + selected the retired path. A launcher that does not declare `wire_contract: True` + and `data_format: feature_frame` is now **refused by name** + (`unfao/launch_config.py`) — ADR-003 applied to the launch config, and register + C-63 closed. + --- ## Appendix A — Version history (review archaeology; non-normative) diff --git a/reports/technical_risk_register.md b/reports/technical_risk_register.md index 8156360..548485c 100644 --- a/reports/technical_risk_register.md +++ b/reports/technical_risk_register.md @@ -5,8 +5,8 @@ | Project | views-postprocessing | | Owner | Dylan Pinheiro / PRIO MD&D Team | | Last Updated | 2026-07-31 | -| Total Concerns | 70 | -| Open Concerns | 33 | +| Total Concerns | 71 | +| Open Concerns | 34 | | Resolved Concerns | 37 | --- @@ -812,6 +812,26 @@ Cross-refs: **C-69** (the packaging split this rides along with), **C-59**/**C-6 --- +### C-71: `appwrite_env.assert_env_declared` raises without logging — ADR-008 non-compliance in an entry-validation seam + +| Field | Value | +|-------|-------| +| ID | C-71 | +| Tier | 4 — the raise is loud and its message is fully diagnostic, so nothing is silently swallowed today. What is missing is the persistent record ADR-008 requires, in the one seam whose whole job is to make a misconfigured launch visible. | +| Source | `review-diff` (2026-07-31) — S1/#149 review; found by mirroring this module when writing its sibling | +| Trigger | When a delivery run refuses on a missing Appwrite variable and the operator goes looking for *why* in the logs rather than the traceback — or when the next entry-validation module is written against this one as the pattern, as `unfao/launch_config.py` was | +| Location | `views_postprocessing/unfao/appwrite_env.py:44-51` (`assert_env_declared`) | + +**ADR-008:48** requires that *"raised structural failures must be logged at `ERROR` level or higher"*, and **:51** that *"raising is not a substitute for logging."* `assert_env_declared` raises `EnvironmentError` naming every missing variable but never logs. A launcher misconfiguration is a structural failure by any reading of that ADR. + +**How it was found, and why that matters:** `unfao/launch_config.py` (S1) was deliberately written to mirror this module — same shape, same failure style, same dependency-light constraint. It **inherited the flaw**, and the S1 diff review caught it in the new code. The new module was fixed to log-before-raise; this one was left alone for scope discipline, which means the pair is now **inconsistent** — the sibling written to match it no longer does. + +Registered rather than fixed in #149 because it is pre-existing (shipped in þing-01 P1 / #134) and outside that story's boundary. It is a two-line change and the natural place to take it is **S8** (#156, epic closeout) or any PR that next touches `appwrite_env.py`. + +Cross-refs: **C-19** (RESOLVED — the ADR-008 log-before-raise sweep whose convention this predates), **C-63** (the declaration-over-inference concern S1 closed), ADR-008, #134, #149, #156. + +--- + ## Disagreements ### D-12: Post-Run-0 infrastructure & naming intents — repo rename, internal-store transport, compute co-location diff --git a/tests/test_append_metadata.py b/tests/test_append_metadata.py deleted file mode 100644 index 5e604ce..0000000 --- a/tests/test_append_metadata.py +++ /dev/null @@ -1,87 +0,0 @@ -"""Tests for the manager's _append_metadata sequence after the ADR-011 swap. - -UNFAOPostProcessorManager cannot be instantiated in this environment (it needs -views-pipeline-core), so — like test_validation.py replicates _validate() — this -replicates the exact _append_metadata sequence with GaulLookupEnricher: - - raw = enricher.enrich_dataframe_with_pg_info(df.reset_index(), ...) - raw = raw[filter_cols].set_index([time_id, entity_id]) - result = predictions.join(raw) - -and verifies the joined result carries the 9 metadata columns aligned to the -predictions, with the fail-loud path intact. Keep in lockstep with -views_postprocessing/unfao/managers/unfao.py:_append_metadata. -""" - -import pandas as pd -import pytest - -from views_postprocessing.unfao.enrichment import GaulLookupEnricher -from views_postprocessing.unfao.gaul_schema import METADATA_COLS - -_TIME_ID = "month_id" -_ENTITY_ID = "priogrid_gid" -# filter_cols as built in _append_metadata: time, entity, then the 9 columns. -_FILTER_COLS = [_TIME_ID, _ENTITY_ID, *METADATA_COLS] - - -@pytest.fixture(scope="module") -def enricher(): - return GaulLookupEnricher() - - -def _predictions(gids, months=(100, 101)): - """A minimal prediction frame indexed like a PGMDataset.dataframe.""" - rows = [(m, g) for m in months for g in gids] - idx = pd.MultiIndex.from_tuples(rows, names=[_TIME_ID, _ENTITY_ID]) - return pd.DataFrame({"pred_ln_sb_best": 0.0}, index=idx) - - -def _append_metadata(enricher, predictions): - """Replica of UNFAOPostProcessorManager._append_metadata.""" - raw = enricher.enrich_dataframe_with_pg_info( - predictions.reset_index(), - pg_id_col=_ENTITY_ID, time_id_col=_TIME_ID, only_metadata=True, - ) - raw = raw[_FILTER_COLS].set_index([_TIME_ID, _ENTITY_ID]) - return predictions.join(raw) - - -class TestAppendMetadata: - def test_nine_columns_added_and_aligned(self, enricher): - gids = enricher._lookup.index[:5].tolist() - out = _append_metadata(enricher, _predictions(gids)) - for c in METADATA_COLS: - assert c in out.columns - # The prediction column survives the join. - assert "pred_ln_sb_best" in out.columns - - def test_row_count_preserved(self, enricher): - gids = enricher._lookup.index[:4].tolist() - preds = _predictions(gids, months=(100, 101, 102)) - out = _append_metadata(enricher, preds) - assert len(out) == len(preds) # 4 gids x 3 months - - def test_metadata_matches_lookup_per_cell(self, enricher): - gids = enricher._lookup.index[:6].tolist() - out = _append_metadata(enricher, _predictions(gids, months=(100,))) - for g in gids: - assert out.loc[(100, g), "country_iso_a3"] == \ - enricher._lookup.loc[g, "country_iso_a3"] - assert out.loc[(100, g), "admin2_gaul2_code"] == \ - enricher._lookup.loc[g, "admin2_gaul2_code"] - - def test_no_nulls_for_known_cells(self, enricher): - gids = enricher._lookup.index[:20].tolist() - out = _append_metadata(enricher, _predictions(gids, months=(100,))) - for c in METADATA_COLS: - assert out[c].isna().sum() == 0, c - - def test_unknown_cell_yields_null_metadata(self, enricher): - # Fail-loud: an out-of-lookup cell flows through as NaN so _validate() - # (not the enricher) is the single place the delivery crashes. - good = int(enricher._lookup.index[0]) - preds = _predictions([good, 999_999], months=(100,)) - out = _append_metadata(enricher, preds) - assert out.loc[(100, good), "country_iso_a3"] is not None - assert pd.isna(out.loc[(100, 999_999), "country_iso_a3"]) diff --git a/tests/test_env_declaration.py b/tests/test_env_declaration.py index 8eee47b..5ad57f7 100644 --- a/tests/test_env_declaration.py +++ b/tests/test_env_declaration.py @@ -90,7 +90,10 @@ def test_declared_names_match_the_manager_reads(): appwrite_env.CONNECTION_ENV + appwrite_env.PROD_FORECASTS_ENV + appwrite_env.UNFAO_ENV ) assert read_names == declared - assert text.count("appwrite_env.assert_env_declared(") == 3 # both stores + legacy save + # One validation per store construction. Was 3 until #149 retired the legacy + # `_save`, whose Appwrite config duplicated `_unfao_appwrite_config` verbatim; + # the two survivors are the production_forecasts read and the unfao_bucket write. + assert text.count("appwrite_env.assert_env_declared(") == 2 def test_secret_env_names_follow_the_platform_naming_rule(): diff --git a/tests/test_falsification_fao_notes_relevance.py b/tests/test_falsification_fao_notes_relevance.py index 2ccd229..b8f1a65 100644 --- a/tests/test_falsification_fao_notes_relevance.py +++ b/tests/test_falsification_fao_notes_relevance.py @@ -20,7 +20,7 @@ def test_falsify_fao_02_schema_country_identifier_m49_vs_iso_a3(): Probe 2 (Category H): FAO contract specifies UN M49; postprocessor uses ISO A3 Finding: Release Note 01 Topic C states "PRIO/VIEWS will use UN M49 - as the standard country identifier." The postprocessor's _append_metadata + as the standard country identifier." The postprocessor's geography attachment enriches with 'country_iso_a3' (ISO Alpha-3). If the postprocessor output feeds into the API that FAO accesses, there is a coding system mismatch. diff --git a/tests/test_input_integrity_e2e.py b/tests/test_input_integrity_e2e.py index a099e16..a2ad4fd 100644 --- a/tests/test_input_integrity_e2e.py +++ b/tests/test_input_integrity_e2e.py @@ -2,16 +2,21 @@ story's primitives unit tests. ``UNFAOPostProcessorManager`` cannot be instantiated here (it needs views-pipeline-core + -Appwrite env — register C-40), so — exactly as ``test_append_metadata.py`` and -``test_validation.py`` do — each test **replicates the manager's own extraction→invariant -chain** on a synthetic delivery frame and proves the guard fires. Keep these in lockstep -with ``views_postprocessing/unfao/managers/unfao.py``: - - S1 coverage _check_coverage cells_of -> assert_complete_coverage - S2 observed range _clip_observed_history months_of -> fabricated_months -> drop_months_above - S3 forecast identity _read_forecast_data file_metadata -> assert_forecast_identity - S4 land_gaul _check_coverage cells_of -> assert_no_excluded_cells - S5 provenance _delivery_description build_provenance(<- cells_of/unmapped_cell_count) +Appwrite env — register C-40), so — exactly as ``test_validation.py`` does — each test +**replicates the delivery's extraction→invariant chain** on a synthetic frame and proves +the guard fires. + +What these prove is the **invariants in** ``views_postprocessing/delivery/``, which is +why they survived #149 untouched: the rules are representation-free, so retiring the +pandas delivery changed which module *calls* them, not whether they hold. The mapping to +manager methods below is therefore informational, and updated for the post-#149 tree: + + S1 coverage _check_coverage cells_of -> assert_complete_coverage + S2 observed range _read_historical_frame months_of -> fabricated_months -> drop_months_above + S3 forecast identity (RETIRED with the legacy reader, #149 — selection is by run + manifest; `delivery/identity.py` itself is retired in #150) + S4 land_gaul _check_coverage cells_of -> assert_no_excluded_cells + S5 provenance _historical_frame_description build_provenance(<- cells_of/unmapped_cell_count) """ import json diff --git a/tests/test_launch_config.py b/tests/test_launch_config.py new file mode 100644 index 0000000..e19fa6b --- /dev/null +++ b/tests/test_launch_config.py @@ -0,0 +1,124 @@ +"""The launch declarations are asserted, not inferred (S1 / #149; register C-63). + +Until #149 this package inferred its delivery mode from the *absence* of config +keys: `configs.get("wire_contract")` returning `None` because a launcher never +mentioned it was indistinguishable from a deliberate `False`, and either quietly +selected a pandas path kept "until run 0 proves the contract path live" and never +removed. Two independent axes gave four delivery modes; production used one; the +other three were reachable only by omission. + +These tests pin the replacement: **one path, and a refusal that names what the +launcher left out.** They are the mechanical form of ADR-003 (authority of +declarations over inference) applied to the launch config, mirroring +`test_env_declaration.py`, which does the same job for the environment. + +`launch_config` is dependency-light by design — no pipeline-core, no pandas, no +frames — so these run anywhere. +""" + +import pytest + +from views_postprocessing.unfao import launch_config +from views_postprocessing.unfao.launch_config import LaunchConfigError + + +# ── the forecast axis: wire_contract ──────────────────────────────────────── + + +def test_missing_wire_contract_raises_naming_the_key(): + """The key's absence must be an error, not a silent route to a retired path.""" + with pytest.raises(LaunchConfigError, match="wire_contract"): + launch_config.assert_contract_mode({}) + + +def test_wire_contract_false_raises_too(): + """An explicit False is not a request for the old path — it no longer exists.""" + with pytest.raises(LaunchConfigError, match="wire_contract"): + launch_config.assert_contract_mode({"wire_contract": False}) + + +def test_the_refusal_says_there_is_no_fallback(): + """A launcher author must not read the error as 'set it False instead'.""" + with pytest.raises(LaunchConfigError) as excinfo: + launch_config.assert_contract_mode({}) + message = str(excinfo.value) + assert "no fallback" in message + assert "#149" in message # where the retired path went, for the reader + + +def test_the_refusal_is_logged_before_it_is_raised(caplog): + """ADR-008: structural failures are logged persistently AND raised. + + Not belt-and-braces — a refused launch must leave a record even if the raise is + swallowed further up the pipeline framework, because the message is the only + diagnostic a launcher author gets. + """ + with caplog.at_level("ERROR"): + with pytest.raises(LaunchConfigError): + launch_config.assert_contract_mode({}) + assert any(r.levelname == "ERROR" for r in caplog.records), ( + "the refusal was raised without being logged (ADR-008)" + ) + assert "wire_contract" in caplog.text + + +def test_the_format_refusal_is_logged_too(caplog): + with caplog.at_level("ERROR"): + with pytest.raises(LaunchConfigError): + launch_config.assert_frame_native_historical("dataframe") + assert any(r.levelname == "ERROR" for r in caplog.records) + + +def test_declared_contract_mode_passes(): + launch_config.assert_contract_mode({"wire_contract": True}) + + +def test_other_keys_are_not_consulted(): + """Only the declared key decides. No inference from neighbouring config.""" + with pytest.raises(LaunchConfigError): + launch_config.assert_contract_mode( + {"wire_upload_enabled": True, "region": "land_gaul", "ensemble": "rusty_bucket"} + ) + + +# ── the historical axis: data_format ──────────────────────────────────────── + + +def test_missing_data_format_raises_naming_the_expected_value(): + with pytest.raises(LaunchConfigError, match="feature_frame"): + launch_config.assert_frame_native_historical(None) + + +def test_a_different_data_format_raises_and_reports_what_it_saw(): + """The error must name the offending value — a queryset author needs to find it.""" + with pytest.raises(LaunchConfigError) as excinfo: + launch_config.assert_frame_native_historical("dataframe") + assert "'dataframe'" in str(excinfo.value) + + +def test_declared_feature_frame_passes(): + launch_config.assert_frame_native_historical("feature_frame") + + +# ── the manager actually calls them ───────────────────────────────────────── + + +def test_the_manager_asserts_both_axes(): + """Source-scan: the manager pulls the full pipeline-core framework and cannot be + imported in every test environment (the repo's standing pattern — see + `test_env_declaration.py`). What matters is that neither axis is left to + `.get()` with a silent default.""" + from pathlib import Path + + source = ( + Path(__file__).resolve().parent.parent + / "views_postprocessing" / "unfao" / "managers" / "unfao.py" + ).read_text() + + assert "launch_config.assert_contract_mode(" in source + assert "launch_config.assert_frame_native_historical(" in source + # And the inference it replaced must not survive anywhere. + assert 'configs.get("wire_contract")' not in source, ( + "the manager still branches on wire_contract — S1 replaced that fork with a " + "single refusal (C-63)." + ) diff --git a/tests/test_selection_guard.py b/tests/test_selection_guard.py index bf3f0ed..1cb87bb 100644 --- a/tests/test_selection_guard.py +++ b/tests/test_selection_guard.py @@ -1,44 +1,88 @@ -"""Golden-string tests for the Hop-A legacy transition guard (ADR-013 §11.4). +"""The Hop-A legacy transition guard (ADR-013 §11.4) — retired, and pinned as retired. -The manager cannot be imported in every test environment (it pulls the full -pipeline-core framework), so — like the design-contract tests — these pin the guard -at source level: the legacy selection filters must pin BOTH `category` and the legacy -`type`, so pipeline-core#269's contract uploads (`sampled_forecast_*`) can never be -selected by the deployed legacy reader. +**What these tests used to do.** Until #149 they pinned a golden string in the +manager: `LEGACY_FORECAST_FILTERS = {"category": "forecast", "type": "ensemble"}`. +Pinning the legacy `type` was what made pipeline-core#269's contract uploads +(`sampled_forecast_*`) unselectable by the deployed legacy reader — the +disjointness §11.4's sequencing constraint turned on. + +**Why they changed.** §11.4 required that guard live *before the first contract +artifact was uploaded*; both guards merged 2026-07-15 and run-0 uploaded +2026-07-27, so the constraint was satisfied and the transition closed. #149 then +deleted the reader the guard protected. A golden string pinning deleted code is +not a guarantee — it is a test that can only ever fail. + +**So the guarantee moved rather than lapsed.** The historical fact now lives where +facts about the contract belong: ADR-013's post-adoption record. These tests pin +it *there*, and add the invariant that replaces it — the legacy selection must not +come back. + +Source-level, like the design-contract tests: the manager pulls the full +pipeline-core framework and cannot be imported in every test environment. """ import re from pathlib import Path -_SRC = ( - Path(__file__).resolve().parent.parent - / "views_postprocessing" / "unfao" / "managers" / "unfao.py" -).read_text() +_ROOT = Path(__file__).resolve().parent.parent +_SRC = (_ROOT / "views_postprocessing" / "unfao" / "managers" / "unfao.py").read_text() +_ADR = (_ROOT / "docs" / "ADRs" / "013_sampled_forecast_wire_contract.md").read_text() -def test_legacy_forecast_filters_pin_category_and_type(): - # The golden string (ADR-013 §3.3 spirit): identity of the guard is this exact dict. - assert 'LEGACY_FORECAST_FILTERS = {"category": "forecast", "type": "ensemble"}' in _SRC +def test_the_legacy_reader_is_gone(): + """The retired reader must not creep back — this is the live invariant now. + Note what is NOT asserted: `_ContractStorePort.latest_file_id` forwards a + `filters` argument, and that is correct — it is the DIP adapter through which + `wire/source_selection` runs its *manifest* queries. The defect was never + "filters exist"; it was a hardcoded category-scoped filter selecting the newest + upload in the whole bucket. That is what the next test pins. + """ + assert "LEGACY_FORECAST_FILTERS" not in _SRC, ( + "LEGACY_FORECAST_FILTERS is back in the manager. The Hop-A legacy reader was " + "retired in #149; forecasts are selected by run manifest, never by a category " + "filter over the whole bucket." + ) -def test_selection_uses_the_guarded_filters(): - # The store selection must go through the guarded constant — no inline - # category-only filter dict may remain on the selection call. - assert "get_latest_file_id(filters=LEGACY_FORECAST_FILTERS)" in _SRC - assert not re.search( - r"get_latest_file_id\(filters=\{[^}]*\}\)", _SRC - ), "inline filter dict on the selection call — must use LEGACY_FORECAST_FILTERS" + +def test_no_category_only_selection_anywhere_in_the_manager(): + """C-25's original hazard: 'newest category=forecast wins regardless of producer'. + + Manifest selection makes it structurally impossible, but an inline filter dict + on any store call would reintroduce it. + """ + assert not re.search(r'get_latest_file_id\(\s*filters\s*=\s*\{', _SRC), ( + "inline selection-filter dict in the manager — selection belongs to " + "wire/source_selection, keyed on the run manifest." + ) -def test_guard_type_is_disjoint_from_contract_vocabulary(): - # The pinned legacy type must never be one of the contract's types. - assert '"type": "ensemble"' in _SRC - for contract_type in ( - "sampled_forecast_shard", - "sampled_forecast_manifest", - "sampled_forecast_sidecar", - ): - assert f'"type": "{contract_type}"' not in _SRC +def test_the_retired_guard_is_recorded_in_the_adr(): + """The disjointness fact outlives the code that carried it. + + A future reader asking "how was the transition made safe?" must find the + answer, and after #149 the only place it can live is the contract's own record. + """ + assert "Hop-A LEGACY READER IS RETIRED" in _ADR, ( + "ADR-013's post-adoption record does not record the Hop-A retirement — the " + "§11.4 transition fact would have no home." + ) + # The retired filter values, so the historical claim stays checkable. + assert '{"category": "forecast", "type": "ensemble"}' in _ADR + # And what made them safe: disjointness from the contract vocabulary. + assert "sampled_forecast_shard" in _ADR + + +def test_hop_b_half_of_the_transition_rule_still_binds(): + """Retiring OUR reader must not be read as relaxing the consumer's obligation. + + §11.4 governs BOTH hops. views-faoapi's deployed selector is a separate + deployment on its own schedule; nothing about #149 touches it. + """ + assert "Its Hop-B clause" in _ADR and "still binds" in _ADR, ( + "the ADR must state that retiring the Hop-A reader leaves the Hop-B " + "constraint intact — otherwise the entry reads as a blanket relaxation." + ) def test_compact_description_fits_the_store_limit(): diff --git a/views_postprocessing/unfao/launch_config.py b/views_postprocessing/unfao/launch_config.py new file mode 100644 index 0000000..2b88f56 --- /dev/null +++ b/views_postprocessing/unfao/launch_config.py @@ -0,0 +1,85 @@ +"""The launch declarations this delivery requires, DECLARED (S1, #149; register C-63). + +Sibling of ``appwrite_env`` — that module asserts the launcher assembled the +*environment*; this one asserts it declared the *delivery mode*. Same rule +(ADR-003: authority of declarations over inference), same shape, same failure +style: name what is missing and refuse. + +**Why this exists.** Until S1 this package inferred its delivery mode from the +*absence* of config keys. ``configs.get("wire_contract")`` returning ``None`` +because the launcher never mentioned it was indistinguishable from a deliberate +``False``, and either quietly selected a retired pandas path that had been kept +"until run 0 proves the contract path live" and never removed. Two independent +axes — ``wire_contract`` and the queryset's ``data_format`` — gave four delivery +modes of which production used one; the other three were reachable only by +omission. That is precisely the inference ADR-003 forbids, in the repository that +authored it. + +There is now **one** delivery path. A launcher that has not declared it does not +get a quiet fallback; it gets an error naming the key it left out. + +Deliberately dependency-light: no pipeline-core, no pandas, no frames — the +queryset's ``data_format`` arrives as a plain string from the caller, so this +module stays importable (and testable) anywhere. +""" + +from __future__ import annotations + +import logging + +logger = logging.getLogger(__name__) + +# The launch-config key that selects the ADR-013 contract delivery. Declared, never +# inferred: `True` is the only accepted value, because the alternative it used to +# select no longer exists. +WIRE_CONTRACT_KEY = "wire_contract" + +# The queryset descriptor value that selects the frame-native historical read +# (#126). Read via pipeline-core's `declared_data_format()` — the one gate — and +# passed here as a string so this module needs no pipeline-core import. +FEATURE_FRAME_FORMAT = "feature_frame" + + +class LaunchConfigError(ValueError): + """The launcher did not declare the delivery this package implements.""" + + +def assert_contract_mode(configs: dict) -> None: + """Raise unless the launch config declares ADR-013 contract delivery. + + Args: + configs: the launcher's config mapping. + + Raises: + LaunchConfigError: naming the missing or falsy key. + """ + if not configs.get(WIRE_CONTRACT_KEY): + err_msg = ( + f"the launcher did not declare {WIRE_CONTRACT_KEY!r} — this package " + f"delivers only the ADR-013 contract wire. Set " + f"{WIRE_CONTRACT_KEY}: True in the postprocessor's config_meta. " + "The pandas delivery this key used to select was retired in #149; " + "there is no fallback to omit your way into." + ) + logger.error(err_msg) # ADR-008: logged persistently AND raised + raise LaunchConfigError(err_msg) + + +def assert_frame_native_historical(data_format: str | None) -> None: + """Raise unless the queryset descriptor declares the frame-native historical read. + + Args: + data_format: the value of ``declared_data_format(queryset)``. + + Raises: + LaunchConfigError: naming the expected descriptor value. + """ + if data_format != FEATURE_FRAME_FORMAT: + err_msg = ( + f"the queryset declares data_format={data_format!r}; this package " + f"reads historical actuals only as {FEATURE_FRAME_FORMAT!r} (#126). " + f"Set data_format: {FEATURE_FRAME_FORMAT!r} in the postprocessor's " + "config_queryset. The pandas historical read was retired in #149." + ) + logger.error(err_msg) # ADR-008: logged persistently AND raised + raise LaunchConfigError(err_msg) diff --git a/views_postprocessing/unfao/managers/unfao.py b/views_postprocessing/unfao/managers/unfao.py index ef3f22c..92499f1 100644 --- a/views_postprocessing/unfao/managers/unfao.py +++ b/views_postprocessing/unfao/managers/unfao.py @@ -2,37 +2,33 @@ PostprocessorManager, PostprocessorPathManager, ) -from views_pipeline_core.files.utils import read_dataframe import logging -from views_pipeline_core.data.handlers import PGMDataset from views_pipeline_core.modules.appwrite.file import AppwriteConfig from views_pipeline_core.modules.datastore import DatastoreModule from views_pipeline_core.managers.model import ForecastingModelManager from views_pipeline_core.managers.ensemble import EnsemblePathManager -import pandas as pd -import io from datetime import datetime import os from views_postprocessing.unfao.enrichment import _DEFAULT_LOOKUP, GaulLookupEnricher -from views_postprocessing.unfao.gaul_schema import METADATA_COLS from views_pipeline_core.modules.dataloaders.datafactory_contract import declared_data_format -from views_postprocessing.unfao import appwrite_env, extraction, frame_extraction, historical, product, source_metadata +from views_postprocessing.unfao import ( + appwrite_env, + extraction, + frame_extraction, + historical, + launch_config, + product, + source_metadata, +) from views_postprocessing.unfao.wire import sink as wire_sink from views_postprocessing.unfao.wire import source_selection -from views_postprocessing.delivery import coverage, identity, observed_range, provenance +from views_postprocessing.delivery import coverage, observed_range, provenance from pathlib import Path logger = logging.getLogger(__name__) -# Hop-A legacy selection filters (ADR-013 §11.4 transition guard). Declared, not -# inferred: the legacy production forecast is uploaded by pipeline-core's ensemble run -# with type="ensemble"; the contract's sampled_forecast_* types are disjoint by design, -# so pinning the legacy type here makes contract artifacts unselectable by this reader. -# Golden-string-tested (tests/test_selection_guard.py); change only with ADR-013. -LEGACY_FORECAST_FILTERS = {"category": "forecast", "type": "ensemble"} - class _ContractStorePort: """Adapts ``DatastoreModule`` to the wire ports (ADR-013 epic #105; DIP — @@ -88,14 +84,9 @@ def __init__( # Add your custom initialization below logger.info(f"Initializing {self.__class__.__name__}") - self._historical_dataframe = None - self._forecast_dataframe = None - - self._historical_dataset = None - self._forecast_dataset = None - self._forecast_resolution = None # contract mode: {target: TargetLease} - self._historical_frame = None # frame-native historical (#126) - self._enricher = GaulLookupEnricher() + self._forecast_resolution = None # {target: TargetLease}, set by _read + self._historical_frame = None # views_frames.FeatureFrame, set by _read + self._enricher = GaulLookupEnricher() # provenance lookup_version (C-15) self.ensemble_path_manager = None def _read_historical_frame(self): @@ -123,42 +114,34 @@ def _read_historical_frame(self): self._historical_frame = frame def _read_historical_data(self): - self._initialize_data_loader() - run_type = "forecasting" - # Declared dispatch (never inferred): the queryset descriptor's - # data_format decides — pipeline-core's contract fn is the one gate. - if declared_data_format(self._model_path.get_queryset()) == "feature_frame": - self._read_historical_frame() - return + """Historical actuals, frame-native only (#126). - self._data_loader.get_data( - use_saved=False, - validate=False, - self_test=False, - partition=run_type - ) - self._historical_dataframe = read_dataframe( - self._data_loader.cached_data_path - ) - self._clip_observed_history() - self._historical_dataset = PGMDataset( - source=self._historical_dataframe, targets=self.configs.get("targets") + The queryset must DECLARE ``data_format: feature_frame`` — pipeline-core's + ``declared_data_format`` is the one gate, and a queryset that declares + anything else is refused rather than quietly read through a retired pandas + path (register C-63, #149). + """ + # Declaration first: the check reads the queryset, not the loader, so a + # refused config must not pay for loader construction. + launch_config.assert_frame_native_historical( + declared_data_format(self._model_path.get_queryset()) ) + self._initialize_data_loader() + self._read_historical_frame() - def _prod_forecasts_datastore(self, *, name_scoped: bool = True) -> DatastoreModule: + def _prod_forecasts_datastore(self) -> DatastoreModule: """The shared internal store (ADR-013's 'shared shelf'), configured from the launcher-assembled environment (validated fail-loud; þing-01 #134 — no dotenv - is loaded here). Used by both the legacy and contract reads. - - ``name_scoped`` controls pipeline-core's automatic ``name == model_name`` query - filter (``DatastoreModule.get_predictions_by_metadata`` injects it on every - lookup). The LEGACY read depends on it: its filters ``{category: forecast, - type: ensemble}`` carry no name, so the injected ensemble-name is the S3/C-25 - identity guard that prevents selecting a *different* ensemble's forecast. The - CONTRACT read must switch it OFF: ADR-013 files are named - ``{run_id}__{target}__manifest.json`` (never the bare ensemble name), so an - injected ``name == "rusty_bucket"`` matches nothing and also clobbers the wire - layer's own run-id / target / name filters.""" + is loaded here). + + pipeline-core's ``DatastoreModule.get_predictions_by_metadata`` injects an + automatic ``name == model_name`` filter on every lookup. The contract read + must **not** have it: ADR-013 artifacts are named + ``{run_id}__{target}__m{month}.arrow.parquet`` (never the bare ensemble + name), so an injected ``name == "rusty_bucket"`` matches nothing and also + clobbers the wire layer's own run-id / target / name filters. Suppressed + unconditionally below — the retired legacy reader was the only caller that + needed it on (#149).""" ensemble_name = self.configs.get("ensemble", None) if not ensemble_name: err_msg = "Ensemble name must be provided in configs with the `ensemble` key for forecasting. Cannot proceed." @@ -195,11 +178,10 @@ def _prod_forecasts_datastore(self, *, name_scoped: bool = True) -> DatastoreMod database_name=os.getenv("APPWRITE_METADATA_DATABASE_NAME"), ) datastore = DatastoreModule(appwrite_file_manager_config=appwrite_config) - if not name_scoped: - # Suppress the automatic name==model_name filter (see docstring). model_path - # is used by DatastoreModule only for that injection and for uploads; the - # contract read neither uploads nor performs any model-scoped lookup. - datastore.model_path = None + # Suppress the automatic name==model_name filter (see docstring). model_path + # is used by DatastoreModule only for that injection and for uploads; the + # contract read neither uploads nor performs any model-scoped lookup. + datastore.model_path = None return datastore def _read_forecast_data_contract(self): @@ -210,7 +192,7 @@ def _read_forecast_data_contract(self): released). The `wire/` package owns the policy; this method only adapts the store (DIP) and declares the product facts (region curation + coverage expectations live in the lease, where frames exist).""" - port = _ContractStorePort(self._prod_forecasts_datastore(name_scoped=False)) + port = _ContractStorePort(self._prod_forecasts_datastore()) region = self.configs.get("region") self._forecast_resolution = source_selection.resolve_run( port, @@ -229,194 +211,68 @@ def _read_forecast_data_contract(self): ) def _read_forecast_data(self): - # Explicit declared mode (ADR-013; never inferred from store contents). - if self.configs.get("wire_contract"): - self._read_forecast_data_contract() - return - loa = "pgm" - try: - prediction_store_manager = self._prod_forecasts_datastore() - - # Resolve the newest legacy forecast, then verify its identity before - # delivering it (S3/C-25): a stray upload must not be silently shipped as - # the configured ensemble. The `type` pin is the Hop-A legacy transition - # guard (ADR-013 §11.4): legacy ensemble forecasts carry type="ensemble" - # (pipeline-core EnsemblePathManager._target), disjoint from the contract's - # sampled_forecast_* vocabulary — so pipeline-core#269's shard/manifest - # uploads can never be selected by this legacy reader. - file_id = prediction_store_manager.get_latest_file_id(filters=LEGACY_FORECAST_FILTERS) - if file_id is None: - raise FileNotFoundError( - f"No forecast file found in the prediction store (filters={LEGACY_FORECAST_FILTERS})." - ) - selected = extraction.file_metadata(prediction_store_manager.get_file_metadata(file_id)) - # Identity contract: the producer's FileMetadata `name`/`loa` written on upload - # must equal the configured ensemble's model_name/loa. Verified against - # pipeline-core's code but NOT a live upload (register C-25) — confirm at the - # first real run. The guard fails loud (not silent) if the contract is off, so - # a mismatch surfaces immediately rather than shipping the wrong file. - expected = {"name": self.ensemble_path_manager.model_name, "loa": loa} - logger.info("Selected forecast file identity: %s (expected %s).", selected, expected) - identity.assert_forecast_identity(selected, expected) - - self._forecast_dataframe = pd.read_parquet(io.BytesIO(prediction_store_manager.download_prediction(file_id).to_dict().get("data", {}).get("file_bytes", None))) - - self._forecast_dataset = PGMDataset(self._forecast_dataframe) - except Exception as e: - logger.error( - f"Encountered an error while trying to download the latest forecast data for level {loa} from Datastore: {e}", - exc_info=True, - ) - raise + """Forecast inbound — ADR-013 contract only. + + The launcher must DECLARE ``wire_contract: True``. The pandas reader this + key used to select was retired in #149; omitting the key is a refusal, not + a fallback (register C-63). + """ + launch_config.assert_contract_mode(self.configs) + self._read_forecast_data_contract() def _read(self) -> any: self._read_historical_data() self._read_forecast_data() - def _append_metadata(self, dataset: PGMDataset) -> pd.DataFrame: - filter_cols = [dataset._time_id, dataset._entity_id, *METADATA_COLS] - raw_result = self._enricher.enrich_dataframe_with_pg_info( - dataset.dataframe.reset_index(), - pg_id_col=dataset._entity_id, - time_id_col=dataset._time_id, - only_metadata=True, - ) - - raw_result = raw_result[filter_cols].set_index([dataset._time_id, dataset._entity_id]) - return dataset.dataframe.join(raw_result) - - def _transform( - self, - ) -> list: - if self._historical_frame is None: - self._historical_dataframe = self._append_metadata(self._historical_dataset) - # frame-native historical attaches geography at artifact build (historical.py) - if self.configs.get("wire_contract"): - # Contract mode: the forecast is frames, not a dataframe; geography ships - # as the §5 sidecar (built at _save), never joined into the payload. - return - self._forecast_dataframe = self._append_metadata(self._forecast_dataset) - - def _validate(self) -> pd.DataFrame: - _necessary_metadata_cols = METADATA_COLS - - if self._historical_frame is not None: - logger.info( - "Historical is frame-native: the metadata null-gate is enforced at " - "artifact build (historical.assert_metadata_complete)." - ) - _historical_cols_to_check = [] - else: - _historical_cols_to_check = _necessary_metadata_cols - for col in _historical_cols_to_check: - if col not in self._historical_dataframe.columns: - err_msg = f"Historical dataframe is missing required metadata column: {col}. Found columns: {self._historical_dataframe.columns.tolist()}" - logger.error(err_msg) - raise ValueError(err_msg) - null_count = self._historical_dataframe[col].isnull().sum() - if null_count > 0: - err_msg = f"Historical dataframe has {null_count} null values in required metadata column: {col} ({null_count}/{len(self._historical_dataframe)} rows)." - logger.error(err_msg) - raise ValueError(err_msg) - logger.info("Historical dataframe metadata validation passed.") - - if self.configs.get("wire_contract"): - # Contract mode: forecast-side null-gating is replaced by the wire's own - # verified chain (hashes + header/payload asserts + coverage inside each - # lease's load; §6 gate + gid parity at _save). This phase asserts the - # RESOLUTION happened — the Template Method stays truthful: _read - # resolves, _save materializes. - if self._forecast_resolution is None: - raise ValueError( - "wire_contract mode: no resolved run — _read did not resolve." - ) - self._check_coverage() - return - - for col in _necessary_metadata_cols: - if col not in self._forecast_dataframe.columns: - err_msg = f"Forecast dataframe is missing required metadata column: {col}. Found columns: {self._forecast_dataframe.columns.tolist()}" - logger.error(err_msg) - raise ValueError(err_msg) - null_count = self._forecast_dataframe[col].isnull().sum() - if null_count > 0: - err_msg = f"Forecast dataframe has {null_count} null values in required metadata column: {col} ({null_count}/{len(self._forecast_dataframe)} rows)." - logger.error(err_msg) - raise ValueError(err_msg) - logger.info("Forecast dataframe metadata validation passed.") - - self._check_coverage() - - def _clip_observed_history(self) -> None: - """Drop fabricated (unobserved) months from the historical delivery (S2/C-26). - - The historical request runs to the current calendar month, but UCDP data ends - earlier (reporting lag) at datafactory's ``last_valid_month_id``; the tail months - are zero-padding, not observed zeros, and would ship to FAO as "zero conflict". + def _transform(self) -> None: + """No-op by design. - The boundary is read straight from the **producer** (views-datafactory) via - ``source_metadata`` — never pipeline-core. Only the *historical* (observed) frame - is clipped; the forecast frame is future-dated by design and untouched. - - Degrade-open policy: if the boundary cannot be resolved — the store predates the - attribute (``None``) or the producer read fails (network) — the clip is skipped - with a WARNING rather than blocking delivery. Both unresolved cases are treated - identically so a transient datafactory hiccup does not crash the historical read. + Geography is not joined into either payload: the historical artifact + attaches it at build time (``unfao/historical.py``) and the forecast ships + it as the §5 GAUL sidecar, built in the sink at ``_save``. The hook stays + so the Template Method's phases remain truthful. """ - try: - lv = source_metadata.last_valid_month_id(self.configs.get("zarr_url")) - except Exception as e: # producer unreachable — degrade open, like lv is None - logger.warning( - "last_valid_month_id could not be read from datafactory (%s); the " - "historical delivery was NOT clipped to observed range (C-26 guard " - "skipped).", - e, - ) - return - if lv is None: - logger.warning( - "last_valid_month_id unavailable from datafactory; the historical " - "delivery was NOT clipped to observed range (C-26 guard skipped)." - ) - return - fabricated = observed_range.fabricated_months( - extraction.months_of(self._historical_dataframe), lv + return + + def _validate(self) -> None: + """Assert the read produced what the save needs, then check coverage. + + Neither payload is null-gated here. The historical artifact's metadata + null-gate fires at build time (``historical.assert_metadata_complete``); + the forecast's guarantees are the wire's own verified chain — content + hashes, header/payload asserts, and per-target coverage inside each + lease's ``load()``, plus the §6 no-collapse gate and gid parity at + ``_save``. This phase asserts the RESOLUTION happened, keeping the + Template Method's phases truthful: ``_read`` resolves, ``_save`` + materializes. + """ + if self._historical_frame is None: + raise ValueError("no historical frame — _read did not run.") + if self._forecast_resolution is None: + raise ValueError("no resolved forecast run — _read did not resolve.") + logger.info( + "Historical is frame-native: the metadata null-gate is enforced at " + "artifact build (historical.assert_metadata_complete)." ) - if len(fabricated): - logger.warning( - "Clipping %d fabricated (unobserved) month(s) > last_valid_month_id=%d " - "from the historical delivery: %s", - len(fabricated), - lv, - fabricated.tolist(), - ) - self._historical_dataframe = extraction.drop_months_above( - self._historical_dataframe, lv - ) + self._check_coverage() def _check_coverage(self) -> None: """Log delivered cell counts and enforce the region coverage contract (S1/C-34). - Orchestration only: extract primitives via ``extraction`` (the pandas seam), - then call the representation-free ``delivery.coverage`` invariant — the rule is + Orchestration only: extract primitives via ``frame_extraction``, then call + the representation-free ``delivery.coverage`` invariant — the rule is *called*, not embedded. The count-gate fires only for regions pinned in ``coverage.EXPECTED_CELLS``; an unpinned/unresolved region logs a skipped gate rather than guessing. + + Only the historical leg is checked here. Forecast coverage is asserted + inside each lease's ``load()``, where the frame actually exists — see + ``wire/source_selection``. """ region = self.configs.get("region") expected = coverage.expected_for(region) excluded = coverage.excluded_for(region) - if self.configs.get("wire_contract"): - # Streaming mode: forecast coverage is asserted inside each lease's - # load() (where the frame exists — see wire/source_selection); only - # the historical (still pandas until #126) is checked here. - sources = [self._historical_coverage_source()] - else: - sources = [ - self._historical_coverage_source(), - ("forecast", extraction.cells_of(self._forecast_dataframe), len(self._forecast_dataframe)), - ] - for label, cells, n_rows in sources: + for label, cells, n_rows in [self._historical_coverage_source()]: logger.info( "%s delivery coverage: %d distinct cells, %d rows.", label, @@ -515,79 +371,16 @@ def _unfao_appwrite_config(self) -> AppwriteConfig: database_name=os.getenv("APPWRITE_METADATA_DATABASE_NAME"), ) - def _save(self) -> list: - # Explicit declared mode (ADR-013; never inferred). - if self.configs.get("wire_contract"): - return self._save_contract() - - if self._historical_dataset is None or self._forecast_dataset is None: - err_msg = "Datasets could not be initialized properly." - logger.error(err_msg) - raise ValueError(err_msg) - - appwrite_env.assert_env_declared( - appwrite_env.CONNECTION_ENV + appwrite_env.UNFAO_ENV, store="unfao_bucket datastore" - ) - unfao_appwrite_config = AppwriteConfig( - path_manager=self._model_path, - endpoint=os.getenv("APPWRITE_ENDPOINT"), - project_id=os.getenv("APPWRITE_DATASTORE_PROJECT_ID"), - credentials=os.getenv("APPWRITE_DATASTORE_API_KEY"), - auth_method="api_key", - cache_ttl_hours=24, - bucket_id=os.getenv("APPWRITE_UNFAO_BUCKET_ID"), - bucket_name=os.getenv("APPWRITE_UNFAO_BUCKET_NAME"), - collection_id=os.getenv("APPWRITE_UNFAO_COLLECTION_ID"), - collection_name=os.getenv("APPWRITE_UNFAO_COLLECTION_NAME"), - database_id=os.getenv("APPWRITE_METADATA_DATABASE_ID"), - database_name=os.getenv("APPWRITE_METADATA_DATABASE_NAME"), - ) - dsm = DatastoreModule(appwrite_file_manager_config=unfao_appwrite_config) - - timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") - forecast_file_path = self._model_path.data_generated / f"forecast_dataset_{timestamp}.parquet" - - if self._historical_frame is not None: - historical_file_path, hist_description, timestamp = self._build_historical_artifact( - self._model_path.data_generated - ) - else: - historical_file_path = self._model_path.data_generated / f"historical_dataset_{timestamp}.parquet" - self._historical_dataframe.to_parquet( - historical_file_path - ) - hist_description = self._delivery_description(self._historical_dataframe, timestamp) - dsm.upload_data(file=historical_file_path, - filename=Path(historical_file_path).name, - name=self._model_path.model_name, - loa="pgm", - type="model", targets=self.configs.get("targets", []), - description=hist_description, - category="historical") - - self._forecast_dataframe.to_parquet( - forecast_file_path - ) - dsm.upload_data(file=forecast_file_path, - filename=Path(forecast_file_path).name, - name=self.ensemble_path_manager.model_name, - loa="pgm", - type="model", targets=["pred_ln_sb_best", "pred_ln_ns_best", "pred_ln_os_best", "pred_ln_sb_prob", "pred_ln_ns_prob", "pred_ln_os_prob"], - description=self._delivery_description(self._forecast_dataframe, timestamp), - category="forecast") + def _save(self) -> dict: + """Deliver the run — ADR-013 contract only (#149).""" + return self._save_contract() def _historical_coverage_source(self): - """(label, cells, n_rows) for whichever historical representation is live.""" - if self._historical_frame is not None: - return ( - "historical", - frame_extraction.cells_of(self._historical_frame), - self._historical_frame.n_rows, - ) + """(label, cells, n_rows) for the historical frame.""" return ( "historical", - extraction.cells_of(self._historical_dataframe), - len(self._historical_dataframe), + frame_extraction.cells_of(self._historical_frame), + self._historical_frame.n_rows, ) def _historical_frame_description(self, table, timestamp: str) -> str: @@ -615,22 +408,3 @@ def _build_historical_artifact(self, directory) -> tuple: path = Path(directory) / f"historical_dataset_{timestamp}.parquet" historical.write_historical_artifact(table, path) return path, self._historical_frame_description(table, timestamp), timestamp - - def _delivery_description(self, df: pd.DataFrame, timestamp: str) -> str: - """Human prefix + structured provenance (S5/C-15) for an upload's metadata. - - Sources every provenance field from the actual delivery — the enricher's lookup - version, the configured region, the S1 coverage counts, the post-enrich unmapped - count — and serializes the representation-free ``delivery.provenance`` dict into - the only structured carrier ``upload_data`` exposes today (the ``description`` - free-text field; a dedicated metadata field is requested upstream, C-15). - """ - region = self.configs.get("region") - prov = provenance.build_provenance( - lookup_version=self._enricher.lookup_version, - region=region, - expected_cell_count=coverage.expected_for(region), - actual_cell_count=len(extraction.cells_of(df)), - unmapped_count=extraction.unmapped_cell_count(df, METADATA_COLS), - ) - return provenance.compact_description(prov)