feat(wire): S6/#111 — contract-mode Hop-A inbound (await-all-targets, declared identity)

tests/test_track_a_source.py

@@ -68,8 +68,11 @@ def test_read_shard_round_trips_the_fixture():
 
 
 def test_frames_for_target_assembles_the_run():
-    frame = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD})
+    frame, headers = tas.frames_for_target(MANIFEST, {SHARD_NAME: SHARD})
     assert frame.n_rows == 6 and frame.sample_count == 4
+    # headers ride along in manifest shard order (provenance pass-through, §10.2)
+    assert [h["time_id"] for h in headers] == [543]
+    assert headers[0]["provenance"]["ensemble"] == "fixture_ensemble"
 
 
 # --- §3.2 hash verification -----------------------------------------------------------

tests/test_wire_source_selection.py

@@ -0,0 +1,127 @@
+"""Contract-mode Hop-A inbound (wire/source_selection.py) — dict-backed fake store
+serving the golden fixture; selection policy, await-all-targets, declared identity."""
+
+import json
+from pathlib import Path
+
+import pytest
+
+from views_postprocessing.unfao.wire import source_selection as sel
+
+_FIX = Path(__file__).resolve().parent / "fixtures" / "wire_contract"
+_MANIFEST_NAME = "fixture_run_0__lr_ged_sb__manifest.json"
+_SHARD_NAME = "fixture_run_0__lr_ged_sb__m000543.tap.zip"
+
+
+class FakeStore:
+    """The three-method store port over a dict of (metadata, bytes) records."""
+
+    def __init__(self, records):
+        # records: list of (file_id, metadata_dict, payload_bytes); newest LAST.
+        self._records = records
+        self.calls = []
+
+    def latest_file_id(self, filters):
+        self.calls.append(("latest_file_id", dict(filters)))
+        for file_id, meta, _ in reversed(self._records):
+            if all(meta.get(k) == v for k, v in filters.items()):
+                return file_id
+        return None
+
+    def file_metadata(self, file_id):
+        return next(m for i, m, _ in self._records if i == file_id)
+
+    def download(self, file_id):
+        return next(b for i, _, b in self._records if i == file_id)
+
+
+def _fixture_store() -> FakeStore:
+    return FakeStore(
+        [
+            (
+                "f-shard",
+                {**sel.HOP_A_SHARD_FILTERS, "name": _SHARD_NAME},
+                (_FIX / _SHARD_NAME).read_bytes(),
+            ),
+            (
+                "f-manifest",
+                {**sel.HOP_A_MANIFEST_FILTERS, "name": _MANIFEST_NAME},
+                (_FIX / _MANIFEST_NAME).read_bytes(),
+            ),
+        ]
+    )
+
+
+def test_happy_path_assembles_the_fixture_run():
+    result = sel.fetch_run(
+        _fixture_store(),
+        expected_targets=["lr_ged_sb"],
+        expected_ensemble="fixture_ensemble",
+    )
+    frame, headers = result["lr_ged_sb"]
+    assert frame.n_rows == 6 and frame.sample_count == 4
+    assert headers[0]["run_id"] == "fixture_run_0"
+
+
+def test_empty_store_refuses_loud():
+    with pytest.raises(sel.SourceSelectionError, match="nothing has been published"):
+        sel.fetch_run(
+            FakeStore([]), expected_targets=["lr_ged_sb"], expected_ensemble="e"
+        )
+
+
+def test_missing_declared_target_refuses_loud():
+    # The store has lr_ged_sb only; the delivery declares two targets (§4.2a).
+    with pytest.raises(sel.SourceSelectionError, match="lr_ged_ns.*awaiting all"):
+        sel.fetch_run(
+            _fixture_store(),
+            expected_targets=["lr_ged_sb", "lr_ged_ns"],
+            expected_ensemble="fixture_ensemble",
+        )
+
+
+def test_wrong_declared_ensemble_refuses_loud():
+    with pytest.raises(sel.SourceSelectionError, match="rusty_bucket.*never falls back"):
+        sel.fetch_run(
+            _fixture_store(),
+            expected_targets=["lr_ged_sb"],
+            expected_ensemble="rusty_bucket",
+        )
+
+
+def test_manifested_shard_missing_from_store_refuses_loud():
+    store = _fixture_store()
+    store._records = [r for r in store._records if r[0] != "f-shard"]
+    with pytest.raises(sel.SourceSelectionError, match="torn run"):
+        sel.fetch_run(
+            store, expected_targets=["lr_ged_sb"], expected_ensemble="fixture_ensemble"
+        )
+
+
+def test_manifest_name_content_mismatch_refuses_loud():
+    # A second target's manifest name serving OTHER content: names are locators,
+    # content is identity (§3.3).
+    manifest = json.loads((_FIX / _MANIFEST_NAME).read_text())
+    imposter_name = "fixture_run_0__lr_ged_ns__manifest.json"
+    store = _fixture_store()
+    store._records.insert(
+        0,
+        (
+            "f-imposter",
+            {**sel.HOP_A_MANIFEST_FILTERS, "name": imposter_name},
+            json.dumps(manifest).encode(),  # still claims target lr_ged_sb
+        ),
+    )
+    with pytest.raises(sel.SourceSelectionError, match="content is identity"):
+        sel.fetch_run(
+            store,
+            expected_targets=["lr_ged_sb", "lr_ged_ns"],
+            expected_ensemble="fixture_ensemble",
+        )
+
+
+def test_selection_filters_are_golden_strings():
+    # Pinned to pipeline-core's shipped publisher constants (re-verified 2026-07-20).
+    assert sel.HOP_A_SHARD_FILTERS == {"category": "forecast", "type": "sampled_forecast_shard"}
+    assert sel.HOP_A_MANIFEST_FILTERS == {"category": "forecast", "type": "sampled_forecast_manifest"}
+    assert sel.HOP_A_MANIFEST_NAME_TEMPLATE == "{run_id}__{target}__manifest.json"

views_postprocessing/unfao/track_a_source.py

@@ -105,15 +105,20 @@ def read_shard(shard_bytes: bytes, *, expected_sha256: str) -> tuple[PredictionF
     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``.
+def frames_for_target(
+    manifest: dict, shard_bytes_by_name: dict
+) -> tuple[PredictionFrame, list[dict]]:
+    """A (run, target)'s verified shards → ``(PredictionFrame, headers)``.
 
     ``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.
+    The returned ``headers`` (manifest shard order) carry the producer-minted
+    provenance the sink passes through untouched (§10.2 — nothing is minted
+    downstream).
     """
-    frames, months_seen = [], []
+    frames, months_seen, headers = [], [], []
     for entry in manifest["shards"]:
         name = entry["name"]
         if name not in shard_bytes_by_name:
@@ -133,6 +138,7 @@ def frames_for_target(manifest: dict, shard_bytes_by_name: dict) -> PredictionFr
                 f"{manifest['expected_cell_count']}."
             )
         frames.append(frame)
+        headers.append(header)
         months_seen.append(int(header.get("time_id")))
 
     if sorted(months_seen) != sorted(int(m) for m in manifest["expected_months"]):
@@ -143,4 +149,4 @@ def frames_for_target(manifest: dict, shard_bytes_by_name: dict) -> PredictionFr
     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)
+    return build_prediction_frame(values, time, unit), headers

views_postprocessing/unfao/wire/source_selection.py

@@ -0,0 +1,102 @@
+"""Contract-mode Hop-A inbound: select and assemble the newest manifested run
+(ADR-013 §3.2/§4.2a).
+
+The store is a **port** (DIP): any object with the three-method surface below —
+production passes the manager's ``DatastoreModule``-backed wrapper, tests pass a
+dict-backed fake. This module owns the *selection policy*; byte verification and
+assembly stay in ``track_a_source`` (mechanism), the §6 policy gate stays at the
+sink.
+
+Selection filters are pinned to pipeline-core's shipped publisher constants
+(``sampled_forecast_publisher.py``: ``SHARD_TYPE``/``MANIFEST_TYPE``/
+``FORECAST_CATEGORY``/``MANIFEST_NAME_TEMPLATE`` — re-verified 2026-07-20) and
+golden-string tested here; a drift on either side fails loud in tests, not in
+production.
+
+Await-all-targets (§4.2a): the newest manifest names the run; the run is usable
+only when EVERY declared target has a manifested leg. A missing target refuses
+loud — a partial run is never assembled. Declared identity (§4.2a): each shard
+header's ``provenance.ensemble`` must equal the launch declaration; mismatches
+fail loud, never fall back.
+
+Store port surface::
+
+    latest_file_id(filters: dict) -> str | None
+    file_metadata(file_id: str) -> dict     # carries at least "name"
+    download(file_id: str) -> bytes
+"""
+
+from __future__ import annotations
+
+from views_postprocessing.unfao import track_a_source
+
+# pipeline-core's shipped vocabulary (§3.1/§3.2/§3.3) — golden-string tested.
+HOP_A_SHARD_FILTERS = {"category": "forecast", "type": "sampled_forecast_shard"}
+HOP_A_MANIFEST_FILTERS = {"category": "forecast", "type": "sampled_forecast_manifest"}
+HOP_A_MANIFEST_NAME_TEMPLATE = "{run_id}__{target}__manifest.json"
+
+
+class SourceSelectionError(ValueError):
+    """The store's contents do not satisfy the declared selection."""
+
+
+def fetch_run(store, *, expected_targets, expected_ensemble: str) -> dict:
+    """Assemble the newest fully-manifested run matching the declaration.
+
+    Returns ``{target: (frame, headers)}`` for exactly ``expected_targets``.
+    """
+    newest_id = store.latest_file_id(dict(HOP_A_MANIFEST_FILTERS))
+    if newest_id is None:
+        raise SourceSelectionError(
+            f"no Hop-A manifest in the store (filters={HOP_A_MANIFEST_FILTERS}) — "
+            f"nothing has been published under the contract."
+        )
+    newest = track_a_source.read_manifest(store.download(newest_id))
+    run_id = newest["run_id"]
+
+    per_target: dict = {}
+    for target in expected_targets:
+        manifest = _manifest_for(store, run_id=run_id, target=target, newest=newest)
+        shard_bytes = {}
+        for entry in manifest["shards"]:
+            shard_id = store.latest_file_id(
+                {**HOP_A_SHARD_FILTERS, "name": entry["name"]}
+            )
+            if shard_id is None:
+                raise SourceSelectionError(
+                    f"run {run_id!r}: manifested shard {entry['name']!r} not found in "
+                    f"the store — a torn run must not be assembled."
+                )
+            shard_bytes[entry["name"]] = store.download(shard_id)
+        frame, headers = track_a_source.frames_for_target(manifest, shard_bytes)
+        for header in headers:
+            found = header.get("provenance", {}).get("ensemble")
+            if found != expected_ensemble:
+                raise SourceSelectionError(
+                    f"run {run_id!r}, target {target!r}: header declares ensemble "
+                    f"{found!r}, the delivery was launched for {expected_ensemble!r} — "
+                    f"identity mismatch fails loud, never falls back (§4.2a)."
+                )
+        per_target[target] = (frame, headers)
+    return per_target
+
+
+def _manifest_for(store, *, run_id: str, target: str, newest: dict) -> dict:
+    """The (run, target) manifest — the newest one if it already is that pair."""
+    if newest["target"] == target:
+        return newest
+    name = HOP_A_MANIFEST_NAME_TEMPLATE.format(run_id=run_id, target=target)
+    file_id = store.latest_file_id({**HOP_A_MANIFEST_FILTERS, "name": name})
+    if file_id is None:
+        raise SourceSelectionError(
+            f"run {run_id!r}: no manifest for declared target {target!r} — the run is "
+            f"incomplete for this delivery; awaiting all of its targets (§4.2a)."
+        )
+    manifest = track_a_source.read_manifest(store.download(file_id))
+    if manifest["run_id"] != run_id or manifest["target"] != target:
+        raise SourceSelectionError(
+            f"manifest named {name!r} declares (run={manifest['run_id']!r}, "
+            f"target={manifest['target']!r}) — names are locators, content is identity "
+            f"(§3.3); refusing the mismatch."
+        )
+    return manifest