diff --git a/tests/test_aggregate_calibration.py b/tests/test_aggregate_calibration.py new file mode 100644 index 00000000..bc90f667 --- /dev/null +++ b/tests/test_aggregate_calibration.py @@ -0,0 +1,139 @@ +"""Guards for the aggregate-calibration tool. + +Built against synthetic posteriors whose answer is known by construction: one where cells are +independent across draws, one where every draw is a coherent joint scenario, and one where the +predictive systematically under-shoots the observed totals. A diagnostic that cannot tell those +three apart is not a diagnostic. +""" + +import json +import subprocess +import sys +from pathlib import Path + +import numpy as np + +REPO = Path(__file__).resolve().parents[1] + +N_CELLS, N_MONTHS, N_DRAWS = 25, 2, 128 + + +def _write(tmp_path: Path, draws: np.ndarray) -> Path: + root = tmp_path / "predictions_forecasting_20260930_000000" / "lr_sb_best" + root.mkdir(parents=True) + n = draws.shape[0] + np.save(root / "y_pred.npy", draws.astype(np.float32)) + np.savez( + root / "identifiers.npz", + time=np.repeat(np.arange(561, 561 + N_MONTHS), n // N_MONTHS).astype(np.int64), + unit=np.tile(np.arange(1000, 1000 + n // N_MONTHS), N_MONTHS).astype(np.int64), + ) + return root.parent + + +def _run(*args): + return subprocess.run( + [sys.executable, "-m", "tools.prereg.aggregate_calibration", *[str(a) for a in args]], + cwd=REPO, capture_output=True, text=True, timeout=300, + ) + + +def _independent(rng): + """Every cell drawn on its own. Var(sum) == sum Var by construction, so ratio -> 1.""" + return rng.gamma(2.0, 3.0, size=(N_CELLS * N_MONTHS, N_DRAWS)) + + +def _joint(rng): + """Every draw is a scenario: one shared multiplier moves all cells together. Positive + dependence, so Var(sum) must exceed sum Var.""" + base = rng.gamma(2.0, 3.0, size=(N_CELLS * N_MONTHS, 1)) + scenario = rng.gamma(2.0, 0.5, size=(1, N_DRAWS)) + return base * scenario + + +class TestDependenceIsDetected: + def test_independent_cells_give_a_ratio_near_one(self, tmp_path): + pooled = _write(tmp_path, _independent(np.random.default_rng(0))) + out = tmp_path / "r.json" + r = _run(pooled, "--json-out", out) + assert r.returncode == 0, r.stdout + r.stderr + ratios = [x["dependence_ratio"] for x in json.loads(out.read_text())["lr_sb_best"]] + assert all(0.7 < x < 1.4 for x in ratios), ( + f"independent cells should give Var(sum) ~ sum Var; got {ratios}" + ) + + def test_joint_scenarios_give_a_ratio_well_above_one(self, tmp_path): + pooled = _write(tmp_path, _joint(np.random.default_rng(0))) + out = tmp_path / "r.json" + _run(pooled, "--json-out", out) + ratios = [x["dependence_ratio"] for x in json.loads(out.read_text())["lr_sb_best"]] + assert all(x > 2.0 for x in ratios), ( + "draws that move together must show positive dependence — this is the property that " + f"makes an aggregate interval wide enough to be honest; got {ratios}" + ) + + def test_the_independent_case_is_called_out_in_the_output(self, tmp_path): + """F1 is the pre-registered falsifier. It has to be visible, not only in the JSON.""" + pooled = _write(tmp_path, _independent(np.random.default_rng(1))) + assert "F1" in _run(pooled).stdout + + +class TestTheForecastCaseIsHonestAboutWhatItCannotScore: + def test_without_actuals_it_says_so(self, tmp_path): + pooled = _write(tmp_path, _independent(np.random.default_rng(2))) + out = _run(pooled).stdout + assert "FORECASTING case" in out and "do not" in out.lower(), ( + "a forecast has no observations yet; the tool must say that rather than imply a " + f"comparison it did not make.\n{out}" + ) + + def test_without_actuals_no_row_claims_a_pit(self, tmp_path): + pooled = _write(tmp_path, _independent(np.random.default_rng(3))) + out = tmp_path / "r.json" + _run(pooled, "--json-out", out) + assert all("pit" not in x for x in json.loads(out.read_text())["lr_sb_best"]) + + +class TestUnderPredictionIsCaught: + @staticmethod + def _actuals(tmp_path, pooled, multiplier): + """Observed totals set to a multiple of the predictive mean, so the expected verdict is + known before the tool runs.""" + import pandas as pd + out = tmp_path / "pre.json" + _run(pooled, "--json-out", out) + rows = json.loads(out.read_text())["lr_sb_best"] + df = pd.DataFrame([ + {"month_id": r["month_id"], "unit": r["unit"], + "observed_total": r["total_mean"] * multiplier} + for r in rows + ]) + path = tmp_path / "actuals.parquet" + df.to_parquet(path) + return path + + def test_a_timid_model_trips_F2_and_F4(self, tmp_path): + """Observed totals ten times the predictive mean: PIT pinned at 1, MCR ~0.1.""" + pooled = _write(tmp_path, _independent(np.random.default_rng(4))) + actuals = self._actuals(tmp_path, pooled, multiplier=10.0) + out = _run(pooled, "--actuals", actuals).stdout + assert "F2" in out, f"under-prediction not flagged\n{out}" + assert "F4" in out, f"MCR not flagged\n{out}" + + def test_a_well_centred_model_trips_neither(self, tmp_path): + """The control. Without it, a tool that printed F2/F4 unconditionally would pass above.""" + pooled = _write(tmp_path, _independent(np.random.default_rng(5))) + actuals = self._actuals(tmp_path, pooled, multiplier=1.0) + out = _run(pooled, "--actuals", actuals).stdout + assert "F2" not in out and "F4" not in out, f"false alarm on a centred model\n{out}" + assert "MCR median" in out + + def test_mcr_is_reported_as_a_ratio_to_observed(self, tmp_path): + pooled = _write(tmp_path, _independent(np.random.default_rng(6))) + actuals = self._actuals(tmp_path, pooled, multiplier=4.0) + out = tmp_path / "r.json" + _run(pooled, "--actuals", actuals, "--json-out", out) + mcrs = [x["mcr"] for x in json.loads(out.read_text())["lr_sb_best"]] + assert all(abs(m - 0.25) < 0.01 for m in mcrs), ( + f"observed set to 4x the predictive mean must give MCR 0.25; got {mcrs}" + ) diff --git a/tests/test_fao_delivery_runner.py b/tests/test_fao_delivery_runner.py index 157bd43d..f2405ee1 100644 --- a/tests/test_fao_delivery_runner.py +++ b/tests/test_fao_delivery_runner.py @@ -182,6 +182,46 @@ def test_the_disk_floor_is_not_below_the_floor_it_delegates_to(self): "later model would refuse, after hours." ) + def test_it_checks_the_appwrite_coordinate_registry(self, result): + """Found during the first rehearsal, with four of eight models already trained. + + The postprocessor leg is fatal without the Appwrite coordinate registry — + `platform_env_require_registry()` says "the registry is the ONLY source of coordinates" + (#308) — and its default path is a relative hop to a SIBLING checkout of views-appwrite, + which a pod that cloned only views-models does not have. It would have failed at step 4, + the last one, after every GPU hour was spent. + + The REGION check passing is not evidence for this: that resolves a COVERAGE declaration, + a different registry entirely. Two registries, two checks. + """ + out = result.stdout + result.stderr + code = _code_only(FAO.read_text()) + assert "coordinate_registry.toml" in code, ( + "preflight does not check for the Appwrite coordinate registry" + ) + assert "APPWRITE_REGISTRY" in code, ( + "preflight must name the override, or an operator on a machine with a different " + "layout has no way to satisfy the check" + ) + assert "registry" in out.lower(), ( + f"preflight ran on a machine with no registry and said nothing about it.\n{out}" + ) + + def test_conda_is_probed_for_CAPABILITY_not_presence(self): + """The 2026-09-30 delivery failed at step 4 — the last — because miniconda 26.7.1 will + not create an environment until its channel Terms of Service are accepted. Preflight + had checked `command -v conda`, which said nothing about that. The binary existing is + not the property the postprocessor needs.""" + code = _code_only(FAO.read_text()) + assert "conda create --dry-run" in code, ( + "preflight must probe that conda can CREATE an environment. `command -v conda` " + "passed on the pod that then failed at the last step." + ) + assert "tos accept" in code, ( + "the refusal must name the remedy — an operator who hits the ToS gate at 4am " + "should not have to find the two commands themselves" + ) + def test_it_reports_every_problem_not_only_the_first(self, result): """A preflight that dies on the first missing thing costs a round trip per problem, on hardware billed by the second.""" @@ -243,6 +283,90 @@ def test_wandb_is_forced_offline_on_the_ensemble_leg(self, code): assert "WANDB_MODE=offline" in tail +class TestThePosteriorIsInspectedBeforeTheDeliveryIsCalledDone: + """The 2026-09-30 delivery passed every structural check and served a posterior whose point + estimate was zero for all 2,333,448 cells. Nothing in the chain said so.""" + + @pytest.fixture(scope="class") + def code(self): + return _code_only(FAO.read_text()) + + def test_posterior_health_runs_on_every_delivery(self, code): + assert "tools.prereg.posterior_health" in code, ( + "nothing reports whether the pooled posterior contains anything. A valid manifest " + "over an empty posterior passes every other check in the chain." + ) + + def test_the_health_check_is_told_which_mode_the_run_is(self, code): + """The numbers cannot distinguish an expected rehearsal from a broken production run.""" + seg = code[code.index("tools.prereg.posterior_health"):] + assert "--mode" in seg[:300], ( + f"posterior_health is invoked without --mode:\n{seg[:300]}" + ) + + def test_anchors_are_captured_by_the_runner_not_by_hand(self, code): + """P8's protection is that anchors are not chosen after seeing the API's answer. Running + it from the runner makes that true by construction; running it by hand afterwards makes + it true only if nobody looked first.""" + assert "tools.prereg.capture_anchors" in code + + def test_the_posterior_is_inspected_before_the_postprocessor_stage(self, code): + """It must run as soon as the pooled frame exists, not after the last step — otherwise a + failure in the postprocessor buries the one signal that the numbers were empty.""" + assert code.index("posterior_health") < code.index("un_fao_postprocessor") + + +class TestTheLastHopIsCheckedNotAssumed: + """On 2026-09-30 every stage reported success and faoapi served NOTHING. + + The artefacts were perfect — that seat downloaded all 108 shards and independently + re-assembled 2,330,712 rows — so no check on this side could have caught it. One of 108 + downloads stalled, there is no retry on that path, and the run was refused. Only asking the + API distinguishes "delivered" from "served". + """ + + @pytest.fixture(scope="class") + def code(self): + return _code_only(FAO.read_text()) + + def test_the_serving_state_is_queried(self, code): + assert "forecast_serving_state" in code, ( + "nothing asks whether faoapi is serving the run. Every other stage can report " + "success while the delivery is dark." + ) + + def test_it_does_not_use_health_status_as_the_authority(self, code): + """`/health`'s `status` and `forecast_freshness` read the newest record in the STORE, + not what is served. During the 2026-09-30 refusal they read 'healthy, age 0.02d, not + stale' while nothing was served at all.""" + assert "forecast_serving_state" in code + assert not re.search(r'jq\s+\.status|"status"\s*==', code), ( + "serving must be judged on forecast_serving_state, not /health's status field" + ) + + def test_recovery_uses_subset_and_explicitly_warns_against_bulk(self, code): + """The route that RECOVERS a refused run is not the route FAO uses. A subset query + performs the full ingest (~200s); /pg/data/forecast/bulk 503s in 0.2s without + attempting anything, so polling it reports failure forever.""" + assert "forecast/subset" in code, ( + "no recovery path — a refused run needs a subset query to trigger the ingest" + ) + # Raw source, NOT comment-stripped: this warning lives in an `echo` the operator reads, + # and `_code_only` strips from the first `#` — which eats the `###` inside echo strings. + # The helper exists to stop a COMMENT standing in for code; an echo string IS the code's + # behaviour, so stripping it here would be the helper defeating the guard. + raw = FAO.read_text() + assert re.search(r"[Dd]o NOT poll.*bulk|bulk.*without attempting", raw, re.S), ( + "the script must warn against polling /pg/data/forecast/bulk to recover; it is the " + "obvious thing to try and it never retries" + ) + + def test_it_refuses_to_claim_delivery_when_it_cannot_check(self): + """No API key means unverified, not fine. The whole failure was a stage reporting + success it had not established. Asserted on raw source for the reason above.""" + assert "CANNOT VERIFY THE LAST HOP" in FAO.read_text() + + class TestARehearsalIsMarkedEverywhereItCanBe: """A rehearsal reaches the FAO shelf, because nothing downstream refuses one (#523). So the marking is the only protection, and it has to be loud.""" diff --git a/tests/test_posterior_health_and_anchors.py b/tests/test_posterior_health_and_anchors.py new file mode 100644 index 00000000..137a40fa --- /dev/null +++ b/tests/test_posterior_health_and_anchors.py @@ -0,0 +1,171 @@ +"""Guards for the two checks added after the 2026-09-30 rehearsal. + +That delivery passed EVERY structural check — valid manifest, right coverage, findability guard +green — while serving a posterior whose point estimate was zero for all 2,333,448 cells. It was +found by hand at 4am, by someone computing anchors for an unrelated probe. + +These tests run the real tools against synthetic posteriors built to be degenerate, sparse and +healthy. They do not read source text: the previous round of guards in this repository went +22/22 green with every fix reverted, because they asserted on strings a comment could satisfy. +""" + +import json +import subprocess +import sys +from pathlib import Path + +import numpy as np +import pytest + +REPO = Path(__file__).resolve().parents[1] +pytest.importorskip("views_frames") +pytest.importorskip("views_frames_summarize") + + +def _make_pooled(tmp_path: Path, draws_by_target: dict) -> Path: + """A pooled ensemble output in the layout the ensemble actually writes. + + y_pred.npy + identifiers.npz — NOT PredictionFrame.save's values.npy. The tools must read + what is on disk, and this fixture is what proves they do. + """ + root = tmp_path / "predictions_forecasting_20260930_024838" + for target, draws in draws_by_target.items(): + d = root / target + d.mkdir(parents=True) + n = draws.shape[0] + np.save(d / "y_pred.npy", draws) + np.savez( + d / "identifiers.npz", + time=np.repeat(np.arange(561, 561 + 4), n // 4).astype(np.int64)[:n], + unit=np.tile(np.arange(1000, 1000 + n // 4), 4).astype(np.int64)[:n], + ) + return root + + +def _run(module: str, *args) -> subprocess.CompletedProcess: + return subprocess.run( + [sys.executable, "-m", module, *[str(a) for a in args]], + cwd=REPO, capture_output=True, text=True, timeout=300, + ) + + +ALL_ZERO = np.zeros((40, 8), dtype=np.float32) + + +def _sparse_with_spike(): + """The 2026-09-30 shape: almost everything zero, a few enormous draws in one cell.""" + a = np.zeros((40, 8), dtype=np.float32) + a[7, :3] = [301274.9, 15000.0, 8000.0] + return a + + +def _healthy(): + rng = np.random.default_rng(1) + return rng.gamma(2.0, 2.0, size=(40, 8)).astype(np.float32) + + +class TestPosteriorHealthDetectsWhatEveryOtherCheckMissed: + def test_an_all_zero_posterior_is_DEGENERATE(self, tmp_path): + pooled = _make_pooled(tmp_path, {"lr_sb_best": ALL_ZERO}) + r = _run("tools.prereg.posterior_health", pooled, "--mode", "rehearsal") + assert "DEGENERATE" in r.stdout, ( + "a posterior of all zeros must be called degenerate. Every structural check passes " + f"on it, which is exactly why this one has to fire.\n{r.stdout}\n{r.stderr}" + ) + + def test_a_degenerate_production_run_exits_non_zero(self, tmp_path): + pooled = _make_pooled(tmp_path, {"lr_sb_best": ALL_ZERO}) + assert _run("tools.prereg.posterior_health", pooled, "--mode", "production").returncode != 0 + + def test_a_degenerate_rehearsal_exits_zero(self, tmp_path): + """A rehearsal at low lesson count is EXPECTED to look like this. Failing the run it was + designed to produce would teach the operator to ignore the exit code, which would cost + more than it saves.""" + pooled = _make_pooled(tmp_path, {"lr_sb_best": ALL_ZERO}) + r = _run("tools.prereg.posterior_health", pooled, "--mode", "rehearsal") + assert r.returncode == 0, r.stdout + r.stderr + assert "EXPECTED for a rehearsal" in r.stdout + + def test_the_two_modes_say_different_things_about_identical_numbers(self, tmp_path): + """The numbers cannot distinguish a rehearsal from a broken production run. The runner + knows which it is, and that is the only reason this can be reported usefully.""" + pooled = _make_pooled(tmp_path, {"lr_sb_best": ALL_ZERO}) + reh = _run("tools.prereg.posterior_health", pooled, "--mode", "rehearsal").stdout + prod = _run("tools.prereg.posterior_health", pooled, "--mode", "production").stdout + assert reh != prod + assert "investigate the" in prod and "investigate the" not in reh + + def test_a_healthy_posterior_is_not_flagged(self, tmp_path): + """The control. Without it, a tool that printed DEGENERATE unconditionally would pass + every test above.""" + pooled = _make_pooled(tmp_path, {"lr_sb_best": _healthy()}) + r = _run("tools.prereg.posterior_health", pooled, "--mode", "production") + assert r.returncode == 0, r.stdout + r.stderr + assert "DEGENERATE" not in r.stdout, r.stdout + + def test_the_json_report_carries_the_numbers_not_only_a_verdict(self, tmp_path): + pooled = _make_pooled(tmp_path, {"lr_sb_best": _sparse_with_spike()}) + out = tmp_path / "health.json" + _run("tools.prereg.posterior_health", pooled, "--mode", "rehearsal", "--json-out", out) + rec = json.loads(out.read_text()) + s = rec["targets"][0] + for k in ("rows", "samples", "draws_nonzero_fraction", "rows_all_zero_fraction", + "max_draw", "point_nonzero"): + assert k in s, f"{k} missing — a verdict without its evidence cannot be argued with" + assert s["max_draw"] == pytest.approx(301274.9, rel=1e-4) + + +class TestAnchorsAreRecordedAndDiscriminating: + def test_it_records_the_raw_draws_not_only_the_point_estimate(self, tmp_path): + """The reason this matters: on 2026-09-30 every point estimate was 0.0, so comparing + point estimates would have passed against an unrelated all-zero dataset. The draws are + what discriminate.""" + pooled = _make_pooled(tmp_path, {"lr_sb_best": _sparse_with_spike()}) + out = tmp_path / "anchors.json" + r = _run("tools.prereg.capture_anchors", pooled, "--mode", "rehearsal", "--out", out) + assert r.returncode == 0, r.stdout + r.stderr + rec = json.loads(out.read_text()) + cells = rec["anchors"]["lr_sb_best"]["cells"] + assert cells, "no anchor cells recorded" + assert all("draws" in c and len(c["draws"]) == 8 for c in cells), ( + "every anchor must carry its full draw vector" + ) + assert any(float(v) > 0 for c in cells for v in c["draws"]), ( + "the spike cell must be among the anchors — the top-N selection is by point estimate " + "and must surface the cells that carry signal" + ) + + def test_the_selection_is_reproducible(self, tmp_path): + """Seeded, so the choice of cells provably was not steered by what the API returned.""" + pooled = _make_pooled(tmp_path, {"lr_sb_best": _healthy()}) + a, b = tmp_path / "a.json", tmp_path / "b.json" + _run("tools.prereg.capture_anchors", pooled, "--mode", "rehearsal", "--out", a) + _run("tools.prereg.capture_anchors", pooled, "--mode", "rehearsal", "--out", b) + ka = [(c["month_id"], c["priogrid_id"]) for c in json.loads(a.read_text())["anchors"]["lr_sb_best"]["cells"]] + kb = [(c["month_id"], c["priogrid_id"]) for c in json.loads(b.read_text())["anchors"]["lr_sb_best"]["cells"]] + assert ka == kb and len(ka) > 0 + + def test_it_records_the_estimator_version(self, tmp_path): + """P8 claimed to remove 'compared the wrong thing' BY CONSTRUCTION because both sides run + the same version. They float independently inside `views-frames>=1.10.2,<2`, and the + SERVING version is not observable from outside. So it is written down.""" + pooled = _make_pooled(tmp_path, {"lr_sb_best": _healthy()}) + out = tmp_path / "anchors.json" + _run("tools.prereg.capture_anchors", pooled, "--mode", "rehearsal", "--out", out) + rec = json.loads(out.read_text()) + assert rec.get("views_frames_version") not in (None, "", "unknown"), ( + "the estimator version must be recorded, or a later mismatch is mysterious " + "rather than diagnosable" + ) + assert "tower_point" in rec["estimator"] + + def test_full_precision_is_preserved(self, tmp_path): + """repr(), not round(): the comparison is made across a boundary where rounding is the + difference between a real disagreement and a spurious one.""" + pooled = _make_pooled(tmp_path, {"lr_sb_best": _sparse_with_spike()}) + out = tmp_path / "anchors.json" + _run("tools.prereg.capture_anchors", pooled, "--mode", "rehearsal", "--out", out) + rec = json.loads(out.read_text()) + vals = [v for c in rec["anchors"]["lr_sb_best"]["cells"] for v in c["draws"]] + assert all(isinstance(v, str) for v in vals), "values must be repr strings, not floats" + assert any("." in v for v in vals) diff --git a/tools/podrun/pod_run_fao_delivery.sh b/tools/podrun/pod_run_fao_delivery.sh index ed794fed..1c521c6b 100755 --- a/tools/podrun/pod_run_fao_delivery.sh +++ b/tools/podrun/pod_run_fao_delivery.sh @@ -80,6 +80,7 @@ mkdir -p "$OUT" 2>/dev/null || { exit 1 } rm -f "$OUT/STATUS" "$OUT/REHEARSAL" "$OUT/PUBLISHED" +rm -rf "$OUT/.conda_probe" exec > >(tee -a "$LOG") 2>&1 stage() { echo "$1" > "$OUT/STAGE"; echo "=== [$(date +%H:%M:%S)] $1 ==="; } @@ -186,7 +187,25 @@ fi # `conda shell.bash hook` / `conda create --prefix` / `conda activate`, while this pod builds # a uv venv — so the two legs need different interpreters and a pod can satisfy one and not # the other. Without conda the delivery dies at step 4, after every GPU hour is spent. -command -v conda >/dev/null 2>&1 || note "conda — the un_fao postprocessor launcher requires it (tools/launcher/postprocessor.sh:72). A uv venv is not enough." +# conda must be able to CREATE A PREFIX, not merely be on PATH. On 2026-09-30 this check +# passed on `command -v conda` and the delivery then failed at step 4 — the last one, after +# every GPU hour — because miniconda 26.7.1 refuses to create an environment until its channel +# Terms of Service are accepted: +# conda tos accept --override-channels --channel https://repo.anaconda.com/pkgs/main +# conda tos accept --override-channels --channel https://repo.anaconda.com/pkgs/r +# The binary existing said nothing about that. A --dry-run create exercises resolution, channel +# access and the ToS gate together, in seconds, which is the whole point of doing it here. +if ! command -v conda >/dev/null 2>&1; then + note "conda — the un_fao postprocessor launcher requires it (tools/launcher/postprocessor.sh:72). A uv venv is not enough." +elif ! CONDA_ERR=$(conda create --dry-run --prefix "$OUT/.conda_probe" python=3.11 2>&1); then + note "conda is installed but CANNOT CREATE AN ENVIRONMENT, which is what the postprocessor needs. + $(printf '%s' "$CONDA_ERR" | grep -iE 'terms of service|tos accept|CondaToS|channel' | head -3) + If this is the Terms of Service gate: + conda tos accept --override-channels --channel https://repo.anaconda.com/pkgs/main + conda tos accept --override-channels --channel https://repo.anaconda.com/pkgs/r" +else + echo " conda can create an environment" +fi # The coordinate registry, which the FAO queryset derives its region from. if [ -x "$VENV/bin/python" ] && [ -d "$REPO" ]; then @@ -199,6 +218,29 @@ if [ -x "$VENV/bin/python" ] && [ -d "$REPO" ]; then fi fi +# The Appwrite COORDINATE REGISTRY, which is a different registry from the coverage one the +# REGION check above resolves — and the postprocessor leg is fatal without it: +# platform_env_require_registry() says so outright, "the registry is the ONLY source of +# coordinates" (#308). Its default path is a relative hop to a SIBLING checkout of +# views-appwrite, which a pod that cloned only views-models does not have. +# +# Found 2026-09-30 during the first rehearsal, with four of eight models already trained. It +# would have failed at step 4 — the last one — after every GPU hour was spent. The REGION check +# passing is not evidence for this one, which is exactly why it needs its own line. +# +# APPWRITE_REGISTRY overrides the path. Note that exporting it AFTER a run has started does not +# help: the delivery sources its environment once, and the postprocessor inherits that. Place +# the file, or set the variable, before launching. +REGISTRY_PATH="${APPWRITE_REGISTRY:-$REPO/../views-appwrite/docs/ADRs/platform/coordinate_registry.toml}" +if [ -f "$REGISTRY_PATH" ]; then + echo " Appwrite coordinate registry: $REGISTRY_PATH" +else + note "the Appwrite coordinate registry is missing — the un_fao postprocessor (step 4, the LAST step) is fatal without it. + looked for: $REGISTRY_PATH + Either place it there, or export APPWRITE_REGISTRY=/path/to/coordinate_registry.toml + BEFORE launching. It holds non-secret identifiers only; its own header says so." +fi + nvidia-smi -L >/dev/null 2>&1 || note "no GPU visible" AVAIL_GB=$(df -BG --output=avail "$ROOT" 2>/dev/null | tail -1 | tr -dc '0-9') # DISK_FLOOR_GB, and the honest state of what it is based on. @@ -262,6 +304,32 @@ export WANDB_MODE=offline WANDB_SILENT=true # the eight runs above are wasted. -p publishes the wire shards (ADR-013). "$VENV/bin/python" main.py -r forecasting -f -sa -p || die "$ENSEMBLE pool-and-publish exited non-zero" +# ── 3b. is there anything IN the posterior? ─────────────────────────────────────────── +# Runs between the publish and the postprocessor because that is the first moment the pooled +# frame exists on disk. It cannot prevent anything — pooling and publishing are one invocation, +# so the numbers are already on the shelf — but on 2026-09-30 a delivery passed every structural +# check while `tower_point` returned zero for all 2,333,448 cells, and that was found by hand at +# 4am. A readout every run turns "remember to look" into "cannot miss". +stage posterior_health +POOLED=$(ls -d "$REPO/ensembles/$ENSEMBLE/data/generated/predictions_forecasting_"* 2>/dev/null | tail -1) +if [ -z "$POOLED" ]; then + echo "!!! no pooled output found under $ENSEMBLE — cannot report posterior health" +else + MODE_ARG=production + [ -n "$REHEARSAL_LESSONS" ] && MODE_ARG=rehearsal + "$VENV/bin/python" -m tools.prereg.posterior_health "$POOLED" \ + --mode "$MODE_ARG" --json-out "$OUT/POSTERIOR_HEALTH.json" \ + || echo "### posterior health reported a problem above — the delivery is already published" + + # P8 anchors, captured HERE and not later: the property the pre-registration protects is that + # anchors are not chosen after seeing what the API returns, and running this from the runner + # is what makes that true by construction rather than by discipline. + stage capture_anchors + "$VENV/bin/python" -m tools.prereg.capture_anchors "$POOLED" \ + --mode "$MODE_ARG" --out "$OUT/P8_ANCHORS.json" \ + || echo "!!! anchor capture failed — P8 cannot be evaluated for this run" +fi + # ── 4. the FAO postprocessor ────────────────────────────────────────────────────────── stage un_fao_postprocessor cd "$REPO" || die "cannot enter repo" @@ -270,6 +338,61 @@ cd "$REPO" || die "cannot enter repo" # it as this script failing. bash postprocessors/un_fao/run.sh || die "the un_fao postprocessor exited non-zero — read the message before assuming the worst; DeliveryNotFindableError means 1.4.0's findability guard fired and it names every object it checked" +# ── 4b. the LAST hop: is faoapi actually serving it? ────────────────────────────────── +# Added after the 2026-09-30 rehearsal, where every stage above reported success and the run +# was NOT served: faoapi refused it with reason="ingest_failed" after one of 108 shard +# downloads stalled, and there is no retry on that path. The artefacts were perfect — that seat +# downloaded all 108 and re-assembled 2,330,712 rows independently — so nothing on this side +# could have detected it. Only asking the API can. +# +# TWO THINGS THAT COST TIME TO LEARN AND ARE NOT GUESSABLE: +# +# * The route that RECOVERS a refused run is not the route FAO uses. A subset query +# re-evaluates the newest run and performs the full ingest (~200s). `/pg/data/forecast/bulk` +# short-circuits on the missing grid artefact and 503s in 0.2s WITHOUT ATTEMPTING ANYTHING +# — polling it reports failure forever and never retries. +# * `/health`'s `status` and `forecast_freshness` read the newest record in the STORE, not what +# is served. During the refusal they read "healthy, age 0.02d, not stale" while nothing was +# served at all. The authoritative field is `forecast_serving_state`. +# +# Needs a consumer API key, which this repo does not hold. Without one the step prints the +# verified manual procedure rather than pretending to check. +stage verify_served +FAOAPI_BASE="${FAOAPI_BASE:-https://faoapi.viewsforecasting.org}" +if [ -z "${FAOAPI_KEY:-}" ]; then + echo "### CANNOT VERIFY THE LAST HOP — no \$FAOAPI_KEY in the environment." + echo "### Everything above can report success while faoapi serves nothing (2026-09-30)." + echo "### Check by hand, and note the two traps:" + echo "### curl -s -H \"x-api-key: \$KEY\" '$FAOAPI_BASE/health' | jq .forecast_serving_state" + echo "### # if it reads {\"degraded\": true, \"reason\": \"ingest_failed\"}, trigger an ingest:" + echo "### curl -s -H \"x-api-key: \$KEY\" \\" + echo "### '$FAOAPI_BASE/pg/data/forecast/subset?time_ids=561&features=pred_lr_ged_sb&entity_ids=100001'" + echo "### # ~200 seconds. Then re-check /health. Do NOT poll /pg/data/forecast/bulk to" + echo "### # recover: it 503s in 0.2s without attempting an ingest." +else + SERVING=$(curl -s --max-time 60 -H "x-api-key: $FAOAPI_KEY" "$FAOAPI_BASE/health" 2>/dev/null \ + | python3 -c 'import json,sys; d=json.load(sys.stdin); print(json.dumps(d.get("forecast_serving_state",{})))' 2>/dev/null) + echo "forecast_serving_state: ${SERVING:-}" + case "$SERVING" in + *'"degraded": false'*|*'"degraded":false'*) + echo "### faoapi IS serving this run." ;; + *ingest_failed*) + echo "### faoapi refused the run (ingest_failed). Triggering an ingest — ~200s." + curl -s --max-time 400 -H "x-api-key: $FAOAPI_KEY" \ + "$FAOAPI_BASE/pg/data/forecast/subset?time_ids=561&features=pred_lr_ged_sb&entity_ids=100001" \ + >/dev/null 2>&1 + SERVING2=$(curl -s --max-time 60 -H "x-api-key: $FAOAPI_KEY" "$FAOAPI_BASE/health" 2>/dev/null \ + | python3 -c 'import json,sys; d=json.load(sys.stdin); print(json.dumps(d.get("forecast_serving_state",{})))' 2>/dev/null) + echo "after ingest attempt: ${SERVING2:-}" + case "$SERVING2" in + *'"degraded": false'*|*'"degraded":false'*) echo "### recovered — faoapi is serving this run." ;; + *) echo "### STILL NOT SERVED. The artefacts are in the bucket; the last hop is not done." ;; + esac ;; + *) + echo "### could not determine serving state — check by hand before calling this delivered." ;; + esac +fi + # ── 5. what actually landed ─────────────────────────────────────────────────────────── stage report # By name, from the live surface, not inferred from an exit code. `python -m tools.liveness` diff --git a/tools/prereg/__init__.py b/tools/prereg/__init__.py new file mode 100644 index 00000000..1e9e7f33 --- /dev/null +++ b/tools/prereg/__init__.py @@ -0,0 +1,31 @@ +"""Evidence about a delivery that its structural checks cannot produce. + +A delivery can pass every check in the chain and still be worthless. On 2026-09-30 one did: the +manifest was valid, coverage was exactly ``land_gaul``'s 64,742 cells, the findability guard +resolved all 111 objects — and ``tower_point``, the estimator views-faoapi serves, returned zero +for every one of 2,333,448 cells. The posterior was empty. Nothing in the chain said so, and it +was found by hand at 4am by someone computing anchor values for an unrelated probe. + +This package produces the two things that would have said so, on every run: + + python -m tools.prereg.posterior_health --mode rehearsal|production + python -m tools.prereg.capture_anchors --mode ... --out P8_ANCHORS.json + +``posterior_health`` answers *is there anything in it?* — the fraction of cells with no signal in +any draw, the largest draw, and how many cells carry a non-zero served point estimate. The same +numbers mean opposite things for a rehearsal and a production run, so it is told which it is; it +cannot be inferred from the data, and that is the whole reason a human kept having to. + +``capture_anchors`` implements pre-registration v3's **P8** — *the values served ARE the values +we produced*. It records both the point estimate and the raw draws, because the point estimate +alone could not discriminate on the run that motivated this: comparing 0.0 to 0.0 would have +passed against an unrelated empty dataset. + +House rules. Both read the pooled frame as the ensemble writes it — ``y_pred.npy`` plus +``identifiers.npz``, not ``PredictionFrame.load``'s ``values.npy``. Anchor selection is seeded, so +it is reproducible and provably not steered by what the API returned. Neither tool can PREVENT +anything: pooling and publishing are a single invocation, so by the time a posterior can be +inspected it is already on the shelf. They exist to make it impossible to miss, not impossible +to happen — the refusal, if one is ever wanted, belongs where the publish decision is made +(views-models#523). +""" diff --git a/tools/prereg/aggregate_calibration.py b/tools/prereg/aggregate_calibration.py new file mode 100644 index 00000000..08109f81 --- /dev/null +++ b/tools/prereg/aggregate_calibration.py @@ -0,0 +1,209 @@ +"""Does the forecast add up? — pre-registered aggregate calibration and dependence diagnostics. + +THE QUESTION THIS ANSWERS. Sum the served tower-map surface over a conflict-affected country and +you get far less than history records. That looks like bad calibration and mostly is not: it is +the wrong arithmetic. `tower_point` is a MODE-like summary — its own docstring says it returns 0 +"when the zero atom dominates, by density" — and for a zero-inflated, heavy-tailed variable the +mode is ~0 almost everywhere while the mean is not. Expectation is linear, modes are not: +``E[sum] = sum E[x]`` always holds, ``mode(sum) != sum mode(x)`` for skewed data. So summing MAP +surfaces is guaranteed to undershoot even for a perfectly calibrated model, and tells you nothing. + +The right object is the distribution of the TOTAL, obtained by summing the DRAWS: + + for each draw d: total_d = sum over cells of x[cell, d] + +That yields a posterior over the aggregate, which can be compared with what was observed. + +THE FAILURE MODE WORTH LOOKING FOR, which the above sets up. Every cell's marginal can be +perfectly calibrated while the TOTAL is badly calibrated, because the total's spread depends on +the DEPENDENCE between cells. If each draw is a coherent joint scenario, totals have realistic +spread. If the draws are independent per-cell marginals stitched together, summing ~64,818 of +them washes out the tail — variances add instead of covariances accumulating — and country totals +become far too narrow and too low. That is checkable and this tool checks it: + + dependence_ratio = Var_d(total) / sum_cells Var_d(cell) + + ~1.0 cells behave independently across draws — the tail of the aggregate is understated + >1.0 positive dependence — draws carry joint structure, which is what a scenario should be + <1.0 negative dependence, which for conflict would itself want explaining + +PRE-REGISTERED FALSIFIERS — stated before the numbers are looked at, which is the only time such +a statement is worth anything: + + F1 dependence_ratio within [0.9, 1.1] for most (unit, target). The draws are then effectively + independent marginals and every aggregate interval is too narrow, regardless of how good + the per-cell marginals look. + F2 with actuals: PIT values concentrated near 1.0 — the observed total sits above almost the + whole predictive distribution. This is the timid-under-prediction signature, and it is the + one the platform already has reason to expect: MSLE rewards timidity, which is why MCR + (yhat/ybar) exists as a magnitude guardrail and why ranking on MSLE alone is refused. + F3 with actuals: coverage of the central 90% interval far below 0.90 across units. + F4 MCR far below 1.0 — the predicted mean total is a fraction of the observed mean total. + +WHAT IT CANNOT DO. A FORECASTING run predicts months with no observations yet, so F2-F4 need a +partition where actuals exist (calibration or validation). Against a forecast the tool reports the +predictive totals and the dependence ratio only, and says so rather than inventing a comparison. +""" + +from __future__ import annotations + +import argparse +import json +from pathlib import Path + +import numpy as np + + +def _load_pooled(target_dir: Path): + """y_pred.npy + identifiers.npz — the layout the ensemble writes, not PredictionFrame.load's.""" + y = np.load(target_dir / "y_pred.npy", mmap_mode="r") + ids = np.load(target_dir / "identifiers.npz") + return y, ids["time"], ids["unit"] + + +def _unit_map(path: Path | None, cells: np.ndarray) -> tuple[np.ndarray, dict]: + """Map each cell to an aggregation unit. Without a mapping every cell is one global unit — + still informative, and it needs nothing we might not have on a pod.""" + if path is None: + return np.zeros(len(cells), dtype=np.int64), {0: "GLOBAL"} + import pandas as pd + + m = pd.read_parquet(path) + cols = [c for c in m.columns if c != "priogrid_id"] + if not cols: + raise SystemExit(f"{path} has no unit column beside priogrid_id") + lookup = dict(zip(m["priogrid_id"].to_numpy(), m[cols[0]].to_numpy())) + names = sorted({str(v) for v in lookup.values()}) + code = {n: i for i, n in enumerate(names)} + return ( + np.array([code.get(str(lookup.get(c, "UNMAPPED")), -1) for c in cells], dtype=np.int64), + {i: n for n, i in code.items()} | {-1: "UNMAPPED"}, + ) + + +def aggregate(target_dir: Path, unit_of_cell: np.ndarray, unit_names: dict, + actuals: dict | None = None) -> list[dict]: + y, time, _ = _load_pooled(target_dir) + out = [] + for month in sorted(set(int(t) for t in time)): + month_rows = time == month + for ucode in sorted(set(int(u) for u in unit_of_cell)): + sel = month_rows & (unit_of_cell == ucode) + n = int(sel.sum()) + if n == 0: + continue + block = np.asarray(y[sel], dtype=np.float64) # (cells, draws) + totals = block.sum(axis=0) # (draws,) + # Dependence: variance of the SUM against the sum of per-cell variances. Equal iff + # the cells are uncorrelated across draws. + per_cell_var = block.var(axis=1, ddof=1).sum() + total_var = float(totals.var(ddof=1)) + row = { + "month_id": month, + "unit": unit_names.get(ucode, str(ucode)), + "cells": n, + "draws": int(block.shape[1]), + "total_mean": float(totals.mean()), + "total_median": float(np.median(totals)), + "total_q05": float(np.quantile(totals, 0.05)), + "total_q95": float(np.quantile(totals, 0.95)), + "total_max": float(totals.max()), + "sum_of_cell_map_like_zeros": int((block.max(axis=1) == 0).sum()), + "dependence_ratio": (total_var / per_cell_var) if per_cell_var > 0 else None, + } + # Scored HERE, where the draw totals are still in hand. Computing PIT later would + # mean carrying 128 floats per aggregate through the report purely to re-derive a + # number available for free at this point. + obs = (actuals or {}).get((month, row["unit"])) + if obs is not None: + obs = float(obs) + row["observed_total"] = obs + # `<=` not `<`: a zero-inflated predictive puts real mass exactly at 0, and an + # observed 0 should not be scored as sitting below all of it. + row["pit"] = float(np.mean(totals <= obs)) + row["covered_90"] = bool(row["total_q05"] <= obs <= row["total_q95"]) + row["mcr"] = (row["total_mean"] / obs) if obs > 0 else None + out.append(row) + return out + + +def main(argv=None) -> int: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("pooled_dir", type=Path) + p.add_argument("--units", type=Path, default=None, + help="parquet mapping priogrid_id -> unit (e.g. the un_fao GAUL sidecar). " + "Without it, one GLOBAL unit.") + p.add_argument("--actuals", type=Path, default=None, + help="parquet with month_id, unit, observed_total. Absent for a FORECASTING " + "run, where no observation exists yet.") + p.add_argument("--json-out", type=Path, default=None) + args = p.parse_args(argv) + + targets = sorted(d for d in args.pooled_dir.iterdir() if d.is_dir() and d.name.startswith("lr_")) + if not targets: + print(f"no lr_* targets under {args.pooled_dir}") + return 2 + + _, _, cells = _load_pooled(targets[0]) + unit_of_cell, unit_names = _unit_map(args.units, cells) + + actuals = None + if args.actuals is not None: + import pandas as pd + a = pd.read_parquet(args.actuals) + missing = {"month_id", "unit", "observed_total"} - set(a.columns) + if missing: + raise SystemExit(f"--actuals is missing columns: {sorted(missing)}") + actuals = {(int(r.month_id), str(r.unit)): float(r.observed_total) for r in a.itertuples()} + + report = {} + for t in targets: + report[t.name] = aggregate(t, unit_of_cell, unit_names, actuals) + + print("=== aggregate calibration ===") + if args.actuals is None: + print(" no --actuals: this is the FORECASTING case. Predictive totals and the") + print(" dependence ratio only — PIT, coverage and MCR need observations that do not") + print(" exist yet. Run against calibration or validation to score those.") + for name, rows in report.items(): + ratios = [r["dependence_ratio"] for r in rows if r["dependence_ratio"] is not None] + means = [r["total_mean"] for r in rows] + print(f" {name}: {len(rows)} (unit, month) aggregates") + if means: + print(f" predictive total per aggregate — mean {np.mean(means):,.1f}, " + f"max {np.max(means):,.1f}") + if ratios: + med = float(np.median(ratios)) + near_one = float(np.mean([(0.9 <= x <= 1.1) for x in ratios])) + print(f" dependence_ratio median {med:.3f}; " + f"{near_one:.0%} of aggregates in [0.9, 1.1]") + if near_one > 0.5: + print(" *** F1: draws behave as INDEPENDENT per-cell marginals. Aggregate") + print(" *** intervals are too narrow no matter how good the marginals are.") + scored = [r for r in rows if "pit" in r] + if scored: + pits = [r["pit"] for r in scored] + cov = float(np.mean([r["covered_90"] for r in scored])) + mcrs = [r["mcr"] for r in scored if r["mcr"] is not None] + print(f" scored against actuals: {len(scored)} aggregates") + print(f" PIT median {np.median(pits):.3f}; share above 0.95: " + f"{np.mean([x > 0.95 for x in pits]):.0%}") + print(f" 90% interval coverage {cov:.0%} (want ~0.90)") + if mcrs: + print(f" MCR median {np.median(mcrs):.3f} (want ~1.0)") + if np.mean([x > 0.95 for x in pits]) > 0.25: + print(" *** F2: observed totals sit above the predictive distribution") + print(" *** in a large share of aggregates — timid under-prediction.") + if cov < 0.75: + print(" *** F3: 90% intervals cover far less than 90%.") + if mcrs and np.median(mcrs) < 0.5: + print(" *** F4: predicted mean total is under half the observed.") + + if args.json_out: + args.json_out.write_text(json.dumps(report, indent=1, default=str)) + print(f" wrote {args.json_out}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/prereg/capture_anchors.py b/tools/prereg/capture_anchors.py new file mode 100644 index 00000000..7fe913f4 --- /dev/null +++ b/tools/prereg/capture_anchors.py @@ -0,0 +1,115 @@ +"""Record what a delivery PRODUCED, before anything queries the API. + +Pre-registration v3's P8 — *the values served ARE the values we produced* — asks for anchor +values written down "before the publish". That ordering is unachievable as written: pooling and +publishing are a single invocation, so the pooled numbers never exist un-published. + +What the rule protects is narrower and IS achievable: anchors must not be chosen after seeing +what the API returns. So this reads the pooled frame on disk — the exact artefact that was +published — and is run automatically at completion, before any API call. Running it by hand +afterwards preserves the property only if nobody has looked; running it from the runner removes +that "only if". + +WHAT IT RECORDS, and why both. The `tower_point` estimate, because that is the statistic faoapi +serves and therefore the thing P8 compares. AND the raw draws, because on 2026-09-30 the point +estimate was zero for every cell in the delivery — so comparing it would have passed while +proving nothing. The draws discriminate where the point estimate cannot: a cell with 3 non-zero +draws of 128 and a specific spike is a fingerprint. A probe that can only DETECT a disagreement +is worth less than one that can also DIAGNOSE it. + +The cell SELECTION is seeded, so it is reproducible and provably not steered by the answer. +""" + +from __future__ import annotations + +import argparse +import hashlib +import json +from datetime import datetime, timezone +from pathlib import Path + +import numpy as np + +SEED = 20260930 +N_RANDOM = 20 +N_TOP = 5 + + +def _load_pooled(target_dir: Path): + import views_frames as vf + + y = np.load(target_dir / "y_pred.npy") + ids = np.load(target_dir / "identifiers.npz") + index = vf.SpatioTemporalIndex(time=ids["time"], unit=ids["unit"], level=vf.SpatialLevel.PGM) + return vf.PredictionFrame(y_pred=y, index=index), y, ids + + +def anchors_for(target_dir: Path, rng: np.random.Generator) -> dict: + from views_frames_summarize import tower_point + + frame, draws, ids = _load_pooled(target_dir) + point = np.asarray(tower_point(frame).values).squeeze() + + picked = rng.choice(len(point), size=min(N_RANDOM, len(point)), replace=False) + top = np.argsort(point)[-N_TOP:][::-1] + cells = [] + for i in list(picked) + list(top): + i = int(i) + cells.append({ + "month_id": int(ids["time"][i]), + "priogrid_id": int(ids["unit"][i]), + # repr() rather than float(): full precision, no rounding at the boundary the + # comparison will be made across. + "tower_point": repr(float(point[i])), + "draws": [repr(float(x)) for x in draws[i]], + }) + return { + "rows": int(draws.shape[0]), + "samples": int(draws.shape[1]), + "y_pred_sha256": hashlib.sha256((target_dir / "y_pred.npy").read_bytes()).hexdigest(), + "point_nonzero": int((point > 0).sum()), + "point_max": repr(float(point.max())), + "cells": cells, + } + + +def main(argv=None) -> int: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("pooled_dir", type=Path) + p.add_argument("--out", type=Path, required=True) + p.add_argument("--mode", choices=("rehearsal", "production"), required=True) + args = p.parse_args(argv) + + import views_frames as vf + + targets = sorted(d for d in args.pooled_dir.iterdir() if d.is_dir() and d.name.startswith("lr_")) + if not targets: + print(f"no lr_* targets under {args.pooled_dir}") + return 2 + + rng = np.random.default_rng(SEED) + rec = { + "captured_utc": datetime.now(timezone.utc).isoformat(), + "mode": args.mode, + "pooled_run": args.pooled_dir.name, + "purpose": "prereg v3 P8 — values PRODUCED, recorded before any API query", + # Recorded because P8 claimed to remove the 'compared the wrong thing' failure mode BY + # CONSTRUCTION, on the basis that both sides run the same version. They float + # independently inside `views-frames>=1.10.2,<2`, and the SERVING version is not + # observable from outside (faoapi's /version reports the app, not its dependencies). + # So the version is written down instead of assumed. + "views_frames_version": getattr(vf, "__version__", "unknown"), + "estimator": "views_frames_summarize.tower_point", + "selection": f"seeded rng({SEED}): {N_RANDOM} random cells + {N_TOP} highest by point estimate", + "anchors": {t.name: anchors_for(t, rng) for t in targets}, + } + args.out.parent.mkdir(parents=True, exist_ok=True) + args.out.write_text(json.dumps(rec, indent=1)) + total = sum(len(v["cells"]) for v in rec["anchors"].values()) + print(f"wrote {args.out}: {len(rec['anchors'])} target(s), {total} anchor cells, " + f"views_frames {rec['views_frames_version']}") + return 0 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/tools/prereg/posterior_health.py b/tools/prereg/posterior_health.py new file mode 100644 index 00000000..8facd680 --- /dev/null +++ b/tools/prereg/posterior_health.py @@ -0,0 +1,141 @@ +"""Report the health of a pooled posterior, so a degenerate one cannot be missed. + +WHY THIS EXISTS. The 2026-09-30 rehearsal delivered successfully and served nothing useful: the +pooled posterior had all 128 draws at zero in 98.2% of cells, and `tower_point` — the estimator +faoapi serves — correctly returned 0 for every one of 2,333,448 rows. Every structural check +passed. The manifest was valid, the coverage was right, the findability guard was green. Nothing +in the delivery said the numbers were empty, and it was found by hand at 4am only because +someone happened to be computing anchor values for an unrelated probe. + +For a REHEARSAL that outcome is expected — 40 lessons is an undertrained model. For a +PRODUCTION run it means the delivery is worthless, and the partner would be the one to notice. +The numbers are identical; only the mode differs, and the runner knows the mode. + +WHAT IT CANNOT DO. It cannot prevent anything. Pooling and publishing are a single invocation, +so by the time a posterior can be inspected it has already been published. This reports; the +refusal, if one is ever wanted, has to live where the publish decision is made. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +import numpy as np + + +def _load_pooled(target_dir: Path): + """Load a pooled PredictionFrame written as y_pred.npy + identifiers.npz. + + Deliberately NOT `PredictionFrame.load`: that expects `values.npy`, and the ensemble writes + `y_pred.npy`. The mismatch cost twenty minutes on 2026-09-30 and is worth stating here rather + than rediscovering. + """ + import views_frames as vf + + y = np.load(target_dir / "y_pred.npy") + ids = np.load(target_dir / "identifiers.npz") + index = vf.SpatioTemporalIndex( + time=ids["time"], unit=ids["unit"], level=vf.SpatialLevel.PGM + ) + return vf.PredictionFrame(y_pred=y, index=index), y, ids + + +def health(target_dir: Path) -> dict: + from views_frames_summarize import tower_point + + frame, draws, _ = _load_pooled(target_dir) + point = np.asarray(tower_point(frame).values).squeeze() + row_max = draws.max(axis=1) + return { + "target": target_dir.name, + "rows": int(draws.shape[0]), + "samples": int(draws.shape[1]), + "draws_nonzero_fraction": float((draws > 0).mean()), + "rows_all_zero_fraction": float((row_max == 0).mean()), + "rows_with_any_signal": int((row_max > 0).sum()), + "max_draw": float(draws.max()), + "point_nonzero": int((point > 0).sum()), + "point_max": float(point.max()), + } + + +def verdict(stats: dict) -> tuple[str, str]: + """DEGENERATE / SPARSE / OK, with the reason in the operator's words. + + The DEGENERATE threshold is deliberately the starkest fact available — the served point + estimate is zero EVERYWHERE — rather than a tuned fraction. A threshold that needs + calibration is a threshold that gets argued with; this one cannot be. + """ + if stats["point_nonzero"] == 0: + return "DEGENERATE", ( + "the served point estimate is zero for EVERY one of " + f"{stats['rows']:,} cells. Whatever is delivered, a consumer reads nothing from it." + ) + if stats["rows_all_zero_fraction"] > 0.999: + return "SPARSE", ( + f"{stats['rows_all_zero_fraction']:.3%} of cells have no signal in any draw; " + f"only {stats['rows_with_any_signal']:,} cells carry anything." + ) + return "OK", ( + f"{stats['point_nonzero']:,} cells carry a non-zero point estimate " + f"(max {stats['point_max']:.6g})." + ) + + +def main(argv=None) -> int: + p = argparse.ArgumentParser(description=__doc__) + p.add_argument("pooled_dir", type=Path, help="predictions_forecasting_ of the ensemble") + p.add_argument("--mode", choices=("rehearsal", "production"), required=True) + p.add_argument("--json-out", type=Path, default=None) + args = p.parse_args(argv) + + targets = sorted(d for d in args.pooled_dir.iterdir() if d.is_dir() and d.name.startswith("lr_")) + if not targets: + print(f"no lr_* target directories under {args.pooled_dir}", file=sys.stderr) + return 2 + + report, worst = [], "OK" + order = {"OK": 0, "SPARSE": 1, "DEGENERATE": 2} + for t in targets: + stats = health(t) + stats["verdict"], stats["reason"] = verdict(stats) + report.append(stats) + if order[stats["verdict"]] > order[worst]: + worst = stats["verdict"] + + print("=== posterior health ===") + for s in report: + print( + f" {s['target']:<14} {s['verdict']:<11} " + f"rows={s['rows']:,} samples={s['samples']} " + f"all-zero={s['rows_all_zero_fraction']:.2%} " + f"point_nonzero={s['point_nonzero']:,} max_draw={s['max_draw']:.6g}" + ) + print(f" {s['reason']}") + + if args.json_out: + args.json_out.write_text(json.dumps({"mode": args.mode, "verdict": worst, "targets": report}, indent=1)) + + if worst == "DEGENERATE": + banner = "#" * 78 + print(banner) + if args.mode == "rehearsal": + print("### POSTERIOR IS DEGENERATE — EXPECTED for a rehearsal at low lesson count.") + print("### The chain is proven; the NUMBERS are not. Do not read these as forecasts.") + else: + print("### POSTERIOR IS DEGENERATE ON A PRODUCTION RUN.") + print("### The delivery is structurally valid and serves nothing. It is already") + print("### published — this cannot unpublish it. Supersede it and investigate the") + print("### MODELS: a valid manifest over an empty posterior passes every other check.") + print(banner) + # Non-zero ONLY for production: a rehearsal is expected to look like this, and failing + # the run it was designed to produce would train the operator to ignore the exit code. + return 1 if args.mode == "production" else 0 + return 0 + + +if __name__ == "__main__": + raise SystemExit(main())