feat(wire): S7/#112 — composed Hop-B sink + manager contract mode + e2e capstone

tests/test_hop_b_sink_e2e.py

@@ -0,0 +1,164 @@
+"""The epic #105 capstone: fixture Track-A artifacts in → Hop-B artifacts out
+(ADR-013 e2e), through source_selection + sink with fake ports only.
+
+Covers the S7 acceptance criteria: (a) §6 gate before any write, (b) upload order
+shards → sidecar → manifest LAST, (c) §4.1a document fields with the pinned
+consumer name, (d) the §11.4 interlock (default config ⇒ ZERO store calls),
+(e) byte parity with the full fixture set (toolchain-pinned like the other byte
+tests — fails loud on drift, never skips silently).
+"""
+
+import hashlib
+import json
+from pathlib import Path
+
+import numpy as np
+import pyarrow as pa
+import pytest
+
+from views_postprocessing.delivery.draws import DrawsCollapseError
+from views_postprocessing.unfao import product
+from views_postprocessing.unfao.frames import build_prediction_frame
+from views_postprocessing.unfao.wire import sink
+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"
+_GIDS = [100001, 100002, 100003, 100004, 100005, 100006]
+
+
+class FakeSourceStore:
+    def __init__(self):
+        self._records = [
+            ("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 latest_file_id(self, 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 download(self, file_id):
+        return next(b for i, _, b in self._records if i == file_id)
+
+
+class UploadLog:
+    def __init__(self):
+        self.calls = []
+
+    def upload(self, file_path, **kwargs):
+        self.calls.append({"file": Path(file_path).name, **kwargs})
+
+
+def _synthetic_lookup() -> pa.Table:
+    return pa.table(
+        {
+            "priogrid_gid": 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]).dictionary_encode(),
+            "admin1_gaul1_code": pa.array([11, 11, 12, 21, 21, None], pa.int64()),
+            "admin1_gaul1_name": pa.array(["A-one", "A-one", "A-two", "B-one", "B-one", None]).dictionary_encode(),
+            "admin1_gaul0_code": pa.array([1, 1, 1, 2, 2, None], pa.int64()),
+            "admin1_gaul0_name": pa.array(["Aland", "Aland", "Aland", "Bland", "Bland", None]).dictionary_encode(),
+            "admin2_gaul2_code": pa.array([111, 112, 121, 211, 212, None], pa.int64()),
+            "admin2_gaul2_name": pa.array(["A-1-1", "A-1-2", "A-2-1", "B-1-1", "B-1-2", None]).dictionary_encode(),
+        }
+    )
+
+
+def _fetched_run():
+    return sel.fetch_run(
+        FakeSourceStore(),
+        expected_targets=["lr_ged_sb"],
+        expected_ensemble="fixture_ensemble",
+    )
+
+
+def test_interlock_default_config_makes_zero_store_calls(tmp_path):
+    # (d) — the default configuration is provably unable to touch the bucket.
+    log = UploadLog()
+    summary = sink.deliver_run(
+        _fetched_run(), lookup=_synthetic_lookup(), staging_dir=tmp_path, store=log
+    )
+    assert product.UPLOAD_ENABLED is False
+    assert summary["uploaded"] is False
+    assert log.calls == []  # ZERO store calls (§11.4)
+    assert (tmp_path / summary["manifest"]).exists()  # staged locally
+
+
+def test_gate_fires_before_any_write(tmp_path):
+    # (a) — a collapsed payload raises and NOTHING is staged.
+    values = np.zeros((6, 4), dtype=np.float32)  # every row draw-degenerate
+    time = np.full(6, 543, dtype=np.int64)
+    unit = np.array(_GIDS, dtype=np.int64)
+    frame = build_prediction_frame(values, time, unit)
+    _, headers = _fetched_run()["lr_ged_sb"]
+    with pytest.raises(DrawsCollapseError):
+        sink.deliver_run(
+            {"lr_ged_sb": (frame, headers)}, lookup=_synthetic_lookup(), staging_dir=tmp_path
+        )
+    assert list(tmp_path.iterdir()) == []  # gate first: no bytes staged
+
+
+def test_upload_order_and_document_fields(tmp_path):
+    # (b) + (c) — order shards → sidecar → manifest LAST; §4.1a fields on every doc.
+    log = UploadLog()
+    summary = sink.deliver_run(
+        _fetched_run(),
+        lookup=_synthetic_lookup(),
+        staging_dir=tmp_path,
+        store=log,
+        upload_enabled=True,  # explicit declaration (test scope only)
+    )
+    assert summary["uploaded"] is True
+    doc_types = [c["doc_type"] for c in log.calls]
+    assert doc_types == [
+        "sampled_forecast_shard",
+        "sampled_forecast_sidecar",
+        "sampled_forecast_manifest",  # LAST: the commit marker (§4.2)
+    ]
+    for call in log.calls:
+        assert call["name"] == "un_fao"  # the §4.1a pin — the F1 fix
+        assert call["category"] == "forecast" and call["loa"] == "pgm"
+    assert log.calls[0]["targets"] == ["lr_ged_sb"]
+    assert log.calls[-1]["targets"] == ["lr_ged_sb"]
+
+
+def test_ragged_run_refused(tmp_path):
+    run = _fetched_run()
+    frame, headers = run["lr_ged_sb"]
+    other = dict(headers[0])
+    other["run_id"] = "another_run"
+    with pytest.raises(sink.SinkError, match="disagree on run_id"):
+        sink.deliver_run(
+            {"lr_ged_sb": (frame, headers), "lr_ged_ns": (frame, [other])},
+            lookup=_synthetic_lookup(),
+            staging_dir=tmp_path,
+        )
+
+
+def test_e2e_byte_parity_with_the_fixture(tmp_path):
+    # (e) — the anti-corruption proof (§10): fixture Track-A in → fixture Hop-B out,
+    # byte for byte. Toolchain-pinned: fails loud on pyarrow drift.
+    import pyarrow
+
+    assert pyarrow.__version__ == "16.1.0", (
+        f"pinned toolchain violated (fixture README): found {pyarrow.__version__}."
+    )
+    summary = sink.deliver_run(
+        _fetched_run(), lookup=_synthetic_lookup(), staging_dir=tmp_path
+    )
+    for name in (
+        "fixture_run_0__lr_ged_sb__m000543.arrow.parquet",
+        "fixture_run_0__sidecar.parquet",
+        "fixture_run_0__manifest.json",
+    ):
+        assert (tmp_path / name).read_bytes() == (_FIX / name).read_bytes(), name
+    manifest = json.loads((tmp_path / summary["manifest"]).read_text())
+    assert manifest["shards"][0]["sha256"] == hashlib.sha256(
+        (_FIX / "fixture_run_0__lr_ged_sb__m000543.arrow.parquet").read_bytes()
+    ).hexdigest()

views_postprocessing/unfao/managers/unfao.py

@@ -17,9 +17,11 @@
 from datetime import datetime
 import os
 from dotenv import load_dotenv
-from views_postprocessing.unfao.enrichment import GaulLookupEnricher
+from views_postprocessing.unfao.enrichment import _DEFAULT_LOOKUP, GaulLookupEnricher
 from views_postprocessing.unfao.gaul_schema import METADATA_COLS
-from views_postprocessing.unfao import extraction, source_metadata
+from views_postprocessing.unfao import extraction, frame_extraction, 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 pathlib import Path
 
@@ -33,6 +35,36 @@
 LEGACY_FORECAST_FILTERS = {"category": "forecast", "type": "ensemble"}
 
 
+class _ContractStorePort:
+    """Adapts ``DatastoreModule`` to the wire ports (ADR-013 epic #105; DIP —
+    ``wire/source_selection`` and ``wire/sink`` never see Appwrite types)."""
+
+    def __init__(self, datastore: DatastoreModule) -> None:
+        self._dsm = datastore
+
+    def latest_file_id(self, filters: dict):
+        return self._dsm.get_latest_file_id(filters=filters)
+
+    def file_metadata(self, file_id: str) -> dict:
+        return extraction.file_metadata(self._dsm.get_file_metadata(file_id))
+
+    def download(self, file_id: str) -> bytes:
+        return (
+            self._dsm.download_prediction(file_id).to_dict().get("data", {}).get("file_bytes", None)
+        )
+
+    def upload(self, file_path, *, filename, name, doc_type, category, loa, targets) -> None:
+        self._dsm.upload_data(
+            file=file_path,
+            filename=filename,
+            name=name,
+            type=doc_type,
+            category=category,
+            loa=loa,
+            targets=targets,
+        )
+
+
 class UNFAOPostProcessorManager(PostprocessorManager, ForecastingModelManager):
     def __init__(
         self,
@@ -49,6 +81,7 @@ def __init__(
 
         self._historical_dataset = None
         self._forecast_dataset = None
+        self._forecast_run = None  # contract mode: {target: (frame, headers)}
         self._enricher = GaulLookupEnricher()
         self.ensemble_path_manager = None
 
@@ -70,14 +103,14 @@ def _read_historical_data(self):
             source=self._historical_dataframe, targets=self.configs.get("targets")
         )
 
-    def _read_forecast_data(self):
-        # Forecast Data
+    def _prod_forecasts_datastore(self) -> DatastoreModule:
+        """The shared internal store (ADR-013's 'shared shelf'), configured from the
+        declared ensemble's environment. Used by both the legacy and contract reads."""
         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."
             logger.error(err_msg)
             raise ValueError(err_msg)
-
         self.ensemble_path_manager = EnsemblePathManager(ensemble_name_or_path=ensemble_name, validate=False)
         # ensemble_configs = EnsembleManager(
         #     ensemble_path=self.ensemble_path_manager,
@@ -107,9 +140,33 @@ def _read_forecast_data(self):
             database_id=os.getenv("APPWRITE_METADATA_DATABASE_ID"),
             database_name=os.getenv("APPWRITE_METADATA_DATABASE_NAME"),
         )
+        return DatastoreModule(appwrite_file_manager_config=appwrite_config)
+
+    def _read_forecast_data_contract(self):
+        """ADR-013 contract inbound (epic #105): assemble the newest fully-manifested
+        run — declared targets awaited, declared ensemble verified. The `wire/`
+        package owns the policy; this method only adapts the store (DIP)."""
+        port = _ContractStorePort(self._prod_forecasts_datastore())
+        self._forecast_run = source_selection.fetch_run(
+            port,
+            expected_targets=product.TARGETS,
+            expected_ensemble=self.configs["ensemble"],
+        )
+        run_id = next(iter(self._forecast_run.values()))[1][0]["run_id"]
+        logger.info(
+            "Contract inbound: run %s assembled (%d targets).",
+            run_id,
+            len(self._forecast_run),
+        )
 
+    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 = DatastoreModule(appwrite_file_manager_config=appwrite_config)
+            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
@@ -163,6 +220,10 @@ def _transform(
         self,
     ) -> list:
         self._historical_dataframe = self._append_metadata(self._historical_dataset)
+        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:
@@ -180,6 +241,13 @@ def _validate(self) -> pd.DataFrame:
                 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 at read; §6 gate + gid
+            # parity at _save). Coverage still applies, via the frame seam.
+            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()}"
@@ -253,16 +321,25 @@ def _check_coverage(self) -> None:
         region = self.configs.get("region")
         expected = coverage.expected_for(region)
         excluded = coverage.excluded_for(region)
-        for label, df in (
-            ("historical", self._historical_dataframe),
-            ("forecast", self._forecast_dataframe),
-        ):
-            cells = extraction.cells_of(df)
+        if self.configs.get("wire_contract"):
+            # Forecast cells come from the frame seam; targets share one cell set
+            # (enforced again at the sink, §4.2). Historical stays the pandas seam.
+            first_frame = next(iter(self._forecast_run.values()))[0]
+            sources = [
+                ("historical", extraction.cells_of(self._historical_dataframe), len(self._historical_dataframe)),
+                ("forecast", frame_extraction.cells_of(first_frame), first_frame.n_rows),
+            ]
+        else:
+            sources = [
+                ("historical", extraction.cells_of(self._historical_dataframe), len(self._historical_dataframe)),
+                ("forecast", extraction.cells_of(self._forecast_dataframe), len(self._forecast_dataframe)),
+            ]
+        for label, cells, n_rows in sources:
             logger.info(
                 "%s delivery coverage: %d distinct cells, %d rows.",
                 label,
                 len(cells),
-                len(df),
+                n_rows,
             )
             # GAUL-uncovered cells the curated region must drop (S4/C-30) — checked
             # before the count gate so a leaked island names itself, not "over-coverage
@@ -280,7 +357,51 @@ def _check_coverage(self) -> None:
                     region,
                 )
 
+    def _save_contract(self) -> dict:
+        """ADR-013 contract outbound (epic #105): the composed sink delivers the run.
+
+        The §11.4 interlock is enforced by ``wire.sink`` itself: with the default
+        ``product.UPLOAD_ENABLED=False`` (overridable only by the explicit
+        ``wire_upload_enabled`` launch-config key), artifacts are staged locally and
+        ZERO store calls occur. First live enablement is gated on C-161 closure.
+        """
+        import pyarrow.parquet as pq
+
+        upload_enabled = bool(self.configs.get("wire_upload_enabled", product.UPLOAD_ENABLED))
+        store = _ContractStorePort(self._unfao_datastore()) if upload_enabled else None
+        return wire_sink.deliver_run(
+            self._forecast_run,
+            lookup=pq.read_table(_DEFAULT_LOOKUP),
+            staging_dir=Path(self._model_path.data_generated) / "wire_contract",
+            store=store,
+            upload_enabled=upload_enabled,
+        )
+
+    def _unfao_datastore(self) -> DatastoreModule:
+        """The FAO-facing store (`unfao_bucket`)."""
+        return DatastoreModule(appwrite_file_manager_config=self._unfao_appwrite_config())
+
+    def _unfao_appwrite_config(self) -> AppwriteConfig:
+        return 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"),
+        )
+
     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)