fix(#499): the delivery that passed every check and served nothing — report the posterior, capture P8 anchors, probe conda for capability

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}"
+        )

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."""