From 73af1899f9468c9b0c2642ae1500247bdf334fb2 Mon Sep 17 00:00:00 2001 From: Max Ghenis Date: Wed, 15 Jul 2026 14:19:19 -0400 Subject: [PATCH] Add release-gate preflight: catch statically-knowable gate failures without a solve MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Build M release attempts 9/10/11 each burned ~2h of calibration to surface ONE gate-group failure that was already determined by the inputs — the base pool values, the frozen selection, and the target/coverage registry — and needed no weight solve to see. This preflight recovers those signals in minutes so a release launch is not the first place a knowable defect is caught. `tools/preflight_us_release_gates.py` + `release_gate_preflight.py` run four checks, each mirroring a live release gate but evaluated pre-solve, reusing the release tool's own register, band math, and materialization algebra rather than re-declaring constants: 1. Selection carryover — the frozen selection-source manifest maps cleanly onto the base pool (the frozen-support recovery contract). 2. Zero-support preview — compiled positive fiscal targets whose materialized support is ~0 under the selection at base weights (direct-column targets checked via the calibrate constraint compiler; engine-derived measures marked not statically checkable). 3. Export-mass parity risk — each export-mass column's pool mass at base weights vs its reference band, honoring US_EXPORT_INPUT_MASS_REVIEWED_EXCLUSIONS. 4. Smoke-probe support audit — every reform-coverage probe leaf's pool vs selected nonzero support and pool sign-leg decomposition; a leaf with pool support but zero selected support FAILs (populace#434 keogh class); a thin selection or a signed leaf whose net contradicts expected_sign is AT-RISK (populace#432 farm class). Reports PASS/FAIL/AT-RISK with measured numbers; exit 1 on FAIL, 2 on AT-RISK, 0 clean. Synthetic-fixture tests exercise each check's FAIL and PASS paths (including the keogh and parity-risk classes) and run in the normal pytest suite with no policyengine-us and no real H5. README documents when to run it. Co-Authored-By: Claude Fable 5 --- README.md | 42 + .../us_runtime/release_gate_preflight.py | 907 ++++++++++++++++++ .../tests/test_us_release_gate_preflight.py | 536 +++++++++++ tools/preflight_us_release_gates.py | 151 +++ 4 files changed, 1636 insertions(+) create mode 100644 packages/populace-build/src/populace/build/us_runtime/release_gate_preflight.py create mode 100644 packages/populace-build/tests/test_us_release_gate_preflight.py create mode 100644 tools/preflight_us_release_gates.py diff --git a/README.md b/README.md index 8ec16b4e..eafdaf9f 100644 --- a/README.md +++ b/README.md @@ -60,6 +60,48 @@ See [SYSTEM_REQUIREMENTS.md](SYSTEM_REQUIREMENTS.md) for the measured memory, disk, and CPU footprint of developing and building locally (and what to budget on a build machine — RAM is the binding constraint). +## Release-gate preflight + +Several release gates fail on facts that are already determined by the base +pool, the frozen selection, and the target/coverage registry — no calibration +solve needed to see them. `tools/preflight_us_release_gates.py` recovers those +signals in **minutes** so a two-hour release launch is not the first place a +knowable defect surfaces: + +```bash +uv run python tools/preflight_us_release_gates.py \ + --base-h5 out/base-m/base_populace_us_2024_puf_support.h5 \ + --selection-source-manifest inputs/buildm_keogh_swap_selection_source.json \ + --export-input-mass-reference-h5 forensics/populace_us_2024.h5 +``` + +It is read-only against the H5 artifacts and reports, per check, `PASS` / +`FAIL` / `AT-RISK` with the measured numbers (exit `1` on any FAIL, `2` on +AT-RISK only, `0` clean): + +1. **Selection carryover** — the frozen selection-source manifest maps cleanly + onto the base pool (the frozen-support recovery contract, run pre-solve). +2. **Zero-support preview** — compiled positive fiscal targets whose + materialized support is ~0 under the selection at base weights stay a + structural zero after the solve. Direct-column targets are checked; + engine-derived measures are marked *not statically checkable* (pass + `--ledger-facts` to compile the target surface). +3. **Export-mass parity risk** — each export-mass column's pool mass at *base* + weights against its reference band, honoring the release tool's + `US_EXPORT_INPUT_MASS_REVIEWED_EXCLUSIONS` register (reused, never + re-declared). A column out of band pre-solve is flagged for review. +4. **Smoke-probe support audit** — every reform-coverage probe leaf's pool vs + *selected* nonzero support and pool sign-leg decomposition. A leaf with pool + support but zero selected support **fails** (the input the frozen selection + cannot express); a thin selection or a signed leaf whose net sign + contradicts the probe's `expected_sign` is AT-RISK. + +**Run it** at base-build exit, before any release launch, and after any change +to the selection-source manifest or the target/coverage registry. The +synthetic-fixture unit tests +(`packages/populace-build/tests/test_us_release_gate_preflight.py`) run in the +normal `uv run pytest` suite; the real-H5 mode above is a local/runbook step. + ## Releasing & alerts Publishing uploads the locally built `releases//` artifacts to the Hugging diff --git a/packages/populace-build/src/populace/build/us_runtime/release_gate_preflight.py b/packages/populace-build/src/populace/build/us_runtime/release_gate_preflight.py new file mode 100644 index 00000000..97ad66a7 --- /dev/null +++ b/packages/populace-build/src/populace/build/us_runtime/release_gate_preflight.py @@ -0,0 +1,907 @@ +"""Statically-knowable US release-gate failures, computed without a solve. + +Build M release attempts 9/10/11 each burned ~2h of calibration to surface ONE +gate-group failure that was already determined by the *inputs* — the base pool +values, the frozen selection, and the target/coverage registry — and needed no +weight solve to see. This module recovers those signals in minutes so a release +launch is not the first place a knowable defect is caught. + +Four checks, each mirroring a live release gate but evaluated pre-solve: + +1. **Selection carryover** — does the frozen selection-source manifest map + cleanly onto the base pool? (Reuses + :meth:`~populace.build.us_runtime.warm_start_selection.SelectionSource.base_selection_mask` + in ``frozen_support`` mode: an unmapped/ambiguous/non-unique key is the same + hard failure the release refuses on, found in seconds instead of after a + base rebuild.) + +2. **Zero-support preview** — for every compiled *positive* fiscal target, its + materialized support under the selection at base weights. A positive target + whose estimate is ~0 pre-solve has no records to move and stays a structural + zero after calibration (the release ``zero_support`` gate's exact condition, + ``initial_estimate ~ final_estimate ~ 0``). Direct-column targets are checked + by reusing the calibrate constraint compiler + (:func:`populace.calibrate.matrix.build_constraint_matrix`); targets whose + measure is engine-derived (needs a microsim materialization pass) are marked + explicitly as *not statically checkable* rather than silently passed. + +3. **Export-mass parity risk** — for every column the export-mass parity gate + checks, the pool mass at *base* weights against the reference band + (``reference ± relative_tolerance``, honoring the release tool's + ``US_EXPORT_INPUT_MASS_REVIEWED_EXCLUSIONS`` register and the + ``minimum_reference_total`` floor). A column already outside its band at base + weights is an AT-RISK signal: the solve optimizes the target surface, not + this untargeted column, so a pre-solve out-of-band mass usually stays out. + The band math is reused verbatim via + :func:`populace.build.gates.input_mass_parity_gate`. + +4. **Smoke-probe support audit** — for every reform-coverage probe, for every + ``binding_inputs`` leaf: pool nonzero support, *selected* nonzero support + (via the frame's declared entity->household linkage), selected weighted mass, + and the pool sign-leg decomposition (positive/negative weighted). A leaf with + pool support but **zero selected support** is the ``keogh_distributions`` + restoration-vs-frozen-selection failure (populace#434): the smoke gate's only + teeth, computed as a set intersection instead of after a solve. A thin + selection (<5 records) or a signed leaf whose net sign contradicts the + probe's declared ``expected_sign`` (the ``farm_operations_income`` net-+$10.4B + Schedule-F-loss-heavy defect, populace#432) is AT-RISK. + +The pure check functions take frames and already-loaded registry material so they +unit-test on tiny synthetic fixtures with no policyengine-us and no large H5. +:func:`run_preflight` is the runbook entry point that loads the real artifacts +and reuses the release tool's own register and engine input-variable surface. +""" + +from __future__ import annotations + +import sys +from collections.abc import Iterable, Mapping, Sequence +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any, Literal + +import numpy as np +import pandas as pd + +from populace.build.gates import input_mass_parity_gate +from populace.build.us_runtime.input_mass import us_input_mass_totals +from populace.build.us_runtime.release_input_coverage import ( + ReformCoverageProbe, + us_release_reform_coverage_probes, +) +from populace.build.us_runtime.warm_start_selection import ( + SelectionSource, + load_selection_source_from_manifest, +) +from populace.calibrate.matrix import build_constraint_matrix +from populace.calibrate.registry import TargetRegistry, TargetSpec +from populace.frame import Frame + +__all__ = [ + "CheckResult", + "PreflightReport", + "check_export_mass_parity_risk", + "check_selection_carryover", + "check_smoke_probe_support", + "check_zero_support_preview", + "run_preflight", + "selected_household_ids", +] + +#: A check's verdict. +PreflightStatus = Literal["PASS", "FAIL", "AT_RISK", "SKIPPED"] + +#: An estimate at or below this absolute value is treated as zero support — +#: the same magnitude the release ``zero_support`` gate uses for a target's +#: initial and final estimate. +_ZERO_ESTIMATE_ATOL = 1e-9 + +#: A probe leaf with fewer than this many selected nonzero records is AT-RISK: +#: the selection can express the input, but so thinly that a tail-heavy column's +#: mass is one record away from a structural zero (populace#434's 4-of-13 case). +_MIN_SELECTED_RECORDS = 5 + +#: A column is "materially signed" — a genuine positive/negative column rather +#: than a one-signed amount that merely has a stray outlier — when its minority +#: leg carries at least this share of its majority leg's weighted mass. Only +#: materially-signed columns are eligible for the sign-contradiction check, so a +#: one-signed deduction leaf (e.g. a refund amount) is never a false positive. +_CANCELLATION_LEG_RATIO = 0.15 + +#: The export-mass parity gate defaults, matching the release tool CLI +#: (``--input-mass-relative-tolerance`` / ``--input-mass-minimum-reference-total``). +_DEFAULT_RELATIVE_TOLERANCE = 0.5 +_DEFAULT_MINIMUM_REFERENCE_TOTAL = 1e9 + +_STATUS_RANK = {"PASS": 0, "SKIPPED": 0, "AT_RISK": 1, "FAIL": 2} + + +@dataclass(frozen=True) +class CheckResult: + """One preflight check's verdict and the numbers behind it. + + Attributes: + name: Stable check id. + status: ``PASS`` / ``FAIL`` / ``AT_RISK`` / ``SKIPPED``. + summary: One-line human summary with the headline number. + failures: Hard-failure lines (drive exit 1). + at_risks: Advisory lines (drive exit 2 when no failure). + rows: Per-item measured numbers, JSON-ready, for the report table. + details: Extra machine-readable context. + """ + + name: str + status: PreflightStatus + summary: str + failures: tuple[str, ...] = () + at_risks: tuple[str, ...] = () + rows: tuple[Mapping[str, Any], ...] = () + details: Mapping[str, Any] = field(default_factory=dict) + + def to_dict(self) -> dict[str, Any]: + return { + "name": self.name, + "status": self.status, + "summary": self.summary, + "failures": list(self.failures), + "at_risks": list(self.at_risks), + "rows": [dict(row) for row in self.rows], + "details": dict(self.details), + } + + +@dataclass(frozen=True) +class PreflightReport: + """Every check's verdict plus the run's inputs, and the process exit code.""" + + checks: tuple[CheckResult, ...] + inputs: Mapping[str, Any] = field(default_factory=dict) + + @property + def status(self) -> PreflightStatus: + """The worst check status: FAIL > AT_RISK > (PASS/SKIPPED).""" + worst = max((_STATUS_RANK[c.status] for c in self.checks), default=0) + for status, rank in (("FAIL", 2), ("AT_RISK", 1)): + if worst == rank: + return status # type: ignore[return-value] + return "PASS" + + @property + def exit_code(self) -> int: + """1 on any FAIL, 2 on AT-RISK only, 0 clean.""" + if any(c.status == "FAIL" for c in self.checks): + return 1 + if any(c.status == "AT_RISK" for c in self.checks): + return 2 + return 0 + + def to_dict(self) -> dict[str, Any]: + return { + "status": self.status, + "exit_code": self.exit_code, + "inputs": dict(self.inputs), + "checks": [check.to_dict() for check in self.checks], + } + + def human_table(self) -> str: + """A compact, terminal-friendly rendering of every check.""" + lines: list[str] = [] + badge = { + "PASS": "PASS ", + "FAIL": "FAIL ", + "AT_RISK": "AT-RISK", + "SKIPPED": "SKIPPED", + } + lines.append("US release-gate preflight (no calibration solve)") + lines.append("=" * 72) + for check in self.checks: + lines.append(f"[{badge[check.status]}] {check.name}") + lines.append(f" {check.summary}") + for failure in check.failures: + lines.append(f" FAIL: {failure}") + for at_risk in check.at_risks: + lines.append(f" RISK: {at_risk}") + lines.append("-" * 72) + lines.append( + f"Overall: {self.status} (exit {self.exit_code}) " + f"[FAIL=1, AT-RISK=2, clean=0]" + ) + return "\n".join(lines) + + +# --------------------------------------------------------------------------- +# Shared linkage helpers +# --------------------------------------------------------------------------- + + +def selected_household_ids(base_frame: Frame, household_mask: np.ndarray) -> np.ndarray: + """The household ids the selection mask marks, in household-table order.""" + household = base_frame.table("household") + id_column = base_frame.schema.id_column("household") + return household[id_column].to_numpy()[np.asarray(household_mask, dtype=bool)] + + +def _selected_entity_mask( + frame: Frame, entity: str, selected_ids: np.ndarray +) -> np.ndarray: + """Boolean mask over ``entity``'s rows: is each row in a selected household? + + The selection is at household granularity and every populace US entity nests + inside exactly one household, so a group entity is selected iff its household + is. The mapping uses only the schema's declared id / membership columns + (``household_id``, ``person_household_id``, ``{group}_id``, + ``person_{group}_id``) — never a guessed name. + """ + schema = frame.schema + selected = np.asarray(selected_ids) + if entity == "household": + ids = frame.table("household")[schema.id_column("household")].to_numpy() + return np.isin(ids, selected) + if entity == schema.person_entity: + person_hh = frame.table("person")[ + schema.membership_column("household") + ].to_numpy() + return np.isin(person_hh, selected) + # A group entity: map each group id to its household through the persons + # (every group nests in exactly one household), then test membership. + person = frame.table("person") + group_to_household = ( + pd.DataFrame( + { + "group": person[schema.membership_column(entity)].to_numpy(), + "household": person[schema.membership_column("household")].to_numpy(), + } + ) + .drop_duplicates("group") + .set_index("group")["household"] + ) + group_ids = frame.table(entity)[schema.id_column(entity)].to_numpy() + group_household = group_to_household.reindex(group_ids).to_numpy() + return np.isin(group_household, selected) + + +def _numeric_column(series: pd.Series) -> np.ndarray: + """A leaf column as float64: booleans become their 1.0/0.0 mass.""" + if pd.api.types.is_bool_dtype(series): + return series.fillna(False).to_numpy(dtype=np.float64) + return pd.to_numeric(series, errors="coerce").fillna(0.0).to_numpy(dtype=np.float64) + + +# --------------------------------------------------------------------------- +# Check 1: selection carryover +# --------------------------------------------------------------------------- + + +def check_selection_carryover( + base_frame: Frame, selection_source: SelectionSource +) -> tuple[CheckResult, np.ndarray | None]: + """Does the frozen selection map cleanly onto the base pool (no solve)? + + Reuses the frozen-support recovery contract directly: any unmapped, + ambiguous, or non-unique-key identity raises the same hard error the release + refuses on, surfaced here in seconds. Returns the household selection mask + (or ``None`` on failure) so the smoke-probe audit can reuse it. + """ + try: + mask, report = selection_source.base_selection_mask( + base_frame, mode="frozen_support" + ) + except ValueError as exc: + return ( + CheckResult( + name="selection_carryover", + status="FAIL", + summary=( + "selection-source manifest does NOT map cleanly onto the base " + "pool (the release would refuse this base)" + ), + failures=(str(exc),), + details={"mode": "frozen_support"}, + ), + None, + ) + + n_selected = int(report.n_selected) + n_source = int(report.n_source) + clean = ( + n_selected == n_source and report.n_unmapped == 0 and report.n_ambiguous == 0 + ) + row = { + "n_source_identities": n_source, + "n_selected": n_selected, + "n_base_candidates": int(report.n_base_candidates), + "n_unmapped": int(report.n_unmapped), + "n_ambiguous": int(report.n_ambiguous), + } + return ( + CheckResult( + name="selection_carryover", + status="PASS" if clean else "FAIL", + summary=( + f"{n_selected}/{n_source} selection identities map onto " + f"{report.n_base_candidates} base households (frozen_support)" + ), + failures=() + if clean + else ( + f"{report.n_unmapped} unmapped / {report.n_ambiguous} ambiguous " + "identities under the join key.", + ), + rows=(row,), + details={"join_key": list(selection_source.join_key), **row}, + ), + mask, + ) + + +# --------------------------------------------------------------------------- +# Check 2: zero-support preview +# --------------------------------------------------------------------------- + + +def _positive_specs(target_specs: Iterable[TargetSpec]) -> list[TargetSpec]: + return [spec for spec in target_specs if float(spec.value) > 0.0] + + +def check_zero_support_preview( + selected_frame: Frame, + target_specs: Iterable[TargetSpec], + *, + weight_entity: str = "household", + country: str = "us", +) -> CheckResult: + """Compiled positive targets whose materialized support is ~0 pre-solve. + + A positive target whose estimate at base weights is ~0 has no records for + calibration to move, so it stays a structural zero after the solve — exactly + the release ``zero_support`` gate's condition, knowable now. Targets whose + measure/filter are direct columns of ``selected_frame`` are checked by + compiling the calibrate constraint system (the release's own materialization + algebra); targets whose measure is engine-derived (needs a microsim pass) are + reported as *not statically checkable* rather than silently passed. + """ + positive = _positive_specs(target_specs) + if not positive: + return CheckResult( + name="zero_support_preview", + status="PASS", + summary="no positive fiscal targets to preview", + details={"positive_targets": 0}, + ) + + registry = TargetRegistry(positive, country=country) + try: + problem = build_constraint_matrix( + selected_frame, registry.to_target_set(), weight_entity=weight_entity + ) + compiled_targets = problem.targets + skipped = problem.skipped + estimates = problem.estimates(problem.initial_weights.values) + row_nnz = np.diff(problem.matrix.indptr) + except ValueError: + # Every positive target's measure is engine-derived on this raw frame: + # nothing is statically checkable without a materialization pass. + compiled_targets = () + skipped = tuple(_SkippedLike(target=spec) for spec in registry.to_target_set()) + estimates = np.zeros(0) + row_nnz = np.zeros(0, dtype=int) + + failures: list[str] = [] + rows: list[dict[str, Any]] = [] + for index, target in enumerate(compiled_targets): + estimate = float(estimates[index]) + support = int(row_nnz[index]) + zero_support = support == 0 or abs(estimate) <= _ZERO_ESTIMATE_ATOL + rows.append( + { + "target": getattr(target, "name", str(target)), + "value": float(getattr(target, "value", 0.0)), + "checkable": True, + "support_records": support, + "base_weight_estimate": estimate, + "verdict": "zero_support" if zero_support else "supported", + } + ) + if zero_support: + failures.append( + f"{getattr(target, 'name', target)}: positive target " + f"(value {float(getattr(target, 'value', 0.0)):,.0f}) materializes " + f"{support} support record(s) and a base-weight estimate of " + f"{estimate:,.2f} under the selection — a structural zero the " + "solve cannot move." + ) + + not_checkable = 0 + for skip in skipped: + target = getattr(skip, "target", skip) + not_checkable += 1 + rows.append( + { + "target": getattr(target, "name", str(target)), + "value": float(getattr(target, "value", 0.0)), + "checkable": False, + "verdict": "not_statically_checkable", + } + ) + + status: PreflightStatus = "FAIL" if failures else "PASS" + checkable = len(compiled_targets) + summary = ( + f"{len(failures)} zero-support of {checkable} statically-checkable " + f"positive targets ({not_checkable} derived/reform-materialized, not " + "statically checkable)" + ) + return CheckResult( + name="zero_support_preview", + status=status, + summary=summary, + failures=tuple(failures), + rows=tuple(rows), + details={ + "positive_targets": len(positive), + "statically_checkable": checkable, + "not_statically_checkable": not_checkable, + "zero_support": len(failures), + }, + ) + + +@dataclass(frozen=True) +class _SkippedLike: + """Uniform ``.target`` holder for the all-skipped fallback path.""" + + target: Any + + +# --------------------------------------------------------------------------- +# Check 3: export-mass parity risk (at base weights) +# --------------------------------------------------------------------------- + + +def check_export_mass_parity_risk( + pool_totals: Mapping[str, float], + reference_totals: Mapping[str, float], + *, + relative_tolerance: float = _DEFAULT_RELATIVE_TOLERANCE, + minimum_reference_total: float = _DEFAULT_MINIMUM_REFERENCE_TOTAL, + reviewed_exclusions: Mapping[str, str] | None = None, +) -> CheckResult: + """Pool mass at base weights vs the export-mass reference band. + + Runs the release export-mass parity gate's *own* band math + (:func:`populace.build.gates.input_mass_parity_gate`) with the pre-solve pool + totals as the candidate. A column outside its ``reference ± tolerance`` band + at base weights is AT-RISK — the solve targets the calibration surface, not + this untargeted column, so a pre-solve out-of-band mass is unlikely to land + in-band. Columns in ``reviewed_exclusions`` are reported as excluded and + never escalate (they are the documented, base-rebuild-tracked drifts). + """ + reviewed_exclusions = dict(reviewed_exclusions or {}) + gate = input_mass_parity_gate( + pool_totals, + reference_totals, + candidate_name="base_pool_at_base_weights", + reference_name="export_mass_reference", + relative_tolerance=relative_tolerance, + minimum_reference_total=minimum_reference_total, + reviewed_exclusions=reviewed_exclusions, + ) + + # gate.failures name the checked columns whose pre-solve pool mass is out of + # band. Pre-solve these are risks, not release failures. + out_of_band = {failure.split(":", 1)[0] for failure in gate.failures} + + rows: list[dict[str, Any]] = [] + at_risks: list[str] = [] + in_band = 0 + below_floor = 0 + excluded = 0 + for name in sorted(reference_totals): + reference = float(reference_totals[name]) + pool = float(pool_totals.get(name, 0.0)) + drift = ( + (pool - reference) / abs(reference) if reference != 0.0 else float("inf") + ) + low = reference - relative_tolerance * abs(reference) + high = reference + relative_tolerance * abs(reference) + if abs(reference) <= minimum_reference_total: + verdict = "below_reference_floor" + below_floor += 1 + elif name in reviewed_exclusions: + verdict = "excluded" + excluded += 1 + elif name in out_of_band: + verdict = "out_of_band_at_base_weights" + at_risks.append( + f"{name}: pool mass {pool:,.0f} at base weights is outside the " + f"band [{low:,.0f}, {high:,.0f}] (reference {reference:,.0f}, " + f"drift {drift:+.1%}); confirm the calibration surface targets " + "this column — a targeted column is lifted into band by the " + "solve, an untargeted one fails the export-mass parity gate " + "(the rental_income/populace#432 class)." + ) + else: + verdict = "in_band" + in_band += 1 + rows.append( + { + "column": name, + "reference_mass": reference, + "pool_mass_at_base_weights": pool, + "band_low": low, + "band_high": high, + "drift": drift, + "verdict": verdict, + } + ) + + status: PreflightStatus = "AT_RISK" if at_risks else "PASS" + summary = ( + f"{len(at_risks)} column(s) out of band at base weights, {in_band} " + f"in-band, {excluded} excluded, {below_floor} below reference floor" + ) + return CheckResult( + name="export_mass_parity_risk", + status=status, + summary=summary, + at_risks=tuple(at_risks), + rows=tuple(rows), + details={ + "relative_tolerance": float(relative_tolerance), + "minimum_reference_total": float(minimum_reference_total), + "in_band": in_band, + "out_of_band_at_base_weights": len(at_risks), + "excluded": excluded, + "below_reference_floor": below_floor, + }, + ) + + +# --------------------------------------------------------------------------- +# Check 4: smoke-probe support audit +# --------------------------------------------------------------------------- + + +def _leaf_entity(frame: Frame, leaf: str) -> str | None: + try: + return frame.column_entity(leaf) + except ValueError: + return None + + +def _sign_legs(values: np.ndarray, weights: np.ndarray) -> tuple[float, float]: + """Weighted (positive_leg, negative_leg) of a signed column; N is <= 0.""" + positive = float(values[values > 0] @ weights[values > 0]) + negative = float(values[values < 0] @ weights[values < 0]) + return positive, negative + + +def check_smoke_probe_support( + base_frame: Frame, + selected_ids: np.ndarray, + probes: Iterable[ReformCoverageProbe], + *, + min_selected_records: int = _MIN_SELECTED_RECORDS, + cancellation_leg_ratio: float = _CANCELLATION_LEG_RATIO, +) -> CheckResult: + """Reform-coverage probe leaves without selection support, or wrong-signed. + + For every probe leaf: pool nonzero support, *selected* nonzero support (via + the frame's declared entity->household linkage), selected weighted mass, and + the pool sign-leg decomposition. + + - **FAIL** when a leaf has pool support but zero selected support — an input + restored after the selection froze, invisible to every count-based gate + (populace#434, ``keogh_distributions``). + - **AT-RISK** when selected support is thin (<``min_selected_records``), or + when a materially-signed leaf's net weighted sign contradicts its probe's + declared ``expected_sign`` (populace#432, ``farm_operations_income`` + imputed net-positive where the Schedule-F instrument is loss-heavy). + """ + probes = tuple(probes) + selected_ids = np.asarray(selected_ids) + + entity_selected: dict[str, np.ndarray] = {} + entity_weights: dict[str, np.ndarray] = {} + + failures: list[str] = [] + at_risks: list[str] = [] + rows: list[dict[str, Any]] = [] + + for probe in probes: + expected_positive = probe.expected_sign == "positive" + for leaf in probe.binding_inputs: + entity = _leaf_entity(base_frame, leaf) + if entity is None: + rows.append( + { + "probe": probe.id, + "leaf": leaf, + "verdict": "absent_from_base_pool", + } + ) + continue + if entity not in entity_selected: + entity_selected[entity] = _selected_entity_mask( + base_frame, entity, selected_ids + ) + entity_weights[entity] = np.asarray( + base_frame.resolve_weights(entity).values, dtype=np.float64 + ) + selected_mask = entity_selected[entity] + weights = entity_weights[entity] + values = _numeric_column(base_frame.table(entity)[leaf]) + + nonzero = values != 0.0 + pool_support = int(nonzero.sum()) + selected_nonzero = nonzero & selected_mask + selected_support = int(selected_nonzero.sum()) + selected_mass = float(values[selected_nonzero] @ weights[selected_nonzero]) + positive_leg, negative_leg = _sign_legs(values, weights) + net = positive_leg + negative_leg + + verdict = "supported" + if pool_support > 0 and selected_support == 0: + verdict = "structural_zero_in_selection" + failures.append( + f"{probe.id}/{leaf}: {pool_support} pool carrier(s) but 0 in " + "the selection — an input the frozen selection cannot express; " + "the reform-coverage smoke probe will score a structural zero " + f"(issue {probe.issue})." + ) + elif 0 < selected_support < min_selected_records: + verdict = "thin_selection" + at_risks.append( + f"{probe.id}/{leaf}: only {selected_support} selected carrier(s) " + f"(pool has {pool_support}); a tail-heavy column is one record " + "from a structural zero under this selection." + ) + + # Sign-structure contradiction, for genuinely-signed columns only. + majority = max(positive_leg, -negative_leg) + minority = min(positive_leg, -negative_leg) + materially_signed = majority > 0 and minority / majority >= ( + cancellation_leg_ratio + ) + net_positive = net > 0 + if ( + materially_signed + and pool_support > 0 + and net != 0.0 + and net_positive != expected_positive + ): + if verdict == "supported": + verdict = "sign_structure_contradicts_probe" + at_risks.append( + f"{probe.id}/{leaf}: pool sign legs +{positive_leg:,.0f} / " + f"{negative_leg:,.0f} net {net:+,.0f} — a " + f"{'positive' if net_positive else 'negative'} net that " + f"contradicts the probe's expected_sign=" + f"{probe.expected_sign!r} (issue {probe.issue})." + ) + + rows.append( + { + "probe": probe.id, + "leaf": leaf, + "entity": entity, + "pool_support": pool_support, + "selected_support": selected_support, + "selected_weighted_mass": selected_mass, + "pool_positive_leg": positive_leg, + "pool_negative_leg": negative_leg, + "pool_net": net, + "expected_sign": probe.expected_sign, + "verdict": verdict, + } + ) + + if failures: + status = "FAIL" + elif at_risks: + status = "AT_RISK" + else: + status = "PASS" + summary = ( + f"{len(failures)} leaf(s) with pool support but zero selected support, " + f"{len(at_risks)} at-risk (thin selection / sign structure)" + ) + return CheckResult( + name="smoke_probe_support", + status=status, + summary=summary, + failures=tuple(failures), + at_risks=tuple(at_risks), + rows=tuple(rows), + details={ + "probes": len(probes), + "min_selected_records": int(min_selected_records), + "cancellation_leg_ratio": float(cancellation_leg_ratio), + }, + ) + + +# --------------------------------------------------------------------------- +# Runbook orchestrator (loads real artifacts; reuses the release tool register) +# --------------------------------------------------------------------------- + + +def _release_tool_module() -> Any: + """Import ``build_us_fiscal_refresh_release`` (the release tool) by locating + the repo ``tools/`` directory relative to this file. + + Keeps the export-mass register and the engine input-variable surface single- + sourced in the release tool — this preflight never re-declares them. + """ + repo_root = Path(__file__).resolve() + for parent in repo_root.parents: + candidate = parent / "tools" / "build_us_fiscal_refresh_release.py" + if candidate.exists(): + if str(candidate.parent) not in sys.path: + sys.path.insert(0, str(candidate.parent)) + import build_us_fiscal_refresh_release as release # type: ignore + + return release + raise ModuleNotFoundError( + "Could not locate tools/build_us_fiscal_refresh_release.py to reuse the " + "export-mass register; pass engine_input_variables and " + "export_mass_reviewed_exclusions explicitly." + ) + + +def _load_ledger_target_specs( + ledger_facts: Path, + *, + target_period: int | str, + ledger_facts_sha256: str | None, +) -> tuple[TargetSpec, ...]: + from populace.build.ledger_artifact import load_ledger_consumer_artifact + from populace.build.us_runtime.fiscal_targets import ( + compile_us_fiscal_target_registry, + ) + + facts = load_ledger_consumer_artifact( + ledger_facts, expected_facts_sha256=ledger_facts_sha256 + ) + registry = compile_us_fiscal_target_registry(facts, target_period=target_period) + return registry.specs + + +def run_preflight( + *, + base_h5: str | Path, + selection_source_manifest: str | Path, + export_input_mass_reference_h5: str | Path | None = None, + ledger_facts: str | Path | None = None, + ledger_facts_sha256: str | None = None, + target_period: int | str = 2024, + relative_tolerance: float = _DEFAULT_RELATIVE_TOLERANCE, + minimum_reference_total: float = _DEFAULT_MINIMUM_REFERENCE_TOTAL, + probes: Iterable[ReformCoverageProbe] | None = None, + engine_input_variables: Sequence[str] | None = None, + export_mass_reviewed_exclusions: Mapping[str, str] | None = None, +) -> PreflightReport: + """Load the real artifacts and run every preflight check (no solve). + + Read-only against ``base_h5`` / ``export_input_mass_reference_h5``. The + export-mass register and engine input-variable surface default to the + release tool's own definitions (single-sourced, never re-declared here). + """ + from populace.build.us_runtime.l0_refit_export import load_us_frame + + base_h5 = Path(base_h5) + base_frame = load_us_frame(base_h5) + selection_source = load_selection_source_from_manifest(selection_source_manifest) + + checks: list[CheckResult] = [] + + carryover, mask = check_selection_carryover(base_frame, selection_source) + checks.append(carryover) + + selected = ( + selected_household_ids(base_frame, mask) if mask is not None else np.array([]) + ) + selected_frame = ( + base_frame.select(_household_person_mask(base_frame, mask)) + if mask is not None + else None + ) + + # Check 2: zero-support preview (requires the compiled fiscal-target surface). + if ledger_facts is not None and selected_frame is not None: + target_specs = _load_ledger_target_specs( + Path(ledger_facts), + target_period=target_period, + ledger_facts_sha256=ledger_facts_sha256, + ) + checks.append(check_zero_support_preview(selected_frame, target_specs)) + else: + checks.append( + CheckResult( + name="zero_support_preview", + status="SKIPPED", + summary=( + "no --ledger-facts feed provided; the compiled fiscal-target " + "surface is unavailable so zero-support cannot be previewed" + ), + details={"reason": "ledger_facts_not_provided"}, + ) + ) + + # Check 3: export-mass parity risk (requires a reference H5). + if export_input_mass_reference_h5 is not None: + if engine_input_variables is None or export_mass_reviewed_exclusions is None: + release = _release_tool_module() + if engine_input_variables is None: + engine_input_variables = release._engine_input_variables() + if export_mass_reviewed_exclusions is None: + export_mass_reviewed_exclusions = ( + release.US_EXPORT_INPUT_MASS_REVIEWED_EXCLUSIONS + ) + reference_frame = load_us_frame(Path(export_input_mass_reference_h5)) + pool_totals = us_input_mass_totals(base_frame, columns=engine_input_variables) + reference_totals = us_input_mass_totals( + reference_frame, columns=engine_input_variables + ) + checks.append( + check_export_mass_parity_risk( + pool_totals, + reference_totals, + relative_tolerance=relative_tolerance, + minimum_reference_total=minimum_reference_total, + reviewed_exclusions=export_mass_reviewed_exclusions, + ) + ) + else: + checks.append( + CheckResult( + name="export_mass_parity_risk", + status="SKIPPED", + summary="no --export-input-mass-reference-h5 provided", + details={"reason": "reference_h5_not_provided"}, + ) + ) + + # Check 4: smoke-probe support audit. + if mask is not None: + probe_set = tuple( + probes if probes is not None else us_release_reform_coverage_probes() + ) + checks.append(check_smoke_probe_support(base_frame, selected, probe_set)) + else: + checks.append( + CheckResult( + name="smoke_probe_support", + status="SKIPPED", + summary="selection did not map onto the base; cannot audit probes", + details={"reason": "selection_carryover_failed"}, + ) + ) + + inputs = { + "base_h5": str(base_h5), + "selection_source_manifest": str(selection_source_manifest), + "export_input_mass_reference_h5": ( + str(export_input_mass_reference_h5) + if export_input_mass_reference_h5 is not None + else None + ), + "ledger_facts": str(ledger_facts) if ledger_facts is not None else None, + "target_period": target_period, + "relative_tolerance": float(relative_tolerance), + "minimum_reference_total": float(minimum_reference_total), + } + return PreflightReport(checks=tuple(checks), inputs=inputs) + + +def _household_person_mask(base_frame: Frame, household_mask: np.ndarray) -> np.ndarray: + """Person mask for the households in ``household_mask`` (for ``Frame.select``).""" + schema = base_frame.schema + household = base_frame.table("household") + selected = household[schema.id_column("household")].to_numpy()[ + np.asarray(household_mask, dtype=bool) + ] + person_hh = base_frame.table("person")[ + schema.membership_column("household") + ].to_numpy() + return np.isin(person_hh, selected) diff --git a/packages/populace-build/tests/test_us_release_gate_preflight.py b/packages/populace-build/tests/test_us_release_gate_preflight.py new file mode 100644 index 00000000..5014c88c --- /dev/null +++ b/packages/populace-build/tests/test_us_release_gate_preflight.py @@ -0,0 +1,536 @@ +"""Release-gate preflight checks (populace#432/#434 defect classes). + +Every fixture is a tiny synthetic US-schema frame plus synthetic target/probe +material — no policyengine-us, no large H5 — so the suite runs in CI's base +environment. Each check's PASS and FAIL/AT-RISK paths are exercised, including +the two live Build M classes: + +- the ``keogh_distributions`` class (a probe leaf with pool support but zero + selected support), and +- the ``rental_income`` parity class (a column whose pool mass at base weights + is below its reference band). + +Each failure-path test asserts the specific verdict, so deleting a check's logic +(making it pass vacuously) breaks the corresponding test. +""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from populace.build.us_runtime.release_gate_preflight import ( + PreflightReport, + check_export_mass_parity_risk, + check_selection_carryover, + check_smoke_probe_support, + check_zero_support_preview, + selected_household_ids, +) +from populace.build.us_runtime.release_input_coverage import ReformCoverageProbe +from populace.build.us_runtime.warm_start_selection import ( + DEFAULT_SELECTION_JOIN_KEY, + SelectionSource, +) +from populace.calibrate.registry import TargetSpec +from populace.frame import US_SCHEMA, Frame, WeightKind, Weights + + +def _frame(households: list[dict[str, object]]) -> Frame: + """A minimal US frame from household specs (one person per household).""" + prows, hrows, spm_rows, mu_rows = [], [], [], [] + tu_rows, fam_rows = [], [] + pid = 1 + for hh in households: + hid = int(hh["hid"]) + prows.append( + { + "person_id": pid, + "person_household_id": hid, + "person_tax_unit_id": hid, + "person_spm_unit_id": hid, + "person_family_id": hid, + "person_marital_unit_id": pid, + "source_year": int(hh["syear"]), + "source_household_id": int(hh["shh"]), + "source_person_id": f"{hh['syear']}{int(hh['shh']):04d}{pid:02d}", + "person_support_channel": str(hh["chan"]), + "person_support_clone_index": int(hh["clone"]), + "keogh_distributions": float(hh.get("keogh", 0.0)), + "farm_operations_income": float(hh.get("farm", 0.0)), + } + ) + hrows.append( + { + "household_id": hid, + "household_support_channel": str(hh["chan"]), + "household_support_clone_index": int(hh["clone"]), + "state_return_count": float(hh.get("state_returns", 0.0)), + } + ) + spm_rows.append( + { + "spm_unit_id": hid, + "spm_unit_energy_subsidy": float(hh.get("energy", 0.0)), + } + ) + tu_rows.append({"tax_unit_id": hid}) + fam_rows.append({"family_id": hid}) + mu_rows.append({"marital_unit_id": pid}) + pid += 1 + weights = np.array([float(hh.get("w", 100.0)) for hh in households]) + return Frame( + { + "person": pd.DataFrame(prows), + "household": pd.DataFrame(hrows), + "tax_unit": pd.DataFrame(tu_rows), + "spm_unit": pd.DataFrame(spm_rows), + "family": pd.DataFrame(fam_rows), + "marital_unit": pd.DataFrame(mu_rows), + }, + US_SCHEMA, + {"household": Weights(weights, WeightKind.CALIBRATED)}, + ) + + +def _selection(picks: list[tuple]) -> SelectionSource: + return SelectionSource( + join_key=DEFAULT_SELECTION_JOIN_KEY, + identities=[list(p) for p in picks], + provenance={"kind": "test"}, + ) + + +def _probe( + *, + leaf: str, + expected_sign: str = "positive", + probe_id: str = "probe", +) -> ReformCoverageProbe: + return ReformCoverageProbe( + id=probe_id, + name=probe_id, + parameter_changes={}, + neutralized_variable=leaf, + budget_measure="income_tax", + binding_inputs=(leaf,), + min_abs_effect=1e6, + reason="synthetic", + issue="PolicyEngine/populace#test", + expected_sign=expected_sign, + ) + + +_POOL = [ + {"hid": 1, "syear": 2024, "shh": 11, "chan": "asec", "clone": 0, "keogh": 500.0}, + {"hid": 2, "syear": 2024, "shh": 22, "chan": "asec", "clone": 0, "keogh": 0.0}, + {"hid": 3, "syear": 2023, "shh": 11, "chan": "puf", "clone": 1, "keogh": 700.0}, + {"hid": 4, "syear": 2023, "shh": 44, "chan": "asec", "clone": 0, "keogh": 0.0}, +] + + +# --------------------------------------------------------------------------- +# Check 1: selection carryover +# --------------------------------------------------------------------------- + + +def test__selection_carryover__maps_cleanly__passes() -> None: + base = _frame(_POOL) + source = _selection([(2024, 11, "asec", 0), (2023, 11, "puf", 1)]) + + result, mask = check_selection_carryover(base, source) + + assert result.status == "PASS" + assert result.details["n_selected"] == 2 + assert result.details["n_unmapped"] == 0 + assert mask.tolist() == [True, False, True, False] + + +def test__selection_carryover__unmapped_identity__fails() -> None: + base = _frame(_POOL) + # (2024, 99, ...) is not in the base pool. + source = _selection([(2024, 11, "asec", 0), (2024, 99, "asec", 0)]) + + result, mask = check_selection_carryover(base, source) + + assert result.status == "FAIL" + assert mask is None + assert result.failures and "unmapped" in result.failures[0].lower() + + +# --------------------------------------------------------------------------- +# Check 2: zero-support preview +# --------------------------------------------------------------------------- + + +def _select_frame(base: Frame, source: SelectionSource) -> Frame: + mask, _ = source.base_selection_mask(base, mode="frozen_support") + hh_ids = selected_household_ids(base, mask) + person_hh = base.table("person")["person_household_id"].to_numpy() + return base.select(np.isin(person_hh, hh_ids)) + + +def test__zero_support_preview__direct_column_target_supported__passes() -> None: + pool = [ + {**hh, "state_returns": 1.0 if hh["hid"] in (1, 3) else 0.0} for hh in _POOL + ] + base = _frame(pool) + source = _selection([(2024, 11, "asec", 0), (2023, 11, "puf", 1)]) + selected = _select_frame(base, source) + + spec = TargetSpec( + name="state_returns_total", + entity="household", + value=100.0, + measure="state_return_count", + source="synthetic", + ) + result = check_zero_support_preview(selected, [spec]) + + assert result.status == "PASS" + (row,) = [r for r in result.rows if r["target"].startswith("state_returns_total")] + assert row["checkable"] is True + assert row["support_records"] == 2 + assert row["verdict"] == "supported" + + +def test__zero_support_preview__positive_target_zero_support__fails() -> None: + # Households 1 and 3 carry the state return indicator; select only #2 and #4. + pool = [ + {**hh, "state_returns": 1.0 if hh["hid"] in (1, 3) else 0.0} for hh in _POOL + ] + base = _frame(pool) + source = _selection([(2024, 22, "asec", 0), (2023, 44, "asec", 0)]) + selected = _select_frame(base, source) + + spec = TargetSpec( + name="state_returns_total", + entity="household", + value=100.0, + measure="state_return_count", + source="synthetic", + ) + result = check_zero_support_preview(selected, [spec]) + + assert result.status == "FAIL" + assert result.details["zero_support"] == 1 + assert result.failures and "structural zero" in result.failures[0] + + +def test__zero_support_preview__derived_measure__not_statically_checkable() -> None: + pool = [{**hh, "state_returns": 1.0} for hh in _POOL] + base = _frame(pool) + source = _selection([(2024, 11, "asec", 0)]) + selected = _select_frame(base, source) + + specs = [ + TargetSpec( + name="state_returns_total", + entity="household", + value=100.0, + measure="state_return_count", + source="synthetic", + ), + TargetSpec( + name="income_tax_total", + entity="household", + value=5000.0, + measure="income_tax", # engine-derived: not a base column + source="synthetic", + ), + ] + result = check_zero_support_preview(selected, specs) + + assert result.status == "PASS" # the derived one does not fail + assert result.details["not_statically_checkable"] == 1 + derived = [r for r in result.rows if r["target"].startswith("income_tax_total")] + assert derived and derived[0]["verdict"] == "not_statically_checkable" + + +# --------------------------------------------------------------------------- +# Check 3: export-mass parity risk +# --------------------------------------------------------------------------- + + +def test__export_mass_parity_risk__in_band__passes() -> None: + reference = {"colA": 100e9, "colB": 200e9} + pool = {"colA": 90e9, "colB": 220e9} # both within +/-50% + + result = check_export_mass_parity_risk(pool, reference) + + assert result.status == "PASS" + assert result.details["in_band"] == 2 + assert result.details["out_of_band_at_base_weights"] == 0 + + +def test__export_mass_parity_risk__pool_below_band_floor__at_risk() -> None: + # The rental_income class: pool mass at base weights far below the band. + reference = {"rental_like": 400e9} + pool = {"rental_like": 90e9} # -77.5%, below the [200e9, 600e9] band + + result = check_export_mass_parity_risk(pool, reference) + + assert result.status == "AT_RISK" + (row,) = result.rows + assert row["verdict"] == "out_of_band_at_base_weights" + assert row["pool_mass_at_base_weights"] < row["band_low"] + assert result.at_risks and "outside the band" in result.at_risks[0] + + +def test__export_mass_parity_risk__reviewed_exclusion__reported_not_at_risk() -> None: + reference = {"rental_income": 400e9} + pool = {"rental_income": 90e9} # out of band, but excluded + + result = check_export_mass_parity_risk( + pool, + reference, + reviewed_exclusions={"rental_income": "documented base-rebuild drift"}, + ) + + assert result.status == "PASS" + (row,) = result.rows + assert row["verdict"] == "excluded" + assert result.details["excluded"] == 1 + + +def test__export_mass_parity_risk__below_minimum_reference_total__skipped() -> None: + reference = {"tiny": 5e8} # below the 1e9 floor + pool = {"tiny": 0.0} + + result = check_export_mass_parity_risk(pool, reference) + + assert result.status == "PASS" + (row,) = result.rows + assert row["verdict"] == "below_reference_floor" + + +# --------------------------------------------------------------------------- +# Check 4: smoke-probe support audit +# --------------------------------------------------------------------------- + + +def test__smoke_probe_support__leaf_supported__passes() -> None: + base = _frame(_POOL) + source = _selection([(2024, 11, "asec", 0), (2023, 11, "puf", 1)]) + mask, _ = source.base_selection_mask(base, mode="frozen_support") + selected = selected_household_ids(base, mask) + + # keogh carriers are households 1 and 3; both selected -> supported. + result = check_smoke_probe_support( + base, + selected, + [_probe(leaf="keogh_distributions", probe_id="keogh")], + min_selected_records=1, + ) + + assert result.status == "PASS" + (row,) = [r for r in result.rows if r["leaf"] == "keogh_distributions"] + assert row["pool_support"] == 2 + assert row["selected_support"] == 2 + assert row["verdict"] == "supported" + + +def test__smoke_probe_support__pool_support_zero_selected__fails() -> None: + # The keogh/#434 class: carriers (hh 1, 3) exist in the pool but the + # selection (#2, #4) includes none of them. + base = _frame(_POOL) + source = _selection([(2024, 22, "asec", 0), (2023, 44, "asec", 0)]) + mask, _ = source.base_selection_mask(base, mode="frozen_support") + selected = selected_household_ids(base, mask) + + result = check_smoke_probe_support( + base, + selected, + [_probe(leaf="keogh_distributions", probe_id="keogh")], + ) + + assert result.status == "FAIL" + (row,) = [r for r in result.rows if r["leaf"] == "keogh_distributions"] + assert row["pool_support"] == 2 + assert row["selected_support"] == 0 + assert row["verdict"] == "structural_zero_in_selection" + assert result.failures and "0 in" in result.failures[0] + + +def test__smoke_probe_support__thin_selection__at_risk() -> None: + base = _frame(_POOL) + # Select only household 1 (one keogh carrier of two). + source = _selection([(2024, 11, "asec", 0)]) + mask, _ = source.base_selection_mask(base, mode="frozen_support") + selected = selected_household_ids(base, mask) + + result = check_smoke_probe_support( + base, + selected, + [_probe(leaf="keogh_distributions", probe_id="keogh")], + min_selected_records=5, + ) + + assert result.status == "AT_RISK" + (row,) = [r for r in result.rows if r["leaf"] == "keogh_distributions"] + assert row["verdict"] == "thin_selection" + + +def test__smoke_probe_support__signed_net_contradicts_expected_sign__at_risk() -> None: + # The farm/#432 class: a materially-signed leaf whose net weighted sign is + # positive while the probe declares expected_sign="negative". + pool = [ + {"hid": 1, "syear": 2024, "shh": 11, "chan": "asec", "clone": 0, "farm": 900.0}, + {"hid": 2, "syear": 2024, "shh": 22, "chan": "asec", "clone": 0, "farm": 800.0}, + {"hid": 3, "syear": 2023, "shh": 11, "chan": "puf", "clone": 1, "farm": -600.0}, + {"hid": 4, "syear": 2023, "shh": 44, "chan": "puf", "clone": 1, "farm": -400.0}, + ] + base = _frame(pool) + source = _selection( + [(2024, 11, "asec", 0), (2024, 22, "asec", 0), (2023, 11, "puf", 1)] + ) + mask, _ = source.base_selection_mask(base, mode="frozen_support") + selected = selected_household_ids(base, mask) + + result = check_smoke_probe_support( + base, + selected, + [ + _probe( + leaf="farm_operations_income", expected_sign="negative", probe_id="farm" + ) + ], + min_selected_records=1, + ) + + assert result.status == "AT_RISK" + (row,) = [r for r in result.rows if r["leaf"] == "farm_operations_income"] + assert row["pool_net"] > 0 + assert row["expected_sign"] == "negative" + assert row["verdict"] == "sign_structure_contradicts_probe" + assert result.at_risks and "contradicts" in result.at_risks[0] + + +def test__smoke_probe_support__one_signed_leaf__no_false_sign_flag() -> None: + # A one-signed (all-positive) leaf with expected_sign="negative" must NOT be + # flagged for a sign contradiction — only genuinely-signed columns are. + base = _frame(_POOL) # keogh_distributions is all-positive + source = _selection([(2024, 11, "asec", 0), (2023, 11, "puf", 1)]) + mask, _ = source.base_selection_mask(base, mode="frozen_support") + selected = selected_household_ids(base, mask) + + result = check_smoke_probe_support( + base, + selected, + [ + _probe( + leaf="keogh_distributions", expected_sign="negative", probe_id="keogh" + ) + ], + min_selected_records=1, + ) + + assert result.status == "PASS" + (row,) = [r for r in result.rows if r["leaf"] == "keogh_distributions"] + assert row["pool_negative_leg"] == 0.0 + assert row["verdict"] == "supported" + + +def test__smoke_probe_support__leaf_absent_from_base__reported_not_failed() -> None: + base = _frame(_POOL) + source = _selection([(2024, 11, "asec", 0)]) + mask, _ = source.base_selection_mask(base, mode="frozen_support") + selected = selected_household_ids(base, mask) + + result = check_smoke_probe_support( + base, + selected, + [_probe(leaf="bank_account_assets", probe_id="ssi")], + ) + + assert result.status == "PASS" + (row,) = [r for r in result.rows if r["leaf"] == "bank_account_assets"] + assert row["verdict"] == "absent_from_base_pool" + + +def test__smoke_probe_support__group_entity_leaf_linkage() -> None: + # An spm_unit-level leaf must map to households through the persons. + pool = [ + { + "hid": 1, + "syear": 2024, + "shh": 11, + "chan": "asec", + "clone": 0, + "energy": 50.0, + }, + { + "hid": 2, + "syear": 2024, + "shh": 22, + "chan": "asec", + "clone": 0, + "energy": 70.0, + }, + ] + base = _frame(pool) + source = _selection([(2024, 11, "asec", 0)]) # household 1 only + mask, _ = source.base_selection_mask(base, mode="frozen_support") + selected = selected_household_ids(base, mask) + + result = check_smoke_probe_support( + base, + selected, + [_probe(leaf="spm_unit_energy_subsidy", probe_id="energy")], + min_selected_records=1, + ) + + (row,) = [r for r in result.rows if r["leaf"] == "spm_unit_energy_subsidy"] + assert row["entity"] == "spm_unit" + assert row["pool_support"] == 2 + assert row["selected_support"] == 1 # only spm_unit of household 1 + + +# --------------------------------------------------------------------------- +# Report semantics +# --------------------------------------------------------------------------- + + +def test__report__exit_code_and_status_precedence() -> None: + from populace.build.us_runtime.release_gate_preflight import CheckResult + + clean = PreflightReport(checks=(CheckResult(name="a", status="PASS", summary=""),)) + assert clean.exit_code == 0 and clean.status == "PASS" + + at_risk = PreflightReport( + checks=( + CheckResult(name="a", status="PASS", summary=""), + CheckResult(name="b", status="AT_RISK", summary="", at_risks=("x",)), + ) + ) + assert at_risk.exit_code == 2 and at_risk.status == "AT_RISK" + + fail = PreflightReport( + checks=( + CheckResult(name="a", status="AT_RISK", summary="", at_risks=("x",)), + CheckResult(name="b", status="FAIL", summary="", failures=("y",)), + ) + ) + assert fail.exit_code == 1 and fail.status == "FAIL" + + +def test__report__to_dict_is_json_ready() -> None: + import json + + from populace.build.us_runtime.release_gate_preflight import CheckResult + + report = PreflightReport( + checks=( + CheckResult( + name="a", + status="FAIL", + summary="s", + failures=("f",), + rows=({"k": 1},), + ), + ), + inputs={"base_h5": "x"}, + ) + payload = report.to_dict() + assert json.loads(json.dumps(payload))["exit_code"] == 1 + assert payload["checks"][0]["rows"] == [{"k": 1}] diff --git a/tools/preflight_us_release_gates.py b/tools/preflight_us_release_gates.py new file mode 100644 index 00000000..0bc088a7 --- /dev/null +++ b/tools/preflight_us_release_gates.py @@ -0,0 +1,151 @@ +"""Preflight the US release gates that are statically knowable without a solve. + +Each Build M release attempt (9/10/11) spent ~2h of calibration to surface one +gate-group failure already determined by the base pool, the frozen selection, +and the registry. This tool recovers those signals in minutes and reports, per +check, PASS / FAIL / AT-RISK with the measured numbers. Run it: + +- when the base build exits (before launching a release), +- before any release launch, and +- after any change to the selection-source manifest or the target/coverage + registry. + +Example (read-only against the artifacts):: + + uv run python tools/preflight_us_release_gates.py \ + --base-h5 out/base-m/base_populace_us_2024_puf_support.h5 \ + --selection-source-manifest inputs/buildm_keogh_swap_selection_source.json \ + --export-input-mass-reference-h5 forensics/populace_us_2024.h5 + +Exit code: 1 on any FAIL, 2 on AT-RISK only, 0 clean. +""" + +from __future__ import annotations + +import argparse +import json +import sys +from pathlib import Path + +# Reuse the release tool's export-mass register and engine input-variable +# surface — never re-declared here (this file sits beside it in tools/). +sys.path.insert(0, str(Path(__file__).resolve().parent)) +sys.path.insert( + 0, str(Path(__file__).resolve().parents[1] / "packages" / "populace-build" / "src") +) + +from populace.build.us_runtime.release_gate_preflight import ( # noqa: E402 + run_preflight, +) + + +def _parser() -> argparse.ArgumentParser: + parser = argparse.ArgumentParser( + description=( + "Statically preview the US release gates (selection carryover, " + "zero-support, export-mass parity risk, reform-coverage smoke " + "support) without running a calibration solve." + ) + ) + parser.add_argument( + "--base-h5", + required=True, + type=Path, + help="Base pool H5 (read-only).", + ) + parser.add_argument( + "--selection-source-manifest", + required=True, + type=Path, + help="Frozen selection-source manifest JSON.", + ) + parser.add_argument( + "--export-input-mass-reference-h5", + type=Path, + default=None, + help=( + "Reference H5 for the export-mass parity band (read-only). Omit to " + "skip the parity-risk check." + ), + ) + parser.add_argument( + "--ledger-facts", + type=Path, + default=None, + help=( + "Optional Ledger consumer facts feed (directory or " + "consumer_facts.jsonl). Required to preview zero-support — the " + "compiled fiscal-target surface comes from it. Omitted -> the " + "zero-support check is SKIPPED." + ), + ) + parser.add_argument( + "--ledger-facts-sha256", + default=None, + help="Optional pin: expected SHA-256 of consumer_facts.jsonl.", + ) + parser.add_argument( + "--target-period", + default=2024, + help="Build period the fiscal targets are compiled for (default 2024).", + ) + parser.add_argument( + "--relative-tolerance", + type=float, + default=0.5, + help=( + "Export-mass parity band half-width (matches the release tool's " + "--input-mass-relative-tolerance; default 0.5)." + ), + ) + parser.add_argument( + "--minimum-reference-total", + type=float, + default=1e9, + help=( + "Reference-mass floor below which parity is not checked (matches " + "the release tool's --input-mass-minimum-reference-total; default " + "1e9)." + ), + ) + parser.add_argument( + "--json-out", + type=Path, + default=None, + help="Write the machine-readable report JSON to this path.", + ) + return parser + + +def main(argv: list[str] | None = None) -> int: + args = _parser().parse_args(argv) + try: + target_period: int | str = int(args.target_period) + except (TypeError, ValueError): + target_period = args.target_period + + report = run_preflight( + base_h5=args.base_h5, + selection_source_manifest=args.selection_source_manifest, + export_input_mass_reference_h5=args.export_input_mass_reference_h5, + ledger_facts=args.ledger_facts, + ledger_facts_sha256=args.ledger_facts_sha256, + target_period=target_period, + relative_tolerance=args.relative_tolerance, + minimum_reference_total=args.minimum_reference_total, + ) + + print(report.human_table()) + payload = report.to_dict() + if args.json_out is not None: + args.json_out.write_text(json.dumps(payload, indent=2, sort_keys=False)) + print(f"\nWrote machine-readable report to {args.json_out}") + else: + print("\n--- machine-readable report ---") + print(json.dumps(payload, indent=2, sort_keys=False)) + + return report.exit_code + + +if __name__ == "__main__": + raise SystemExit(main())