diff --git a/tests/test_hop_b_sink_e2e.py b/tests/test_hop_b_sink_e2e.py
index 898c056..22f8203 100644
--- a/tests/test_hop_b_sink_e2e.py
+++ b/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,14 +149,35 @@ 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,
         )
@@ -143,22 +185,23 @@ def test_ragged_run_refused(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()
diff --git a/tests/test_wire_source_selection.py b/tests/test_wire_source_selection.py
index 9545881..6800a63 100644
--- a/tests/test_wire_source_selection.py
+++ b/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"
diff --git a/views_postprocessing/unfao/managers/unfao.py b/views_postprocessing/unfao/managers/unfao.py
index 42ad944..c75db4b 100644
--- a/views_postprocessing/unfao/managers/unfao.py
+++ b/views_postprocessing/unfao/managers/unfao.py
@@ -19,7 +19,7 @@
 from dotenv import load_dotenv
 from views_postprocessing.unfao.enrichment import _DEFAULT_LOOKUP, GaulLookupEnricher
 from views_postprocessing.unfao.gaul_schema import METADATA_COLS
-from views_postprocessing.unfao import extraction, frame_extraction, product, source_metadata
+from views_postprocessing.unfao import 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
@@ -81,7 +81,7 @@ def __init__(
 
         self._historical_dataset = None
         self._forecast_dataset = None
-        self._forecast_run = None  # contract mode: {target: (frame, headers)}
+        self._forecast_resolution = None  # contract mode: {target: TargetLease}
         self._enricher = GaulLookupEnricher()
         self.ensemble_path_manager = None
 
@@ -103,9 +103,19 @@ def _read_historical_data(self):
             source=self._historical_dataframe, targets=self.configs.get("targets")
         )
 
-    def _prod_forecasts_datastore(self) -> DatastoreModule:
+    def _prod_forecasts_datastore(self, *, name_scoped: bool = True) -> 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."""
+        declared ensemble's environment. Used by both the legacy and contract reads.
+
+        ``name_scoped`` controls pipeline-core's automatic ``name == model_name`` query
+        filter (``DatastoreModule.get_predictions_by_metadata`` injects it on every
+        lookup). The LEGACY read depends on it: its filters ``{category: forecast,
+        type: ensemble}`` carry no name, so the injected ensemble-name is the S3/C-25
+        identity guard that prevents selecting a *different* ensemble's forecast. The
+        CONTRACT read must switch it OFF: ADR-013 files are named
+        ``{run_id}__{target}__manifest.json`` (never the bare ensemble name), so an
+        injected ``name == "rusty_bucket"`` matches nothing and also clobbers the wire
+        layer's own run-id / target / name filters."""
         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."
@@ -140,38 +150,38 @@ def _prod_forecasts_datastore(self) -> DatastoreModule:
             database_id=os.getenv("APPWRITE_METADATA_DATABASE_ID"),
             database_name=os.getenv("APPWRITE_METADATA_DATABASE_NAME"),
         )
-        return DatastoreModule(appwrite_file_manager_config=appwrite_config)
+        datastore = DatastoreModule(appwrite_file_manager_config=appwrite_config)
+        if not name_scoped:
+            # Suppress the automatic name==model_name filter (see docstring). model_path
+            # is used by DatastoreModule only for that injection and for uploads; the
+            # contract read neither uploads nor performs any model-scoped lookup.
+            datastore.model_path = None
+        return datastore
 
     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(
+        """ADR-013 contract inbound (epic #105; streaming since the run-0 OOM fix):
+        RESOLVE the newest fully-manifested run — manifests + pinned shard
+        file_ids only, no heavy bytes. Frames materialize one target at a time
+        inside the sink at _save (each lease loads → verifies → curates → is
+        released). The `wire/` package owns the policy; this method only adapts
+        the store (DIP) and declares the product facts (region curation +
+        coverage expectations live in the lease, where frames exist)."""
+        port = _ContractStorePort(self._prod_forecasts_datastore(name_scoped=False))
+        region = self.configs.get("region")
+        self._forecast_resolution = source_selection.resolve_run(
             port,
             expected_targets=product.TARGETS,
             expected_ensemble=self.configs["ensemble"],
+            excluded_gids=coverage.excluded_for(region),
+            expected_cells=coverage.expected_for(region),
         )
-        # Declared-region curation (C-30/S4, at the anti-corruption layer): the
-        # producer publishes its full model grid; the FAO product excludes the
-        # declared GAUL-uncovered cells. Explicit frozenset, never inference.
-        excluded = coverage.excluded_for(self.configs.get("region"))
-        if excluded:
-            self._forecast_run = {
-                target: (frame_extraction.drop_units(frame, excluded), headers)
-                for target, (frame, headers) in self._forecast_run.items()
-            }
-            logger.info(
-                "Declared-region curation applied: %d excluded cells dropped per "
-                "target (region=%r).",
-                len(excluded),
-                self.configs.get("region"),
-            )
-        run_id = next(iter(self._forecast_run.values()))[1][0]["run_id"]
+        run_id = next(iter(self._forecast_resolution.values())).run_id
         logger.info(
-            "Contract inbound: run %s assembled (%d targets).",
+            "Contract inbound resolved: run %s (%d targets leased; region=%r; "
+            "heavy fetch deferred to delivery).",
             run_id,
-            len(self._forecast_run),
+            len(self._forecast_resolution),
+            region,
         )
 
     def _read_forecast_data(self):
@@ -258,8 +268,14 @@ def _validate(self) -> pd.DataFrame:
 
         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.
+            # verified chain (hashes + header/payload asserts + coverage inside each
+            # lease's load; §6 gate + gid parity at _save). This phase asserts the
+            # RESOLUTION happened — the Template Method stays truthful: _read
+            # resolves, _save materializes.
+            if self._forecast_resolution is None:
+                raise ValueError(
+                    "wire_contract mode: no resolved run — _read did not resolve."
+                )
             self._check_coverage()
             return
 
@@ -337,12 +353,11 @@ def _check_coverage(self) -> None:
         expected = coverage.expected_for(region)
         excluded = coverage.excluded_for(region)
         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]
+            # Streaming mode: forecast coverage is asserted inside each lease's
+            # load() (where the frame exists — see wire/source_selection); only
+            # the historical (still pandas until #126) is checked here.
             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 = [
@@ -382,10 +397,14 @@ def _save_contract(self) -> dict:
         """
         import pyarrow.parquet as pq
 
+        if self._forecast_resolution is None:
+            raise ValueError(
+                "contract _save called without a resolved run — _read must run first."
+            )
         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,
+            self._forecast_resolution,
             lookup=pq.read_table(_DEFAULT_LOOKUP),
             staging_dir=Path(self._model_path.data_generated) / "wire_contract",
             store=store,
diff --git a/views_postprocessing/unfao/wire/sink.py b/views_postprocessing/unfao/wire/sink.py
index 019fdd7..e6d5c9d 100644
--- a/views_postprocessing/unfao/wire/sink.py
+++ b/views_postprocessing/unfao/wire/sink.py
@@ -1,23 +1,29 @@
-"""The Hop-B sink orchestrator (ADR-013 §4, §5, §6, §11.4 — epic #105).
-
-Composes the wire pieces in the contract's order, with the store injected as a
-port (DIP) and every value passed through from declarations or Hop-A headers —
-the sink mints nothing (§10.2):
-
-1. **§6 policy gate first** — ``delivery.draws.assert_draws_uncollapsed`` per
-   target, before a single byte is staged.
-2. Per-(target, month) arrow shards (§4.1), re-embedding the producer's own
-   headers (one header, both envelopes).
-3. Sidecar built, **parity-checked** (§5.2), then staged.
-4. The run manifest staged LAST-in-spirit and uploaded LAST-in-fact (§4.2 —
-   after every shard AND the sidecar: the commit marker).
-5. Every store document carries the §4.1a fields with the pinned consumer
-   ``name`` — the fix for the F1 invisibility.
-6. **The §11.4 interlock**: ``upload_enabled`` defaults to
-   ``product.UPLOAD_ENABLED`` (False). Disabled means artifacts are staged
-   locally and logged — ZERO store calls; enabling requires an explicit
-   declaration at launch, and the first live enablement is gated on faoapi's
-   C-161 closure (outside this epic).
+"""The Hop-B sink orchestrator (ADR-013 §4, §5, §6, §11.4 — epic #105; streaming
+per-target since the run-0 OOM fix, 2026-07-27).
+
+Composes the wire in the contract's order with the store injected as a port
+(DIP) and every value passed through from declarations or Hop-A headers — the
+sink mints nothing (§10.2). Memory-bounded by construction: targets are
+**leases** (anything with ``.run_id`` and ``.load() -> (frame, headers)``), and
+each target's frame lives only for its own gate + shard-writing, then is
+released. Peak resident data ≈ one target, never the whole run.
+
+Per target: load (verified, curated by the lease) → **§6 policy gate** → month
+shards → release. Run-level facts (run_id, month set, cell set) are checked
+incrementally against the first target — a ragged run refuses loud (§4.2).
+Then: sidecar built + parity-checked (§5.2) → run manifest **uploaded LAST**
+(§4.2, the commit marker; the sidecar is inside the commit) → every document
+under the §4.1a fields with the pinned consumer name. Staging goes into a
+per-run_id subdirectory; each upload is ledger-logged.
+
+**The §11.4 interlock:** ``upload_enabled`` defaults to
+``product.UPLOAD_ENABLED`` (False) — the default configuration stages locally
+and makes ZERO store calls; enabling requires an explicit launch declaration,
+first live enablement gated on faoapi's C-161 closure.
+
+Note (pipeline-core C-217): manifest bytes are NOT hash-stable through the
+store (SDK JSON re-serialization) — never hash-verify manifest bytes; only
+shard/sidecar binaries are hash-stable.
 
 Upload store-port surface::
 
@@ -61,56 +67,71 @@ def deliver_run(
 ) -> dict:
     """Deliver one run to `unfao_bucket` (or stage it, when the interlock holds).
 
-    ``per_target`` is ``source_selection.fetch_run``'s output:
-    ``{target: (frame, headers)}``. Returns a summary dict (run_id, records,
-    ``uploaded`` flag, staging paths).
+    ``per_target`` maps target → lease (``.run_id``, ``.load()``); production
+    leases come from ``source_selection.resolve_run``, tests pass trivial ones.
+    Returns a summary dict (run_id, records, ``uploaded`` flag, staging path).
     """
     if not per_target:
         raise SinkError("empty run: no targets to deliver.")
-    staging = Path(staging_dir)
-    staging.mkdir(parents=True, exist_ok=True)
 
-    # 1. §6 policy gate — before any write.
-    for target, (frame, headers) in per_target.items():
+    reference: tuple | None = None  # (run_id, months_tuple, gids) from target #1
+    staging: Path | None = None
+    shard_records: list = []
+
+    for target, lease in per_target.items():
+        frame, headers = lease.load()
+
+        # 1. §6 policy gate — before any of this target's bytes are staged.
         assert_draws_uncollapsed(
-            frame.values,
-            headers[0]["sample_count"],
-            s_min=s_min,
-            label=f"forecast[{target}]",
+            frame.values, headers[0]["sample_count"], s_min=s_min, label=f"forecast[{target}]"
         )
 
-    # Run-level facts, cross-checked across targets (§4.2: ONE months set, ONE
-    # per-shard cell count, ONE run id for the whole run).
-    run_id = _single_value(per_target, lambda h: h["run_id"], "run_id")
-    expected_months = sorted(
-        _single_value(per_target, lambda h: h["time_id"], "months", collect=True)
-    )
-    gids = _single_gid_set(per_target)
-    expected_cell_count = len(gids)
-
-    # 2. Shards (§4.1) — the producer's headers re-embedded untouched.
-    shard_records = []
-    for target, (frame, headers) in per_target.items():
+        # 2. Run-level facts, checked incrementally (§4.2: ONE run_id, ONE month
+        #    set, ONE cell set across the whole run — ragged runs are malformed).
+        run_ids = {h["run_id"] for h in headers}
+        if len(run_ids) != 1:
+            raise SinkError(f"target {target!r} headers disagree on run_id: {sorted(run_ids)}.")
+        facts = (
+            next(iter(run_ids)),
+            tuple(sorted(int(h["time_id"]) for h in headers)),
+            frozenset(cells_of(frame)),
+        )
+        if reference is None:
+            reference = facts
+            staging = Path(staging_dir) / facts[0]  # per-run_id subdir
+            staging.mkdir(parents=True, exist_ok=True)
+        else:
+            for name, ours, theirs in zip(("run_id", "months", "cell set"), reference, facts):
+                if ours != theirs:
+                    detail = f"{len(ours)} vs {len(theirs)} cells" if name == "cell set" else f"{ours!r} vs {theirs!r}"
+                    raise SinkError(
+                        f"targets disagree on {name} ({detail}) — a ragged run is malformed (§4.2)."
+                    )
+
+        # 3. This target's shards (§4.1) — producer headers re-embedded untouched.
         for header in headers:
             values, time, unit = month_slice(frame, header["time_id"])
             name, sha = write_shard(values, time, unit, header=header, directory=staging)
             shard_records.append(
                 {"name": name, "target": target, "time_id": header["time_id"], "sha256": sha}
             )
+        del frame  # release before the next target loads (the OOM fix)
+
+    run_id, months, gids = reference
 
-    # 3. Sidecar (§5) — built, parity-checked, staged.
+    # 4. Sidecar (§5) — built, parity-checked, staged.
     table = build_sidecar(lookup, gids)
     assert_gid_set_parity(gids, table.column("priogrid_id").to_pylist())
     sidecar_file, sidecar_sha = write_table(table, run_id=run_id, directory=staging)
     sidecar_record = {"name": sidecar_file, "sha256": sidecar_sha}
 
-    # 4. Run manifest (§4.2) — staged; uploaded LAST below.
+    # 5. Run manifest (§4.2) — staged; uploaded LAST below.
     manifest_bytes = build_run_manifest(
         run_id=run_id,
         targets=list(per_target),
         shard_records=shard_records,
-        expected_months=expected_months,
-        expected_cell_count=expected_cell_count,
+        expected_months=list(months),
+        expected_cell_count=len(gids),
         sidecar_record=sidecar_record,
     )
     manifest_file = run_manifest_name(run_id)
@@ -126,7 +147,7 @@ def deliver_run(
         "uploaded": False,
     }
 
-    # 5+6. Upload — behind the §11.4 interlock.
+    # 6. Upload — behind the §11.4 interlock; shards → sidecar → manifest LAST.
     if not upload_enabled:
         logger.info(
             "ADR-013 upload interlock holding (upload_enabled=False): run %s staged "
@@ -139,53 +160,14 @@ def deliver_run(
         raise SinkError("upload enabled but no store provided — refusing to guess.")
 
     common = {"name": consumer_name, "category": "forecast", "loa": "pgm"}
-    for record in shard_records:  # shards first …
-        store.upload(
-            staging / record["name"],
-            filename=record["name"],
-            doc_type=SHARD_DOC_TYPE,
-            targets=[record["target"]],
-            **common,
-        )
-    store.upload(  # … sidecar next …
-        staging / sidecar_file,
-        filename=sidecar_file,
-        doc_type=SIDECAR_DOC_TYPE,
-        targets=list(per_target),
-        **common,
-    )
-    store.upload(  # … manifest LAST: the commit marker (§4.2).
-        staging / manifest_file,
-        filename=manifest_file,
-        doc_type=MANIFEST_DOC_TYPE,
-        targets=list(per_target),
-        **common,
-    )
-    summary["uploaded"] = True
-    return summary
 
+    def _upload(file_name: str, doc_type: str, targets: list) -> None:
+        store.upload(staging / file_name, filename=file_name, doc_type=doc_type, targets=targets, **common)
+        logger.info("uploaded %s (type=%s, run=%s)", file_name, doc_type, run_id)  # the ledger
 
-def _single_value(per_target: dict, pick, what: str, *, collect: bool = False):
-    """One agreed value across every header of every target — or fail loud."""
-    per_target_values = {
-        target: sorted({pick(h) for h in headers}) if collect else {pick(h) for h in headers}
-        for target, (_, headers) in per_target.items()
-    }
-    distinct = {tuple(v) if isinstance(v, list) else tuple(sorted(v)) for v in per_target_values.values()}
-    if len(distinct) != 1:
-        raise SinkError(f"targets disagree on {what}: {per_target_values} — a ragged run is malformed (§4.2).")
-    value = next(iter(per_target_values.values()))
-    if collect:
-        return value
-    if len(value) != 1:
-        raise SinkError(f"multiple {what} values within one target: {value}.")
-    return next(iter(value))
-
-
-def _single_gid_set(per_target: dict) -> set[int]:
-    """One agreed cell set across targets (§4.2/§5.2) — or fail loud."""
-    sets = {target: frozenset(cells_of(frame)) for target, (frame, _) in per_target.items()}
-    if len(set(sets.values())) != 1:
-        sizes = {t: len(s) for t, s in sets.items()}
-        raise SinkError(f"targets disagree on the cell set (sizes: {sizes}) — ragged run (§4.2).")
-    return set(next(iter(sets.values())))
+    for record in shard_records:
+        _upload(record["name"], SHARD_DOC_TYPE, [record["target"]])
+    _upload(sidecar_file, SIDECAR_DOC_TYPE, list(per_target))
+    _upload(manifest_file, MANIFEST_DOC_TYPE, list(per_target))  # the commit marker
+    summary["uploaded"] = True
+    return summary
diff --git a/views_postprocessing/unfao/wire/source_selection.py b/views_postprocessing/unfao/wire/source_selection.py
index 67c5487..f96ea7c 100644
--- a/views_postprocessing/unfao/wire/source_selection.py
+++ b/views_postprocessing/unfao/wire/source_selection.py
@@ -1,34 +1,39 @@
-"""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.
-
+"""Contract-mode Hop-A inbound: resolve the newest manifested run cheaply, then
+lease each target for lazy, memory-bounded loading (ADR-013 §3.2/§4.2a; the
+run-0 OOM fix, 2026-07-27 expert review amendments).
+
+Two phases, deliberately split:
+
+- ``resolve_run`` — CHEAP. Downloads only manifests: newest manifest names the
+  run; every declared target must have a content-verified (run, target) manifest
+  (§4.2a await-all-targets); every listed shard's store **file_id is pinned here**
+  so later fetches are race-free — a newer run published mid-delivery cannot be
+  mixed in (fetch-by-id, never re-filtered).
+- ``TargetLease.load()`` — HEAVY, one target at a time. Downloads the pinned
+  shards, delegates byte/manifest verification to ``track_a_source``, verifies
+  the **declared identity** (every header's ``provenance.ensemble`` equals the
+  launch declaration — fails loud, never falls back), then returns the *product*
+  frame: declared-region curation applied (``drop_units``) and the coverage
+  invariants asserted. Callers downstream never see raw producer frames.
+
+The store is a **port** (DIP): any object with ``latest_file_id(filters)`` /
+``download(file_id)``; production adapts ``DatastoreModule``, tests use a dict.
 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
+(re-verified 2026-07-20) and golden-string tested.
+
+Note (pipeline-core C-217): manifest bytes are NOT hash-stable through the store
+— the Appwrite SDK auto-parses JSON payloads and they are re-serialized in
+compact form on download. Only shard/sidecar (binary) bytes are hash-stable.
+**Never hash-verify manifest bytes.**
 """
 
 from __future__ import annotations
 
+from dataclasses import dataclass, field
+
+from views_postprocessing.delivery import coverage
 from views_postprocessing.unfao import track_a_source
+from views_postprocessing.unfao.frame_extraction import cells_of, drop_units
 
 # pipeline-core's shipped vocabulary (§3.1/§3.2/§3.3) — golden-string tested.
 HOP_A_SHARD_FILTERS = {"category": "forecast", "type": "sampled_forecast_shard"}
@@ -40,10 +45,63 @@ 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``.
+@dataclass
+class TargetLease:
+    """One target's claim on a resolved run: manifest + pinned shard file_ids +
+    everything needed to load, verify, and curate it later — without re-querying
+    the store's 'latest' state."""
+
+    target: str
+    manifest: dict
+    shard_file_ids: dict  # shard name -> pinned store file_id
+    store: object
+    expected_ensemble: str
+    excluded_gids: frozenset = field(default_factory=frozenset)
+    expected_cells: int | None = None
+
+    @property
+    def run_id(self) -> str:
+        return self.manifest["run_id"]
+
+    def load(self):
+        """Fetch (by pinned id), verify, curate — return the PRODUCT ``(frame, headers)``."""
+        shard_bytes = {
+            name: self.store.download(file_id)
+            for name, file_id in self.shard_file_ids.items()
+        }
+        frame, headers = track_a_source.frames_for_target(self.manifest, shard_bytes)
+        for header in headers:
+            found = header.get("provenance", {}).get("ensemble")
+            if found != self.expected_ensemble:
+                raise SourceSelectionError(
+                    f"run {self.run_id!r}, target {self.target!r}: header declares "
+                    f"ensemble {found!r}, the delivery was launched for "
+                    f"{self.expected_ensemble!r} — identity mismatch fails loud, "
+                    f"never falls back (§4.2a)."
+                )
+        # Declared-region curation (C-30): the producer publishes its full model
+        # grid; the product excludes the declared GAUL-uncovered cells.
+        frame = drop_units(frame, self.excluded_gids)
+        cells = cells_of(frame)
+        if self.excluded_gids:
+            coverage.assert_no_excluded_cells(cells, self.excluded_gids, label=f"forecast[{self.target}]")
+        if self.expected_cells is not None:
+            coverage.assert_complete_coverage(cells, self.expected_cells, label=f"forecast[{self.target}]")
+        return frame, headers
+
+
+def resolve_run(
+    store,
+    *,
+    expected_targets,
+    expected_ensemble: str,
+    excluded_gids=frozenset(),
+    expected_cells: int | None = None,
+) -> dict:
+    """Resolve the newest fully-manifested run — manifests and file_ids only.
+
+    Returns ``{target: TargetLease}`` covering exactly ``expected_targets``.
+    Heavy bytes move only when a lease's ``load()`` is called.
     """
     newest_id = store.latest_file_id(dict(HOP_A_MANIFEST_FILTERS))
     if newest_id is None:
@@ -54,31 +112,28 @@ def fetch_run(store, *, expected_targets, expected_ensemble: str) -> dict:
     newest = track_a_source.read_manifest(store.download(newest_id))
     run_id = newest["run_id"]
 
-    per_target: dict = {}
+    leases: dict = {}
     for target in expected_targets:
         manifest = _manifest_for(store, run_id=run_id, target=target, newest=newest)
-        shard_bytes = {}
+        shard_file_ids = {}
         for entry in manifest["shards"]:
-            shard_id = store.latest_file_id(
-                {**HOP_A_SHARD_FILTERS, "name": entry["name"]}
-            )
+            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
+            shard_file_ids[entry["name"]] = shard_id
+        leases[target] = TargetLease(
+            target=target,
+            manifest=manifest,
+            shard_file_ids=shard_file_ids,
+            store=store,
+            expected_ensemble=expected_ensemble,
+            excluded_gids=frozenset(excluded_gids),
+            expected_cells=expected_cells,
+        )
+    return leases
 
 
 def _manifest_for(store, *, run_id: str, target: str, newest: dict) -> dict: