feat(wire): S3/#108 — arrow shard writer (byte parity, header pass-through)

docs/ADRs/013_sampled_forecast_wire_contract.md

@@ -1029,6 +1029,21 @@ record execution progress against it.
   ADR (§0–§11) has now been independently falsification-audited; every finding
   fixed same-day; 40 permanent guards enforce the fixes
   (`tests/test_falsify_adr013_*.py`).**
+- **2026-07-20 — §10 fixture RE-BASELINED to the production toolchain (maintainer
+  decision, Option A; a §10 contract-change event).** The fixture's Hop-B byte
+  artifacts had been generated under pyarrow 23.0.1 (an unlocked dev
+  environment); the delivery repo's actual locked toolchain is **pyarrow 16.1.0**
+  and cannot be newer — `views-pipeline-core → viewser → views-storage` hard-pins
+  `pyarrow <17`. The three Hop-B artifacts (arrow shard, sidecar, run manifest)
+  + `SHA256SUMS` were regenerated under 16.1.0; **Track-A bytes are unchanged**
+  (verified byte-identical — producer-side conformance unaffected). New root
+  hash: `9658a6484cc9d975412e52624d52f328985f14cf58e3fc9fbdf3e64ab5a0564b`.
+  pyarrow is now an explicit first-class pin in this repo's pyproject
+  (`>=16.1.0,<17`). Vendors notified to re-vendor (their §10.1 pinned-hash tests
+  catch it regardless — the mechanism's second live save). **Platform debt noted
+  (maintainer: "fix this through the pipeline"): lifting the `<17` ceiling at its
+  source is filed as pipeline-core#280; when it lifts, the fixture gets one
+  planned re-baseline, coordinated, never a drive-by.**
 - **2026-07-19 — retention direction given (maintainer): a configurable retention
   period with automatic deletion** (e.g. 12 or 36 months — the value to be decided
   with the owner). This settles the *shape* of the §3.5 policy; the owner

pyproject.toml

@@ -12,6 +12,7 @@ license = "MIT"
 python = ">=3.11,<3.15"
 views-pipeline-core = ">=2.1.3,<3.0.0"
 views-frames = ">=1.0,<2"
+pyarrow = ">=16.1.0,<17.0.0"
 
 [build-system]
 requires = ["poetry-core>=2.0.0,<3.0.0"]

tests/fixtures/wire_contract/README.md

@@ -8,7 +8,7 @@ cross-repo contract**; on mismatch, re-vendor from here.
 **The pinned root hash (SHA-256 of `SHA256SUMS`):**
 
 ```
-b1f3878df9ef74b25dce53a070e1711db39dfdf1c6ca3e1f5a716875ceb32f44
+9658a6484cc9d975412e52624d52f328985f14cf58e3fc9fbdf3e64ab5a0564b
 ```
 
 ## Contents (1 run × 1 target × 1 month × 6 cells × S=4)
@@ -29,6 +29,6 @@ injected fixed literals (`run_id="fixture_run_0"`, `generated_at="2026-07-15T00:
 ## 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
+tool versions** (numpy per lockfile, `pyarrow 16.1.0`, `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.

tests/fixtures/wire_contract/SHA256SUMS

@@ -1,5 +1,5 @@
-e0fc42b615e30aaa11933c69a3efa0eb19285fe36a69b23820a33149dbbf5ac0  fixture_run_0__lr_ged_sb__m000543.arrow.parquet
+203650fd6933b2366ed373ad8669d1c8b4c97553e30db9d2f07985a201012c54  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
+bf516f0f2318f1d2d7d21b5adaac07f78ba89bc8418f956494084b7301292990  fixture_run_0__manifest.json
+eb959b153a2eafcb5917ee80ed0ac40d88ca9c24243796fefd272afea5c71401  fixture_run_0__sidecar.parquet

tests/fixtures/wire_contract/fixture_run_0__manifest.json

@@ -9,7 +9,7 @@
       "name": "fixture_run_0__lr_ged_sb__m000543.arrow.parquet",
       "target": "lr_ged_sb",
       "time_id": 543,
-      "sha256": "e0fc42b615e30aaa11933c69a3efa0eb19285fe36a69b23820a33149dbbf5ac0"
+      "sha256": "203650fd6933b2366ed373ad8669d1c8b4c97553e30db9d2f07985a201012c54"
     }
   ],
   "expected_months": [
@@ -18,6 +18,6 @@
   "expected_cell_count": 6,
   "sidecar": {
     "name": "fixture_run_0__sidecar.parquet",
-    "sha256": "3056eeda144bafa6bcc0b6ff6da041addd57d9cc8bc333526387dd2d463f0617"
+    "sha256": "eb959b153a2eafcb5917ee80ed0ac40d88ca9c24243796fefd272afea5c71401"
   }
 }

tests/test_wire_shard.py

@@ -0,0 +1,90 @@
+"""Hop-B shard writer (wire/shard.py) — byte parity with the fixture arrow shard.
+
+The byte tests assert the pinned toolchain (fixture README: pyarrow 23.0.1) and
+FAIL LOUD on drift — never skip-silent: a quiet skip would read as conformance.
+"""
+
+import hashlib
+import zipfile
+from pathlib import Path
+
+import pyarrow
+import pytest
+
+from views_postprocessing.unfao import track_a_source
+from views_postprocessing.unfao.frame_extraction import month_slice
+from views_postprocessing.unfao.wire import header as wire_header
+from views_postprocessing.unfao.wire import shard as wire_shard
+
+_FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract"
+_PINNED_PYARROW = "16.1.0"
+
+
+def _fixture_frame_and_header():
+    manifest_bytes = (_FIX / "fixture_run_0__lr_ged_sb__manifest.json").read_bytes()
+    manifest = track_a_source.read_manifest(manifest_bytes)
+    shard_bytes = (_FIX / manifest["shards"][0]["name"]).read_bytes()
+    return track_a_source.read_shard(
+        shard_bytes, expected_sha256=manifest["shards"][0]["sha256"]
+    )
+
+
+def test_toolchain_is_pinned():
+    assert pyarrow.__version__ == _PINNED_PYARROW, (
+        f"pinned toolchain violated: byte-parity oracle requires pyarrow "
+        f"{_PINNED_PYARROW}, found {pyarrow.__version__} — re-pin the environment "
+        f"or regenerate the fixture (a contract change, §10)."
+    )
+
+
+def test_byte_parity_with_the_fixture_arrow_shard(tmp_path):
+    frame, track_a_header = _fixture_frame_and_header()
+    values, time, unit = month_slice(frame, 543)
+    built_header = wire_header.build_header(
+        sample_count=track_a_header["sample_count"],
+        dtype=track_a_header["dtype"],
+        target=track_a_header["target"],
+        time_id=track_a_header["time_id"],
+        run_id=track_a_header["run_id"],
+        generated_at=track_a_header["generated_at"],
+        provenance=track_a_header["provenance"],
+        sharding_index=track_a_header["sharding"]["index"],
+        sharding_count=track_a_header["sharding"]["count"],
+    )
+    name, sha = wire_shard.write_shard(
+        values, time, unit, header=built_header, directory=tmp_path
+    )
+    assert name == "fixture_run_0__lr_ged_sb__m000543.arrow.parquet"
+    canonical = (_FIX / name).read_bytes()
+    assert (tmp_path / name).read_bytes() == canonical
+    assert sha == hashlib.sha256(canonical).hexdigest()
+
+
+def test_sha_matches_sha256sums(tmp_path):
+    frame, track_a_header = _fixture_frame_and_header()
+    values, time, unit = month_slice(frame, 543)
+    _, sha = wire_shard.write_shard(
+        values, time, unit, header=track_a_header, directory=tmp_path
+    )
+    sums = (_FIX / "SHA256SUMS").read_text()
+    assert f"{sha}  fixture_run_0__lr_ged_sb__m000543.arrow.parquet" in sums
+
+
+def test_month_slice_missing_month_fails_loud():
+    frame, _ = _fixture_frame_and_header()
+    with pytest.raises(ValueError, match="month 999"):
+        month_slice(frame, 999)
+
+
+def test_header_passthrough_survives_roundtrip(tmp_path):
+    # The embedded metadata equals the Track-A header — one header, both envelopes.
+    from views_frames.io import arrow as vf_arrow
+
+    frame, track_a_header = _fixture_frame_and_header()
+    values, time, unit = month_slice(frame, 543)
+    name, _ = wire_shard.write_shard(
+        values, time, unit, header=track_a_header, directory=tmp_path
+    )
+    with zipfile.ZipFile(_FIX / "fixture_run_0__lr_ged_sb__m000543.tap.zip") as zf:
+        assert zf.read("metadata.json")  # fixture sanity
+    assert vf_arrow.load(tmp_path / name)["metadata"] == track_a_header

views_postprocessing/unfao/frame_extraction.py

@@ -37,3 +37,20 @@ def cells_of(frame: PredictionFrame) -> set[int]:
 def months_of(frame: PredictionFrame) -> NDArray[np.int64]:
     """The distinct month ids present in the frame, ascending (its index ``time`` axis)."""
     return np.unique(np.asarray(frame.index.time, dtype=np.int64))
+
+
+def month_slice(
+    frame: PredictionFrame, month_id: int
+) -> tuple[NDArray[np.float32], NDArray[np.int64], NDArray[np.int64]]:
+    """One month's ``(values, time, unit)`` primitives, row order preserved.
+
+    The per-(target, month) sharding cut (ADR-013 §4.1) as a seam concern: the
+    shard writer stays frame-API-free by receiving primitives. A month absent
+    from the frame fails loud — the caller declared it, the frame must carry it.
+    """
+    time = np.asarray(frame.index.time, dtype=np.int64)
+    mask = time == month_id
+    if not mask.any():
+        raise ValueError(f"month {month_id} not present in frame (months: {months_of(frame)}).")
+    unit = np.asarray(frame.index.unit, dtype=np.int64)
+    return frame.values[mask], time[mask], unit[mask]

views_postprocessing/unfao/wire/shard.py

@@ -0,0 +1,35 @@
+"""The Hop-B arrow shard writer (ADR-013 §4.1).
+
+One ``views_frames.io.arrow`` file per (target, month), the §2 header embedded —
+written from declared primitives (one month's ``values/time/unit`` slice, cut by
+``frame_extraction.month_slice``) so this module never touches the frame API.
+The shard's file name derives from the header's own declarations (never invented
+here), and the returned SHA-256 of the complete file bytes is what the §4.2 run
+manifest will pin.
+
+Byte determinism: given identical inputs and the pinned toolchain (pyarrow per the
+fixture README), the emitted bytes equal the §10 fixture's — the test is byte
+parity, not schema similarity.
+"""
+
+from __future__ import annotations
+
+import hashlib
+from pathlib import Path
+
+from views_frames.io import arrow as vf_arrow
+
+from views_postprocessing.unfao.wire.naming import shard_name
+
+
+def write_shard(values, time, unit, *, header: dict, directory: Path) -> tuple[str, str]:
+    """Write one (target, month) shard; return ``(file_name, sha256_of_bytes)``.
+
+    ``header`` is the §2 dict from ``wire.header.build_header`` — its ``run_id``,
+    ``target`` and ``time_id`` name the file (§4.1b); the whole dict is embedded
+    as the arrow metadata (one header, both envelopes).
+    """
+    name = shard_name(header["run_id"], header["target"], header["time_id"])
+    path = Path(directory) / name
+    vf_arrow.save(path, values=values, time=time, unit=unit, level="pgm", metadata=header)
+    return name, hashlib.sha256(path.read_bytes()).hexdigest()