diff --git a/docs/ADRs/013_sampled_forecast_wire_contract.md b/docs/ADRs/013_sampled_forecast_wire_contract.md index bd14e20..f6d9574 100644 --- a/docs/ADRs/013_sampled_forecast_wire_contract.md +++ b/docs/ADRs/013_sampled_forecast_wire_contract.md @@ -282,3 +282,10 @@ Skeleton ordering therefore is: **consumer guards → producer legs → run 0.** (`66328be`, 12 tests): Track A archives + manifest-last + torn-run abort + §3.4 emission assert + §3.3 golden-string names + §7a wire mapping (fail-loud on unmapped) + §10.2 injectable provenance. The wire now exists end-to-end in code from PFE to the store. +- **2026-07-15 — §10 golden fixture PUBLISHED (canonical source):** + `tests/fixtures/wire_contract/` in this repo — 1 run × 1 target × 1 month × 6 cells × + S=4; Track-A shard + Hop-A manifest (E1: null sidecar hash) + arrow shard + sidecar + (NaN row pinned) + Hop-B run manifest. Root hash (SHA-256 of `SHA256SUMS`): + `b1f3878df9ef74b25dce53a070e1711db39dfdf1c6ca3e1f5a716875ceb32f44`. Deterministic + generator: `scripts/build_wire_fixture.py` (pyarrow 23.0.1 / views_frames 1.0.0 pinned + in the fixture README). Consumers vendor + pinned-hash test per §10.1. diff --git a/scripts/build_wire_fixture.py b/scripts/build_wire_fixture.py new file mode 100644 index 0000000..769bcdc --- /dev/null +++ b/scripts/build_wire_fixture.py @@ -0,0 +1,192 @@ +"""Build the ADR-013 §10 golden fixture — the wire contract's executable spec. + +Generates the canonical small-S artifact set into ``tests/fixtures/wire_contract/``: + + fixture_run_0__lr_ged_sb__m000543.tap.zip Hop-A Track-A shard (zip) + fixture_run_0__lr_ged_sb__manifest.json Hop-A (run, target) manifest + fixture_run_0__lr_ged_sb__m000543.arrow.parquet Hop-B arrow shard (views_frames.io.arrow) + fixture_run_0__sidecar.parquet §5 GAUL sidecar (9 columns, NaN row preserved) + fixture_run_0__manifest.json Hop-B run manifest (commit marker) + SHA256SUMS the pinned hashes (§10.1 — THE cross-repo contract) + +Deterministic by construction (§10.2): provenance (`run_id`, `generated_at`) is injected +as fixed literals, values come from a fixed seed, and every zip member carries a pinned +1980-01-01 timestamp — regenerating with the same tool versions reproduces the bytes. +The *committed bytes + SHA256SUMS are canonical*; regeneration requires the pinned tool +versions recorded in the fixture README (parquet bytes vary across pyarrow versions). + +Data shape (canonical-minimal): 1 target × 1 month × 6 cells × S=4. Row 0 is +draw-degenerate on purpose (pins §6's per-row-zero-variance-is-legal rule); the sidecar's +last gid carries NaN GAUL codes (pins §5.1's NaN-preserved rule). Hop-A manifest carries +``sidecar_sha256: null`` (Erratum E1). + +A change to this fixture is a change to the contract (§10). +""" + +from __future__ import annotations + +import hashlib +import io +import json +import zipfile +from pathlib import Path + +import numpy as np +import pyarrow as pa +import pyarrow.parquet as pq +from views_frames.io import arrow as vf_arrow + +OUT = Path(__file__).resolve().parent.parent / "tests" / "fixtures" / "wire_contract" + +# ---- declared fixture identity (§10.2 injected provenance — fixed literals) ----------- +RUN_ID = "fixture_run_0" +GENERATED_AT = "2026-07-15T00:00:00Z" +TARGET = "lr_ged_sb" # canonical wire vocabulary (§7a) +TIME_ID = 543 +S = 4 +GIDS = np.array([100001, 100002, 100003, 100004, 100005, 100006], dtype=np.int64) +N = len(GIDS) +_ZIP_EPOCH = (1980, 1, 1, 0, 0, 0) # pinned member timestamp → reproducible zips + + +def _header(sharding_index: int = 0, sharding_count: int = 1) -> dict: + """The §2 contract header, in contract field order.""" + return { + "contract_version": "1.5", + "frame_type": "prediction", + "representation": "samples", + "sample_count": S, + "dtype": "float32", + "spatial_level": "pgm", + "target": TARGET, + "time_id": TIME_ID, + "run_id": RUN_ID, + "generated_at": GENERATED_AT, + "id_semantics": {"time": "views_month_id", "unit": "priogrid_id"}, + "provenance": { + "ensemble": "fixture_ensemble", + "pipeline_core_version": "0.0.0-fixture", + "reconciled": False, + }, + "sharding": {"scheme": "per_month", "index": sharding_index, "count": sharding_count}, + } + + +def _values() -> np.ndarray: + """(N, S) float32 draws; row 0 deliberately draw-degenerate (per-row zeros are legal).""" + rng = np.random.default_rng(1305) + values = rng.gamma(2.0, size=(N, S)).astype(np.float32) + values[0] = 0.0 + return values + + +def _npy_bytes(arr: np.ndarray) -> bytes: + buf = io.BytesIO() + np.save(buf, arr) + return buf.getvalue() + + +def _pinned_zip(members: dict[str, bytes]) -> bytes: + """A byte-reproducible zip: pinned member timestamps, stored (no compression).""" + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: + for name, data in members.items(): + zf.writestr(zipfile.ZipInfo(name, date_time=_ZIP_EPOCH), data) + return buf.getvalue() + + +def _sha256(data: bytes) -> str: + return hashlib.sha256(data).hexdigest() + + +def build() -> None: + OUT.mkdir(parents=True, exist_ok=True) + values = _values() + time = np.full(N, TIME_ID, dtype=np.int64) + written: dict[str, bytes] = {} + + # --- Hop-A Track-A shard (§3.1): zip of y_pred.npy + identifiers.npz + metadata.json + identifiers_npz = _pinned_zip({"time.npy": _npy_bytes(time), "unit.npy": _npy_bytes(GIDS)}) + shard_a_name = f"{RUN_ID}__{TARGET}__m{TIME_ID:06d}.tap.zip" + shard_a = _pinned_zip( + { + "y_pred.npy": _npy_bytes(values), + "identifiers.npz": identifiers_npz, + "metadata.json": json.dumps(_header(), indent=2).encode(), + } + ) + written[shard_a_name] = shard_a + + # --- Hop-A (run, target) manifest (§3.2; Erratum E1: sidecar_sha256 null at Hop A) + manifest_a_name = f"{RUN_ID}__{TARGET}__manifest.json" + manifest_a = json.dumps( + { + "contract_version": "1.5", + "run_id": RUN_ID, + "target": TARGET, + "shards": [{"name": shard_a_name, "sha256": _sha256(shard_a)}], + "expected_months": [TIME_ID], + "expected_cell_count": N, + "sidecar_sha256": None, + }, + indent=2, + ).encode() + written[manifest_a_name] = manifest_a + + # --- Hop-B arrow shard (§4.1): views_frames.io.arrow with the §2 header as metadata + shard_b_name = f"{RUN_ID}__{TARGET}__m{TIME_ID:06d}.arrow.parquet" + shard_b_path = OUT / shard_b_name + vf_arrow.save( + shard_b_path, values=values, time=time, unit=GIDS, level="pgm", metadata=_header() + ) + shard_b = shard_b_path.read_bytes() + written[shard_b_name] = shard_b + + # --- §5 sidecar: 9 pinned columns, gid-keyed; last gid carries NaN GAUL codes + sidecar_name = f"{RUN_ID}__sidecar.parquet" + nan = float("nan") + sidecar_table = pa.table( + { + "priogrid_id": pa.array(GIDS, pa.int64()), + "pg_xcoord": pa.array([10.25, 10.75, 11.25, 11.75, 12.25, 12.75], pa.float64()), + "pg_ycoord": pa.array([5.25, 5.25, 5.25, 5.75, 5.75, 5.75], pa.float64()), + "country_iso_a3": pa.array(["AAA", "AAA", "AAA", "BBB", "BBB", None], pa.string()), + "admin1_gaul1_code": pa.array([11.0, 11.0, 12.0, 21.0, 21.0, nan], pa.float64()), + "admin1_gaul1_name": pa.array(["A-one", "A-one", "A-two", "B-one", "B-one", None], pa.string()), + "admin1_gaul0_code": pa.array([1.0, 1.0, 1.0, 2.0, 2.0, nan], pa.float64()), + "admin1_gaul0_name": pa.array(["Aland", "Aland", "Aland", "Bland", "Bland", None], pa.string()), + "admin2_gaul2_code": pa.array([111.0, 112.0, 121.0, 211.0, 212.0, nan], pa.float64()), + "admin2_gaul2_name": pa.array(["A-1-1", "A-1-2", "A-2-1", "B-1-1", "B-1-2", None], pa.string()), + } + ) + pq.write_table(sidecar_table, OUT / sidecar_name) + sidecar = (OUT / sidecar_name).read_bytes() + written[sidecar_name] = sidecar + + # --- Hop-B run manifest (§4.2): ONE per run, spanning targets; sidecar hash pinned here + manifest_b_name = f"{RUN_ID}__manifest.json" + manifest_b = json.dumps( + { + "contract_version": "1.5", + "run_id": RUN_ID, + "targets": [TARGET], + "shards": [{"name": shard_b_name, "target": TARGET, "time_id": TIME_ID, "sha256": _sha256(shard_b)}], + "expected_months": [TIME_ID], + "expected_cell_count": N, + "sidecar": {"name": sidecar_name, "sha256": _sha256(sidecar)}, + }, + indent=2, + ).encode() + written[manifest_b_name] = manifest_b + + # --- write byte artifacts + SHA256SUMS (§10.1: the hash is the cross-repo contract) + for name, data in written.items(): + (OUT / name).write_bytes(data) + sums = "".join(f"{_sha256((OUT / n).read_bytes())} {n}\n" for n in sorted(written)) + (OUT / "SHA256SUMS").write_text(sums) + print(sums) + print(f"fixture written to {OUT}") + + +if __name__ == "__main__": + build() diff --git a/tests/fixtures/wire_contract/README.md b/tests/fixtures/wire_contract/README.md new file mode 100644 index 0000000..73d3d91 --- /dev/null +++ b/tests/fixtures/wire_contract/README.md @@ -0,0 +1,34 @@ +# ADR-013 §10 Golden Fixture — the wire contract's executable spec + +**Contract:** ADR-013 "The Sampled-Forecast Wire Contract" v1.5 (incl. Erratum E1). +**Canonical source:** THIS directory in views-postprocessing. Other repos **vendor a copy** +and carry a pinned-hash equality test (§10.1) — **the hash, not the bytes, is the +cross-repo contract**; on mismatch, re-vendor from here. + +**The pinned root hash (SHA-256 of `SHA256SUMS`):** + +``` +b1f3878df9ef74b25dce53a070e1711db39dfdf1c6ca3e1f5a716875ceb32f44 +``` + +## Contents (1 run × 1 target × 1 month × 6 cells × S=4) + +| File | Contract § | What it pins | +|---|---|---| +| `fixture_run_0__lr_ged_sb__m000543.tap.zip` | §3.1 | Hop-A Track-A shard: `y_pred.npy` (6,4) float32 + `identifiers.npz` (time/unit) + `metadata.json` (the §2 header, byte-pinned) | +| `fixture_run_0__lr_ged_sb__manifest.json` | §3.2 | Hop-A (run,target) manifest — shard hash, expected months/cells, `sidecar_sha256: null` (**Erratum E1**) | +| `fixture_run_0__lr_ged_sb__m000543.arrow.parquet` | §4.1 | Hop-B shard via `views_frames.io.arrow` — §2 header in the `views_frames` KV metadata; `sample` column = `tile(arange(4), 6)` (the §4.5(b) ordering oracle) | +| `fixture_run_0__sidecar.parquet` | §5.1 | The 9 pinned GAUL columns, gid-keyed; **last gid carries NaN codes** (the NaN-preserved rule, pinned) | +| `fixture_run_0__manifest.json` | §4.2 | Hop-B run manifest — one per run, spans targets, shard hashes + **the sidecar hash** (E1: it lives here, only here) | +| `SHA256SUMS` | §10.1 | Per-file hashes; its own SHA-256 is the root hash above | + +Deliberate data properties: values row 0 is draw-degenerate (pins §6's +per-row-zero-variance-is-legal); target vocabulary is `lr_ged_sb` (§7a); provenance is +injected fixed literals (`run_id="fixture_run_0"`, `generated_at="2026-07-15T00:00:00Z"`, §10.2). + +## Regeneration + +`PYTHONPATH=. python3 scripts/build_wire_fixture.py` — byte-reproducible **with the pinned +tool versions** (numpy per lockfile, `pyarrow 23.0.1`, `views_frames 1.0.0`; parquet bytes +vary across pyarrow versions). The committed bytes + `SHA256SUMS` are canonical regardless. +**A change to this fixture is a change to the contract (§10)** — do not regenerate casually. diff --git a/tests/fixtures/wire_contract/SHA256SUMS b/tests/fixtures/wire_contract/SHA256SUMS new file mode 100644 index 0000000..98f760f --- /dev/null +++ b/tests/fixtures/wire_contract/SHA256SUMS @@ -0,0 +1,5 @@ +e0fc42b615e30aaa11933c69a3efa0eb19285fe36a69b23820a33149dbbf5ac0 fixture_run_0__lr_ged_sb__m000543.arrow.parquet +61a78b9f569dff0cd97e6de673edaf55799649dffbf2c555d3733a475a55f081 fixture_run_0__lr_ged_sb__m000543.tap.zip +29d408ad8673fef692a954d26caa7890d769ad4db1c270369f772f6ebc544834 fixture_run_0__lr_ged_sb__manifest.json +ee33863ed1a67945ca2d0481d64741568fe3205449c5f3fa0791e6ec87c3535f fixture_run_0__manifest.json +3056eeda144bafa6bcc0b6ff6da041addd57d9cc8bc333526387dd2d463f0617 fixture_run_0__sidecar.parquet diff --git a/tests/fixtures/wire_contract/fixture_run_0__lr_ged_sb__manifest.json b/tests/fixtures/wire_contract/fixture_run_0__lr_ged_sb__manifest.json new file mode 100644 index 0000000..e200b13 --- /dev/null +++ b/tests/fixtures/wire_contract/fixture_run_0__lr_ged_sb__manifest.json @@ -0,0 +1,16 @@ +{ + "contract_version": "1.5", + "run_id": "fixture_run_0", + "target": "lr_ged_sb", + "shards": [ + { + "name": "fixture_run_0__lr_ged_sb__m000543.tap.zip", + "sha256": "61a78b9f569dff0cd97e6de673edaf55799649dffbf2c555d3733a475a55f081" + } + ], + "expected_months": [ + 543 + ], + "expected_cell_count": 6, + "sidecar_sha256": null +} \ No newline at end of file diff --git a/tests/fixtures/wire_contract/fixture_run_0__manifest.json b/tests/fixtures/wire_contract/fixture_run_0__manifest.json new file mode 100644 index 0000000..6b5b154 --- /dev/null +++ b/tests/fixtures/wire_contract/fixture_run_0__manifest.json @@ -0,0 +1,23 @@ +{ + "contract_version": "1.5", + "run_id": "fixture_run_0", + "targets": [ + "lr_ged_sb" + ], + "shards": [ + { + "name": "fixture_run_0__lr_ged_sb__m000543.arrow.parquet", + "target": "lr_ged_sb", + "time_id": 543, + "sha256": "e0fc42b615e30aaa11933c69a3efa0eb19285fe36a69b23820a33149dbbf5ac0" + } + ], + "expected_months": [ + 543 + ], + "expected_cell_count": 6, + "sidecar": { + "name": "fixture_run_0__sidecar.parquet", + "sha256": "3056eeda144bafa6bcc0b6ff6da041addd57d9cc8bc333526387dd2d463f0617" + } +} \ No newline at end of file diff --git a/tests/test_wire_fixture.py b/tests/test_wire_fixture.py new file mode 100644 index 0000000..6dd9d82 --- /dev/null +++ b/tests/test_wire_fixture.py @@ -0,0 +1,115 @@ +"""Conformance tests over the ADR-013 §10 golden fixture — the executable spec. + +These are the canonical-source-side checks: integrity (hashes), the Track-A shard's +round-trip, the arrow shard's header + ordering oracle, the sidecar's pinned schema +(incl. the NaN-preserved rule), both manifests' consistency, and Erratum E1. The other +repos vendor these bytes and run their own side (§10.1). +""" + +import hashlib +import io +import json +import zipfile +from pathlib import Path + +import numpy as np +import pyarrow.parquet as pq +from views_frames.io import arrow as vf_arrow + +from views_postprocessing.delivery.draws import assert_draws_uncollapsed + +FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract" +SHARD_A = FIX / "fixture_run_0__lr_ged_sb__m000543.tap.zip" +MANIFEST_A = FIX / "fixture_run_0__lr_ged_sb__manifest.json" +SHARD_B = FIX / "fixture_run_0__lr_ged_sb__m000543.arrow.parquet" +SIDECAR = FIX / "fixture_run_0__sidecar.parquet" +MANIFEST_B = FIX / "fixture_run_0__manifest.json" + +_SIDECAR_COLS = [ + "pg_xcoord", "pg_ycoord", "country_iso_a3", + "admin1_gaul1_code", "admin1_gaul1_name", + "admin1_gaul0_code", "admin1_gaul0_name", + "admin2_gaul2_code", "admin2_gaul2_name", +] + + +def _sha256(path: Path) -> str: + return hashlib.sha256(path.read_bytes()).hexdigest() + + +# --- §10.1 integrity --------------------------------------------------------------- + + +def test_sha256sums_matches_every_file(): + for line in (FIX / "SHA256SUMS").read_text().splitlines(): + digest, name = line.split() + assert _sha256(FIX / name) == digest, f"fixture drift: {name}" + + +# --- §3.1 Track-A shard ------------------------------------------------------------ + + +def test_track_a_shard_round_trips(): + with zipfile.ZipFile(SHARD_A) as zf: + assert sorted(zf.namelist()) == ["identifiers.npz", "metadata.json", "y_pred.npy"] + values = np.load(io.BytesIO(zf.read("y_pred.npy"))) + header = json.loads(zf.read("metadata.json")) + with zipfile.ZipFile(io.BytesIO(zf.read("identifiers.npz"))) as ids: + time = np.load(io.BytesIO(ids.read("time.npy"))) + unit = np.load(io.BytesIO(ids.read("unit.npy"))) + assert values.shape == (6, 4) and values.dtype == np.float32 + assert time.shape == unit.shape == (6,) + assert header["contract_version"] == "1.5" + assert header["target"] == "lr_ged_sb" # §7a vocabulary + assert header["id_semantics"] == {"time": "views_month_id", "unit": "priogrid_id"} + # the payload satisfies the §6 policy gate against its own declared header + assert_draws_uncollapsed(values, header["sample_count"], s_min=2) is None + + +# --- §4.1 / §4.5 arrow shard ------------------------------------------------------- + + +def test_arrow_shard_header_and_values_match_track_a(): + state = vf_arrow.load(SHARD_B) + with zipfile.ZipFile(SHARD_A) as zf: + a_values = np.load(io.BytesIO(zf.read("y_pred.npy"))) + a_header = json.loads(zf.read("metadata.json")) + np.testing.assert_array_equal(np.asarray(state["values"], dtype=np.float32), a_values) + assert state["metadata"] == a_header # ONE header, both envelopes — byte-level parity + + +def test_arrow_shard_sample_column_is_the_ordering_oracle(): + # §4.5(b): consumers must verify this before positional reshape. + table = pq.read_table(SHARD_B) + sample = table.column("sample").to_numpy() + np.testing.assert_array_equal(sample, np.tile(np.arange(4, dtype=np.int32), 6)) + + +# --- §5.1 sidecar ------------------------------------------------------------------- + + +def test_sidecar_schema_pinned_and_nan_row_preserved(): + table = pq.read_table(SIDECAR) + assert table.column_names == ["priogrid_id"] + _SIDECAR_COLS + last = {c: table.column(c)[-1].as_py() for c in _SIDECAR_COLS} + assert last["country_iso_a3"] is None # the NaN-GAUL row EXISTS — never pre-dropped + assert last["admin1_gaul0_code"] != last["admin1_gaul0_code"] # NaN + assert table.num_rows == 6 # gid-set matches the forecast gid-set (§5.2) + + +# --- §3.2 / §4.2 manifests ---------------------------------------------------------- + + +def test_hop_a_manifest_consistent_and_erratum_e1(): + m = json.loads(MANIFEST_A.read_text()) + assert m["sidecar_sha256"] is None # Erratum E1: no sidecar hash at Hop A + assert m["shards"][0]["sha256"] == _sha256(SHARD_A) + assert m["expected_months"] == [543] and m["expected_cell_count"] == 6 + + +def test_hop_b_run_manifest_is_the_commit_marker(): + m = json.loads(MANIFEST_B.read_text()) + assert m["targets"] == ["lr_ged_sb"] # one manifest per run, spanning targets (§4.2) + assert m["shards"][0]["sha256"] == _sha256(SHARD_B) + assert m["sidecar"]["sha256"] == _sha256(SIDECAR) # the E1 home of the sidecar hash + assert m["expected_months"] == [543] and m["expected_cell_count"] == 6