diff --git a/tests/test_hop_b_sink_e2e.py b/tests/test_hop_b_sink_e2e.py new file mode 100644 index 0000000..898c056 --- /dev/null +++ b/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() diff --git a/views_postprocessing/unfao/managers/unfao.py b/views_postprocessing/unfao/managers/unfao.py index 194d245..1259310 100644 --- a/views_postprocessing/unfao/managers/unfao.py +++ b/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) diff --git a/views_postprocessing/unfao/wire/sidecar.py b/views_postprocessing/unfao/wire/sidecar.py index 907d515..9c03caf 100644 --- a/views_postprocessing/unfao/wire/sidecar.py +++ b/views_postprocessing/unfao/wire/sidecar.py @@ -73,9 +73,14 @@ def build_sidecar(lookup: pa.Table, gids) -> pa.Table: return pa.table(columns) -def write_sidecar(lookup: pa.Table, gids, *, run_id: str, directory: Path) -> tuple[str, str]: - """Write the run's sidecar; return ``(file_name, sha256_of_bytes)`` for §4.2.""" +def write_table(table: pa.Table, *, run_id: str, directory: Path) -> tuple[str, str]: + """Serialize an already-built (and parity-checked) sidecar table (§5).""" name = sidecar_name(run_id) path = Path(directory) / name - pq.write_table(build_sidecar(lookup, gids), path) + pq.write_table(table, path) return name, hashlib.sha256(path.read_bytes()).hexdigest() + + +def write_sidecar(lookup: pa.Table, gids, *, run_id: str, directory: Path) -> tuple[str, str]: + """Build + write the run's sidecar; return ``(file_name, sha256_of_bytes)``.""" + return write_table(build_sidecar(lookup, gids), run_id=run_id, directory=directory) diff --git a/views_postprocessing/unfao/wire/sink.py b/views_postprocessing/unfao/wire/sink.py new file mode 100644 index 0000000..019fdd7 --- /dev/null +++ b/views_postprocessing/unfao/wire/sink.py @@ -0,0 +1,191 @@ +"""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). + +Upload store-port surface:: + + upload(file_path, *, filename, name, doc_type, category, loa, targets) -> None +""" + +from __future__ import annotations + +import logging +from pathlib import Path + +from views_postprocessing.delivery.draws import assert_draws_uncollapsed +from views_postprocessing.delivery.parity import assert_gid_set_parity +from views_postprocessing.unfao import product +from views_postprocessing.unfao.frame_extraction import cells_of, month_slice +from views_postprocessing.unfao.wire.naming import run_manifest_name +from views_postprocessing.unfao.wire.run_manifest import build_run_manifest +from views_postprocessing.unfao.wire.shard import write_shard +from views_postprocessing.unfao.wire.sidecar import build_sidecar, write_table + +logger = logging.getLogger(__name__) + +SHARD_DOC_TYPE = "sampled_forecast_shard" +MANIFEST_DOC_TYPE = "sampled_forecast_manifest" +SIDECAR_DOC_TYPE = "sampled_forecast_sidecar" + + +class SinkError(ValueError): + """The assembled run cannot be delivered as declared.""" + + +def deliver_run( + per_target: dict, + *, + lookup, + staging_dir: Path, + store=None, + upload_enabled: bool = product.UPLOAD_ENABLED, + consumer_name: str = product.CONSUMER_DOCUMENT_NAME, + s_min: int = product.S_MIN, +) -> 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). + """ + 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(): + assert_draws_uncollapsed( + 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(): + 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} + ) + + # 3. 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. + 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, + sidecar_record=sidecar_record, + ) + manifest_file = run_manifest_name(run_id) + (staging / manifest_file).write_bytes(manifest_bytes) + + summary = { + "run_id": run_id, + "targets": list(per_target), + "shards": shard_records, + "sidecar": sidecar_record, + "manifest": manifest_file, + "staging_dir": str(staging), + "uploaded": False, + } + + # 5+6. Upload — behind the §11.4 interlock. + if not upload_enabled: + logger.info( + "ADR-013 upload interlock holding (upload_enabled=False): run %s staged " + "at %s, ZERO store calls. First live enablement is gated on C-161.", + run_id, + staging, + ) + return summary + if store is None: + 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 _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())))