From b8f1d4e17b38d542cdc5a4551ba818e1a7fe7725 Mon Sep 17 00:00:00 2001 From: Polichinl Date: Thu, 16 Jul 2026 00:47:03 +0200 Subject: [PATCH] =?UTF-8?q?feat(track-a-source):=20ADR-013=20leg=202b=20?= =?UTF-8?q?=E2=80=94=20Hop-A=20source=20adapter=20(archive=20=E2=86=92=20P?= =?UTF-8?q?redictionFrame)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The consumer side of §3: read a Track-A archive, verify the shard bytes against the Hop-A manifest's declared sha256 (§3.2 — the verification this repo owns on read), check the §3.1 member triple, assert the payload matches the header's own declarations (sample_count, dtype, identifier lengths — the per-hop ingest assert), gate on contract major version (§2.1), and build the interior PredictionFrame via frames.build_prediction_frame. frames_for_target() assembles a (run, target) from its manifest — manifest content is identity (§3.3): listed shards must all be present, months must cover exactly expected_months, cell counts must match; a torn run is never assembled. Pandas-free by construction; mechanism guards only — the §6 policy gate (delivery.draws, S_min) stays at the FAO-facing upload. Tests use the §10 golden fixture as oracle: happy path from canonical bytes, tamper paths from surgically corrupted copies breaking exactly one declared fact each (hash, member set, S, dtype, major version, target, months, cell count, identifier length). MINOR header drift (1.9) accepted per §2.1. Co-Authored-By: Claude Fable 5 --- tests/test_track_a_source.py | 187 +++++++++++++++++++ views_postprocessing/unfao/track_a_source.py | 146 +++++++++++++++ 2 files changed, 333 insertions(+) create mode 100644 tests/test_track_a_source.py create mode 100644 views_postprocessing/unfao/track_a_source.py diff --git a/tests/test_track_a_source.py b/tests/test_track_a_source.py new file mode 100644 index 0000000..16fe0e8 --- /dev/null +++ b/tests/test_track_a_source.py @@ -0,0 +1,187 @@ +"""The Hop-A source adapter against the §10 golden fixture — happy path from the +canonical bytes, tamper paths from surgically corrupted copies of them. + +Every tamper case starts from the REAL fixture and breaks exactly one declared fact +(hash, member set, sample_count, dtype, major version, target, month set, cell count, +identifier length) — proving the adapter rejects precisely what the contract says it +must, not merely malformed junk. +""" + +import io +import json +import zipfile +from pathlib import Path + +import numpy as np +import pytest + +from views_postprocessing.unfao import track_a_source as tas + +FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract" +SHARD = (FIX / "fixture_run_0__lr_ged_sb__m000543.tap.zip").read_bytes() +MANIFEST = json.loads((FIX / "fixture_run_0__lr_ged_sb__manifest.json").read_text()) +SHARD_NAME = MANIFEST["shards"][0]["name"] +SHARD_SHA = MANIFEST["shards"][0]["sha256"] + + +def _retouched_shard(**member_overrides: bytes) -> bytes: + """The fixture shard with named members replaced — everything else byte-identical.""" + with zipfile.ZipFile(io.BytesIO(SHARD)) as zf: + members = {name: zf.read(name) for name in zf.namelist()} + members.update(member_overrides) + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: + for name, data in members.items(): + zf.writestr(name, data) + return buf.getvalue() + + +def _header(**overrides) -> bytes: + with zipfile.ZipFile(io.BytesIO(SHARD)) as zf: + header = json.loads(zf.read("metadata.json")) + header.update(overrides) + return json.dumps(header).encode() + + +def _sha(data: bytes) -> str: + import hashlib + + return hashlib.sha256(data).hexdigest() + + +# --- happy path: the golden fixture is the oracle ----------------------------------- + + +def test_read_manifest_accepts_the_fixture_manifest(): + manifest = tas.read_manifest((FIX / "fixture_run_0__lr_ged_sb__manifest.json").read_bytes()) + assert manifest["run_id"] == "fixture_run_0" + assert manifest["sidecar_sha256"] is None # Erratum E1 + + +def test_read_shard_round_trips_the_fixture(): + frame, header = tas.read_shard(SHARD, expected_sha256=SHARD_SHA) + assert frame.n_rows == 6 and frame.sample_count == 4 + assert frame.values.dtype == np.float32 + np.testing.assert_array_equal(np.asarray(frame.index.unit), np.arange(100001, 100007)) + np.testing.assert_array_equal(np.asarray(frame.index.time), np.full(6, 543)) + assert header["contract_version"] == "1.5" and header["target"] == "lr_ged_sb" + + +def test_frames_for_target_assembles_the_run(): + frame = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD}) + assert frame.n_rows == 6 and frame.sample_count == 4 + + +# --- §3.2 hash verification ----------------------------------------------------------- + + +def test_tampered_bytes_rejected(): + with pytest.raises(tas.TrackASourceError, match="sha256 mismatch"): + tas.read_shard(SHARD + b"\x00", expected_sha256=SHARD_SHA) + + +# --- §3.1 member set ------------------------------------------------------------------- + + +def test_missing_member_rejected(): + with zipfile.ZipFile(io.BytesIO(SHARD)) as zf: + members = {n: zf.read(n) for n in zf.namelist() if n != "identifiers.npz"} + buf = io.BytesIO() + with zipfile.ZipFile(buf, "w", zipfile.ZIP_STORED) as zf: + for name, data in members.items(): + zf.writestr(name, data) + bad = buf.getvalue() + with pytest.raises(tas.TrackASourceError, match="archive members"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +# --- header-vs-payload (per-hop ingest asserts) ---------------------------------------- + + +def test_header_lying_about_sample_count_rejected(): + bad = _retouched_shard(**{"metadata.json": _header(sample_count=1024)}) + with pytest.raises(tas.TrackASourceError, match="sample_count=1024"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +def test_header_lying_about_dtype_rejected(): + bad = _retouched_shard(**{"metadata.json": _header(dtype="float64")}) + with pytest.raises(tas.TrackASourceError, match="dtype"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +def test_identifier_length_mismatch_rejected(): + with zipfile.ZipFile(io.BytesIO(SHARD)) as zf: + 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"))) + short_ids = io.BytesIO() + with zipfile.ZipFile(short_ids, "w", zipfile.ZIP_STORED) as zf: + for name, arr in (("time.npy", time[:-1]), ("unit.npy", unit)): + buf = io.BytesIO() + np.save(buf, arr) + zf.writestr(name, buf.getvalue()) + bad = _retouched_shard(**{"identifiers.npz": short_ids.getvalue()}) + with pytest.raises(tas.TrackASourceError, match="identifier lengths"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +# --- §2.1 versioning ------------------------------------------------------------------- + + +def test_major_version_bump_rejected_in_shard_header(): + bad = _retouched_shard(**{"metadata.json": _header(contract_version="2.0")}) + with pytest.raises(tas.TrackASourceError, match="major"): + tas.read_shard(bad, expected_sha256=_sha(bad)) + + +def test_major_version_bump_rejected_in_manifest(): + with pytest.raises(tas.TrackASourceError, match="major"): + tas.read_manifest(json.dumps({**MANIFEST, "contract_version": "2.0"}).encode()) + + +def test_minor_version_drift_accepted(): + bumped = _retouched_shard(**{"metadata.json": _header(contract_version="1.9")}) + frame, header = tas.read_shard(bumped, expected_sha256=_sha(bumped)) + assert header["contract_version"] == "1.9" and frame.n_rows == 6 + + +# --- run assembly (§3.2 completeness; §3.3 manifest-is-identity) ---------------------- + + +def test_missing_shard_bytes_rejected(): + with pytest.raises(tas.TrackASourceError, match="not provided"): + tas.frames_for_target(MANIFEST, {}) + + +def test_shard_target_disagreeing_with_manifest_rejected(): + bad = _retouched_shard(**{"metadata.json": _header(target="lr_ged_ns")}) + manifest = {**MANIFEST, "shards": [{"name": SHARD_NAME, "sha256": _sha(bad)}]} + with pytest.raises(tas.TrackASourceError, match="target"): + tas.frames_for_target(manifest, {SHARD_NAME: bad}) + + +def test_wrong_month_coverage_rejected(): + manifest = {**MANIFEST, "expected_months": [543, 544]} + with pytest.raises(tas.TrackASourceError, match="not provided"): + # month 544's shard is absent entirely — caught at the bytes gate + tas.frames_for_target( + {**manifest, "shards": MANIFEST["shards"] + [{"name": "m544", "sha256": "0" * 64}]}, + {SHARD_NAME: SHARD}, + ) + bad = _retouched_shard(**{"metadata.json": _header(time_id=999)}) + manifest = {**MANIFEST, "shards": [{"name": SHARD_NAME, "sha256": _sha(bad)}]} + with pytest.raises(tas.TrackASourceError, match="months covered"): + tas.frames_for_target(manifest, {SHARD_NAME: bad}) + + +def test_wrong_cell_count_rejected(): + manifest = {**MANIFEST, "expected_cell_count": 7} + with pytest.raises(tas.TrackASourceError, match="cells"): + tas.frames_for_target(manifest, {SHARD_NAME: SHARD}) + + +def test_manifest_missing_required_field_rejected(): + truncated = {k: v for k, v in MANIFEST.items() if k != "expected_months"} + with pytest.raises(tas.TrackASourceError, match="expected_months"): + tas.read_manifest(json.dumps(truncated).encode()) diff --git a/views_postprocessing/unfao/track_a_source.py b/views_postprocessing/unfao/track_a_source.py new file mode 100644 index 0000000..8dc9204 --- /dev/null +++ b/views_postprocessing/unfao/track_a_source.py @@ -0,0 +1,146 @@ +"""Hop-A source adapter: verified Track-A archives → interior ``PredictionFrame``s +(ADR-013 §3 consumer side; the delivery *source* adapter required by the #88 review +amendment — the manager calls this, representation logic never lives inline in it). + +Pandas-free by construction: a Track-A archive is zip + npy + json; this module turns it +into declared numpy primitives and hands them to ``frames.build_prediction_frame``. The +new wire path never touches pandas. + +Declare-don't-guess, per hop: every check here is a **mechanism** guard — the artifact +must match what its own manifest and header *declare* (hashes, counts, shapes, versions). +The §6 *policy* gate (``delivery.draws``, S_min) is separate and runs at the FAO-facing +upload, not here. + +What this module verifies (fail loud, naming expected vs found): +- manifest: contract major-version compatible (§2.1 — MINOR additive, MAJOR breaking); +- shard bytes match the manifest's declared sha256 (§3.2: views-postprocessing verifies + Hop-A hashes on read); +- archive members are exactly the §3.1 triple; header major-version compatible; +- payload matches the header's own declaration: ``(N, sample_count)`` float32, identifier + arrays of length N (the per-hop ingest assert); +- a run's shards cover exactly the manifest's expected months and cell count (§4.2a + feeds on this when the manager awaits all targets). +""" + +from __future__ import annotations + +import hashlib +import io +import json +import zipfile + +import numpy as np +from views_frames import PredictionFrame + +from views_postprocessing.unfao.frames import build_prediction_frame + +_ARCHIVE_MEMBERS = {"y_pred.npy", "identifiers.npz", "metadata.json"} +_SUPPORTED_CONTRACT_MAJOR = "1" + + +class TrackASourceError(ValueError): + """A Track-A artifact does not match what its manifest or header declares.""" + + +def _require_contract_major(version: str, *, where: str) -> None: + major = str(version).split(".", 1)[0] + if major != _SUPPORTED_CONTRACT_MAJOR: + raise TrackASourceError( + f"{where}: contract_version {version!r} has major {major!r}; this consumer " + f"speaks major {_SUPPORTED_CONTRACT_MAJOR!r} (§2.1 — MAJOR bumps are breaking)." + ) + + +def read_manifest(manifest_bytes: bytes) -> dict: + """Parse and validate a Hop-A (run, target) manifest (§3.2).""" + manifest = json.loads(manifest_bytes) + _require_contract_major(manifest.get("contract_version", "?"), where="manifest") + for key in ("run_id", "target", "shards", "expected_months", "expected_cell_count"): + if key not in manifest: + raise TrackASourceError(f"manifest: required field {key!r} is missing.") + return manifest + + +def read_shard(shard_bytes: bytes, *, expected_sha256: str) -> tuple[PredictionFrame, dict]: + """One verified Track-A shard → ``(PredictionFrame, header)``. + + ``expected_sha256`` is the manifest's declaration for these bytes — the §3.2 + hash verification this repo owns on read. + """ + actual = hashlib.sha256(shard_bytes).hexdigest() + if actual != expected_sha256: + raise TrackASourceError( + f"shard: sha256 mismatch — manifest declares {expected_sha256}, " + f"bytes hash to {actual}. Refusing a tampered or torn artifact." + ) + with zipfile.ZipFile(io.BytesIO(shard_bytes)) as zf: + members = set(zf.namelist()) + if members != _ARCHIVE_MEMBERS: + raise TrackASourceError( + f"shard: archive members {sorted(members)} != required {sorted(_ARCHIVE_MEMBERS)} (§3.1)." + ) + header = json.loads(zf.read("metadata.json")) + values = np.load(io.BytesIO(zf.read("y_pred.npy"))) + 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"))) + + _require_contract_major(header.get("contract_version", "?"), where="shard header") + declared_s = header.get("sample_count") + if values.ndim != 2 or values.shape[1] != declared_s: + found = values.shape if values.ndim == 2 else f"{values.ndim}-D" + raise TrackASourceError( + f"shard: header declares sample_count={declared_s} but y_pred is {found} — " + f"the artifact lies about its own draws (per-hop ingest assert)." + ) + if str(values.dtype) != header.get("dtype"): + raise TrackASourceError( + f"shard: header declares dtype={header.get('dtype')!r}, y_pred is {values.dtype}." + ) + n = values.shape[0] + if time.shape != (n,) or unit.shape != (n,): + raise TrackASourceError( + f"shard: identifier lengths (time={time.shape}, unit={unit.shape}) != N={n}." + ) + return build_prediction_frame(values, time, unit), header + + +def frames_for_target(manifest: dict, shard_bytes_by_name: dict) -> PredictionFrame: + """A (run, target)'s verified shards → one interior ``PredictionFrame``. + + ``shard_bytes_by_name`` maps shard ``name`` → downloaded bytes; the manifest is the + only source of which shards exist (§3.3: names are locators, manifest content is + identity). Verifies run completeness against the manifest's own declarations — + months covered exactly, cell count per month — then stacks months into one frame. + """ + frames, months_seen = [], [] + for entry in manifest["shards"]: + name = entry["name"] + if name not in shard_bytes_by_name: + raise TrackASourceError( + f"run: manifest lists shard {name!r} but its bytes were not provided — " + f"an unmanifested or missing shard must not be silently skipped." + ) + frame, header = read_shard(shard_bytes_by_name[name], expected_sha256=entry["sha256"]) + if header.get("target") != manifest["target"]: + raise TrackASourceError( + f"run: shard header target {header.get('target')!r} != manifest target " + f"{manifest['target']!r}." + ) + if frame.n_rows != manifest["expected_cell_count"]: + raise TrackASourceError( + f"run: shard {name!r} carries {frame.n_rows} cells, manifest expects " + f"{manifest['expected_cell_count']}." + ) + frames.append(frame) + months_seen.append(int(header.get("time_id"))) + + if sorted(months_seen) != sorted(int(m) for m in manifest["expected_months"]): + raise TrackASourceError( + f"run: months covered {sorted(months_seen)} != manifest expected " + f"{sorted(manifest['expected_months'])} — a torn run must not be assembled." + ) + values = np.concatenate([f.values for f in frames], axis=0) + time = np.concatenate([np.asarray(f.index.time) for f in frames]) + unit = np.concatenate([np.asarray(f.index.unit) for f in frames]) + return build_prediction_frame(values, time, unit)