feat(wire): A1 — streaming per-target delivery (run-0 OOM fix)

tests/test_hop_b_sink_e2e.py

@@ -1,11 +1,13 @@
-"""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).
+"""The epic #105 capstone, streaming edition: fixture Track-A artifacts in →
+Hop-B artifacts out (ADR-013 e2e), through resolve_run leases + the streaming
+sink with fake ports only.
+
+Covers the S7 acceptance criteria under the OOM-fix design: (a) §6 gate before
+any of a target's bytes are staged, (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 — fails loud on drift, never skips silently),
+plus the streaming-specific ragged-run refusals.
 """
 
 import hashlib
@@ -45,6 +47,17 @@ def download(self, file_id):
         return next(b for i, _, b in self._records if i == file_id)
 
 
+class FakeLease:
+    """Trivial lease: preloaded values, for failure-path tests."""
+
+    def __init__(self, run_id, frame, headers):
+        self.run_id = run_id
+        self._value = (frame, headers)
+
+    def load(self):
+        return self._value
+
+
 class UploadLog:
     def __init__(self):
         self.calls = []
@@ -70,45 +83,53 @@ def _synthetic_lookup() -> pa.Table:
     )
 
 
-def _fetched_run():
-    return sel.fetch_run(
+def _leases():
+    """Real leases through the real resolver — the full inbound chain."""
+    return sel.resolve_run(
         FakeSourceStore(),
         expected_targets=["lr_ged_sb"],
         expected_ensemble="fixture_ensemble",
     )
 
 
+def _fixture_frame_and_headers():
+    return _leases()["lr_ged_sb"].load()
+
+
 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
+        _leases(), 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
+    assert (Path(summary["staging_dir"]) / summary["manifest"]).exists()
+    assert Path(summary["staging_dir"]).name == "fixture_run_0"  # per-run_id subdir
 
 
 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"]
+    frame = build_prediction_frame(
+        values, np.full(6, 543, dtype=np.int64), np.array(_GIDS, dtype=np.int64)
+    )
+    _, headers = _fixture_frame_and_headers()
     with pytest.raises(DrawsCollapseError):
         sink.deliver_run(
-            {"lr_ged_sb": (frame, headers)}, lookup=_synthetic_lookup(), staging_dir=tmp_path
+            {"lr_ged_sb": FakeLease("fixture_run_0", frame, headers)},
+            lookup=_synthetic_lookup(),
+            staging_dir=tmp_path,
         )
-    assert list(tmp_path.iterdir()) == []  # gate first: no bytes staged
+    assert list(tmp_path.iterdir()) == []  # gate first: no bytes staged, no subdir
 
 
 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(),
+        _leases(),
         lookup=_synthetic_lookup(),
         staging_dir=tmp_path,
         store=log,
@@ -128,37 +149,59 @@ def test_upload_order_and_document_fields(tmp_path):
     assert log.calls[-1]["targets"] == ["lr_ged_sb"]
 
 
-def test_ragged_run_refused(tmp_path):
-    run = _fetched_run()
-    frame, headers = run["lr_ged_sb"]
+def test_ragged_run_refused_and_first_target_released(tmp_path):
+    frame, headers = _fixture_frame_and_headers()
     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])},
+            {
+                "lr_ged_sb": FakeLease("fixture_run_0", frame, headers),
+                "lr_ged_ns": FakeLease("fixture_run_0", frame, [other]),
+            },
+            lookup=_synthetic_lookup(),
+            staging_dir=tmp_path,
+        )
+    # first target's shards were staged (harmless: no manifest = invisible, §4.2)
+    staged = list((tmp_path / "fixture_run_0").iterdir())
+    assert all("manifest" not in p.name for p in staged)
+
+
+def test_targets_disagreeing_on_cells_refused(tmp_path):
+    frame, headers = _fixture_frame_and_headers()
+    small = build_prediction_frame(
+        frame.values[:5], np.asarray(frame.index.time)[:5], np.asarray(frame.index.unit)[:5]
+    )
+    with pytest.raises(sink.SinkError, match="cell set"):
+        sink.deliver_run(
+            {
+                "lr_ged_sb": FakeLease("fixture_run_0", frame, headers),
+                "lr_ged_ns": FakeLease("fixture_run_0", small, headers),
+            },
             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.
+    # byte for byte, through resolve → lease → streamed sink. Toolchain-pinned.
     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
+        _leases(), lookup=_synthetic_lookup(), staging_dir=tmp_path
     )
+    staging = Path(summary["staging_dir"])
     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 (staging / name).read_bytes() == (_FIX / name).read_bytes(), name
+    manifest = json.loads((staging / summary["manifest"]).read_text())
     assert manifest["shards"][0]["sha256"] == hashlib.sha256(
         (_FIX / "fixture_run_0__lr_ged_sb__m000543.arrow.parquet").read_bytes()
     ).hexdigest()

tests/test_wire_source_selection.py

@@ -1,5 +1,6 @@
-"""Contract-mode Hop-A inbound (wire/source_selection.py) — dict-backed fake store
-serving the golden fixture; selection policy, await-all-targets, declared identity."""
+"""Contract-mode Hop-A inbound (wire/source_selection.py) — resolve + lease model:
+dict-backed fake store serving the golden fixture; cheap resolution with pinned
+file_ids, lazy verified+curated loading, loud refusals."""
 
 import json
 from pathlib import Path
@@ -14,114 +15,129 @@
 
 
 class FakeStore:
-    """The three-method store port over a dict of (metadata, bytes) records."""
+    """The two-method source port over (file_id, metadata, payload) records."""
 
     def __init__(self, records):
-        # records: list of (file_id, metadata_dict, payload_bytes); newest LAST.
-        self._records = records
-        self.calls = []
+        self._records = records  # newest LAST
+        self.downloads = []
 
     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):
+        self.downloads.append(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(),
-            ),
+            ("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",
+def _resolve(store=None, **overrides):
+    kwargs = dict(
+        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"
+    kwargs.update(overrides)
+    return sel.resolve_run(store or _fixture_store(), **kwargs)
+
+
+# --- resolution: cheap, pinned, complete -------------------------------------------
+
+
+def test_resolve_is_cheap_and_pins_file_ids():
+    store = _fixture_store()
+    leases = sel.resolve_run(
+        store, expected_targets=["lr_ged_sb"], expected_ensemble="fixture_ensemble"
+    )
+    lease = leases["lr_ged_sb"]
+    assert lease.run_id == "fixture_run_0"
+    assert lease.shard_file_ids == {_SHARD_NAME: "f-shard"}
+    assert store.downloads == ["f-manifest"]  # manifests only — no shard bytes moved
 
 
 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"
-        )
+        _resolve(FakeStore([]))
 
 
-def test_missing_declared_target_refuses_loud():
-    # The store has lr_ged_sb only; the delivery declares two targets (§4.2a).
+def test_missing_declared_target_refuses_at_resolve():
     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",
-        )
+        _resolve(expected_targets=["lr_ged_sb", "lr_ged_ns"])
 
 
-def test_manifested_shard_missing_from_store_refuses_loud():
+def test_manifested_shard_missing_refuses_at_resolve():
     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"
-        )
+        _resolve(store)
 
 
 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},
+            {**sel.HOP_A_MANIFEST_FILTERS, "name": "fixture_run_0__lr_ged_ns__manifest.json"},
             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",
-        )
+        _resolve(store, expected_targets=["lr_ged_sb", "lr_ged_ns"])
+
+
+# --- lease loading: verified, identity-checked, curated ----------------------------
+
+
+def test_lease_load_assembles_the_fixture():
+    frame, headers = _resolve()["lr_ged_sb"].load()
+    assert frame.n_rows == 6 and frame.sample_count == 4
+    assert headers[0]["run_id"] == "fixture_run_0"
+
+
+def test_lease_load_downloads_by_pinned_id_only():
+    store = _fixture_store()
+    leases = sel.resolve_run(
+        store, expected_targets=["lr_ged_sb"], expected_ensemble="fixture_ensemble"
+    )
+    store.downloads.clear()
+    leases["lr_ged_sb"].load()
+    assert store.downloads == ["f-shard"]  # by id — never re-filtered
+
+
+def test_wrong_declared_ensemble_refuses_at_load():
+    leases = _resolve(expected_ensemble="rusty_bucket")
+    with pytest.raises(sel.SourceSelectionError, match="rusty_bucket.*never falls back"):
+        leases["lr_ged_sb"].load()
+
+
+def test_lease_applies_declared_curation():
+    leases = _resolve(excluded_gids=frozenset({100006}))
+    frame, _ = leases["lr_ged_sb"].load()
+    assert frame.n_rows == 5  # the declared exclusion is gone from the product frame
+    from views_postprocessing.unfao.frame_extraction import cells_of
+
+    assert 100006 not in cells_of(frame)
+
+
+def test_lease_coverage_expectation_fails_loud():
+    from views_postprocessing.delivery.coverage import CoverageError
+
+    leases = _resolve(expected_cells=7)  # fixture has 6
+    with pytest.raises(CoverageError):
+        leases["lr_ged_sb"].load()
 
 
 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"