diff --git a/.github/workflows/reform-validation-backfill.yml b/.github/workflows/reform-validation-backfill.yml new file mode 100644 index 00000000..b230588d --- /dev/null +++ b/.github/workflows/reform-validation-backfill.yml @@ -0,0 +1,281 @@ +name: Reform validation backfill + +# Per-release producer for the populace/microcosm reform-validation registry, +# re-homed from calibration-diagnostics #112 (scorecard issue #15 call 3). The +# producer needs a full Microsimulation per batch plus the microcosm/torch +# import stack, which no longer fits GitHub's 7GB ubuntu-latest runner, so the +# simulation runs on Modal (tools/reform_validation/modal_backfill_app.py, 64GB +# container). This workflow is only the trigger/commit layer and every tick +# finishes in seconds: +# +# drain: any release with a published artifact on the Volume but no +# raw/.json (and no OPEN PR) is harvested — one per tick, +# oldest-first — landed under raw/, ingested, and PR'd, regardless +# of which release is currently `latest` (so a run that finished +# after latest advanced is never stranded). +# spawn: otherwise, every release that still needs work — the current +# release plus any stranded pending run (spawned, became non-latest, +# then died before publishing) — is (re-)spawned if its recorded +# Modal call isn't live. +# +# Liveness (tools/reform_validation/spawn_or_wait.py): the spawner records the +# Modal call id + release id on the Volume; each tick consults that call's real +# status before spawning. Marker reads FAIL CLOSED — an unreadable marker is a +# WAIT, never a "no run recorded" respawn — so a transport blip never +# double-spawns or re-pins live work, and a completed call re-checks the Volume +# for its artifact before respawning finished work. The producer ref AND the +# driver (scorecard) commit are pinned/stamped, and merge() refuses partials +# that mix or omit either revision, purging the workdir to recompute clean. +# +# The PR is opened for review, not auto-merged: a new release brings new engine +# pins and can shift the exact-count ingest tests, so a human confirms the DB +# delta before it lands. It records the scorecard commit + artifact sha256 for +# provenance; the ingest also consumes the artifact's `_attestation`. +# +# Requirements: +# - repository secrets MODAL_TOKEN_ID / MODAL_TOKEN_SECRET (PolicyEngine +# Modal workspace); +# - Settings -> Actions -> General -> "Allow GitHub Actions to create and +# approve pull requests" MUST be enabled, or the "Open PR" step 403s. +# +# Note: PolicyEngine/populace was renamed to PolicyEngine/microcosm (2026-08); +# the producer is cloned from microcosm and imported as `microcosm.*`. +# +# Run by hand for a specific release via the "Run workflow" button +# (release_id input), or leave it to the schedule. + +on: + schedule: + - cron: "0 */6 * * *" # every 6 hours + workflow_dispatch: + inputs: + release_id: + description: "Release id to backfill (blank = current latest)" + required: false + default: "" + +permissions: + contents: write + pull-requests: write + +concurrency: + group: reform-validation-backfill + cancel-in-progress: false + +env: + HF_BASE: https://huggingface.co/datasets/policyengine/populace-us/resolve/main + MODAL_VOLUME: scorecard-reform-validation + RAW_DIR: sources/populace-reform-validation/raw + PRODUCER_REPO: PolicyEngine/microcosm + # A release id and nothing that could carry shell/codegen metacharacters — + # this value reaches the shell and Modal while credentials are live, so it is + # allow-listed before any use (blocker 1). Accepts the current populace-us-* + # ids and the microcosm-us-* line the rename moves to. + RELEASE_ID_RE: '(populace|microcosm)-us-[0-9]{4}-[A-Za-z0-9._-]+' + +jobs: + resolve: + runs-on: ubuntu-latest + outputs: + release_id: ${{ steps.r.outputs.release_id }} + steps: + - uses: actions/checkout@v4 + - id: r + # The dispatch input is bound to an env var, never interpolated into + # the script body (blocker 1) — a `${{ }}` splice would let a crafted + # release_id inject shell. + env: + RELEASE_INPUT: ${{ github.event.inputs.release_id }} + run: | + set -euo pipefail + RID="$RELEASE_INPUT" + if [ -z "$RID" ]; then + RID=$(curl -sL "$HF_BASE/latest.json" \ + | python3 -c "import sys,json;print(json.load(sys.stdin)['release_id'])") + fi + # Reject multiline BEFORE the regex: `grep -qxE` matches if ANY line + # matches, so a value whose first line is valid would pass and then + # inject via the multiline `$GITHUB_OUTPUT` heredoc (blocker 1). + if [ "$(printf '%s' "$RID" | wc -l)" -ne 0 ]; then + echo "::error::release id must be a single line" + exit 1 + fi + if ! printf '%s' "$RID" | grep -qxE "$RELEASE_ID_RE"; then + echo "::error::release id '$RID' failed validation (expected ${RELEASE_ID_RE})" + exit 1 + fi + echo "release_id=$RID" >> "$GITHUB_OUTPUT" + + backfill: + needs: resolve + runs-on: ubuntu-latest + timeout-minutes: 30 + env: + RID: ${{ needs.resolve.outputs.release_id }} + steps: + - name: Checkout scorecard + uses: actions/checkout@v4 + + - uses: actions/setup-python@v5 + with: + python-version: "3.13" + + - name: Install Modal client + # No Modal credentials in this step: `pip install modal` builds + # third-party sdists (grpclib compiles from source), so the tokens must + # not be in-scope here (non-blocker). Pinned EXACTLY so a client + # release can't silently change spawn/poll behaviour under us — none of + # the Modal paths run in CI, so the pin is the only guard. + run: | + set -euo pipefail + python -m pip install --quiet 'modal==0.77.0' + + - name: Drain a finished artifact, else spawn + id: harvest + env: + MODAL_TOKEN_ID: ${{ secrets.MODAL_TOKEN_ID }} + MODAL_TOKEN_SECRET: ${{ secrets.MODAL_TOKEN_SECRET }} + GH_TOKEN: ${{ github.token }} + run: | + set -euo pipefail + if [ -z "$MODAL_TOKEN_ID" ] || [ -z "$MODAL_TOKEN_SECRET" ]; then + echo "::error::MODAL_TOKEN_ID / MODAL_TOKEN_SECRET are not set — the simulation runs on Modal and cannot proceed without them." + exit 1 + fi + + valid() { printf '%s' "$1" | grep -qxE "$RELEASE_ID_RE"; } + sha8() { RID_="$1" python3 -c 'import hashlib,os;print(hashlib.sha256(os.environ["RID_"].encode()).hexdigest()[:8])'; } + published() { python3 tools/reform_validation/spawn_or_wait.py artifacts | grep -qxF "$1"; } + # An OPEN PR (not mere branch existence) means a prior tick already + # harvested this release — gate on the PR so a failed/closed/collided + # branch doesn't strand it, and FAIL CLOSED: an errored probe returns + # "covered" so a transient gh failure never re-pushes (blocker 3). + open_pr() { + local n + if ! n=$(gh pr list --repo "$GITHUB_REPOSITORY" --head "reform-validation/$1" --state open --json number --jq 'length' 2>/dev/null); then + return 0 + fi + [ "${n:-1}" != "0" ] + } + + # 1) DRAIN one finished-but-unharvested artifact (oldest-first). + TARGET="" + for id in $(python3 tools/reform_validation/spawn_or_wait.py artifacts); do + valid "$id" || { echo "::warning::skipping malformed artifact id"; continue; } + [ -f "$RAW_DIR/$id.json" ] && continue + if open_pr "$id"; then + echo "::notice::open PR already covers $id — skipping." + continue + fi + TARGET="$id" + break + done + + if [ -n "$TARGET" ]; then + mkdir -p "$RUNNER_TEMP/harvest" + modal volume get "$MODAL_VOLUME" "reform_validation_$TARGET.json" "$RUNNER_TEMP/harvest/$TARGET.json" --force + # Cross-check the harvested artifact's Modal call id against the id + # the spawner recorded on the Volume — a mismatch means the artifact + # came from a run we didn't track, so refuse it (blocker 2). Both + # present required; the recorded id is read before the workdir goes. + if modal volume get "$MODAL_VOLUME" "rv_$(sha8 "$TARGET")/call_id" "$RUNNER_TEMP/recorded_call" --force 2>/dev/null; then + RECORDED=$(tr -d '\r\n' < "$RUNNER_TEMP/recorded_call") + ARTIFACT_CALL=$(python3 -c "import json,sys;print((json.load(open(sys.argv[1])).get('_attestation') or {}).get('modal_call_id',''))" "$RUNNER_TEMP/harvest/$TARGET.json") + if [ -n "$RECORDED" ] && [ -n "$ARTIFACT_CALL" ] && [ "$RECORDED" != "$ARTIFACT_CALL" ]; then + echo "::error::$TARGET Modal call id mismatch (artifact=$ARTIFACT_CALL recorded=$RECORDED) — refusing a run we didn't track." + exit 1 + fi + fi + # Drop the bulky workdir (H5 + partials); the published artifact + # stays on the Volume as the done-signal until the PR merges. + modal volume rm "$MODAL_VOLUME" "rv_$(sha8 "$TARGET")" -r || true + echo "harvested=$TARGET" >> "$GITHUB_OUTPUT" + echo "::notice::harvested $TARGET — ingesting and opening a PR." + exit 0 + fi + echo "harvested=" >> "$GITHUB_OUTPUT" + + # 2) Nothing to drain: ensure every release that still needs work is + # running — the current $RID plus any stranded pending run that + # became non-latest and died before publishing (blocker 3). Deploy + # the app once, only if we actually spawn. + CANDIDATES=$(printf '%s\n' "$RID"; python3 tools/reform_validation/spawn_or_wait.py pending) + DEPLOYED=0 + for cand in $(printf '%s\n' "$CANDIDATES" | awk 'NF' | sort -u); do + valid "$cand" || { echo "::warning::skipping malformed pending id"; continue; } + [ -f "$RAW_DIR/$cand.json" ] && continue + if published "$cand"; then + echo "::notice::$cand artifact is published — a later tick drains it." + continue + fi + DECISION=$(python3 tools/reform_validation/spawn_or_wait.py check "$cand") + if [ "$DECISION" = "WAIT" ]; then + echo "::notice::a Modal run for $cand is in flight — a later tick harvests it." + continue + fi + if [ "$DEPLOYED" = "0" ]; then + modal deploy tools/reform_validation/modal_backfill_app.py + DEPLOYED=1 + fi + PRODUCER_REF=$(gh api "repos/$PRODUCER_REPO/commits/main" --jq .sha) + python3 tools/reform_validation/spawn_or_wait.py spawn "$cand" "$PRODUCER_REF" "$GITHUB_SHA" + echo "::notice::spawned Modal backfill for $cand at microcosm ${PRODUCER_REF:0:9} (driver ${GITHUB_SHA:0:9}) — a later tick harvests it." + done + + - name: Register release + rebuild DB + if: steps.harvest.outputs.harvested != '' + env: + PYTHONPATH: . + TARGET: ${{ steps.harvest.outputs.harvested }} + run: | + set -euo pipefail + python3 tools/reform_validation/register_release.py "$TARGET" "$RUNNER_TEMP/harvest/$TARGET.json" + + - name: Verify the DB builds with the new artifact + # Post-#74 `data/scorecard.db` is a derived artifact (gitignored, built + # in CI from the committed sources), never committed. So the verification + # is: build the whole DB from scratch to a THROWAWAY path and confirm the + # newly-registered raw artifact ingests — the DB is never staged, and the + # no-drift gates on the PR do the rest. + if: steps.harvest.outputs.harvested != '' + env: + PYTHONPATH: . + run: | + set -euo pipefail + python3 -m scorecard_db.build_db "$RUNNER_TEMP/verify.db" + python3 -c "import sqlite3,sys;n=sqlite3.connect(sys.argv[1]).execute(\"SELECT COUNT(*) FROM pe_results WHERE run_id LIKE 'populace-rv-%'\").fetchone()[0];print(n,'reform-validation results in the freshly built DB');assert n>0" "$RUNNER_TEMP/verify.db" + + - name: Open PR + if: steps.harvest.outputs.harvested != '' + env: + GH_TOKEN: ${{ github.token }} + TARGET: ${{ steps.harvest.outputs.harvested }} + run: | + set -euo pipefail + BR="reform-validation/$TARGET" + # Provenance the reviewer can verify: the scorecard commit this ran + # at and the sha256 of the artifact being ingested (blocker 6 — this + # is the scorecard-commit + artifact-sha binding the attestation + # names; the artifact also carries its own producer/driver stamps). + ART_SHA=$(sha256sum "$RAW_DIR/$TARGET.json" | cut -d' ' -f1) + git config user.name "github-actions[bot]" + git config user.email "41898282+github-actions[bot]@users.noreply.github.com" + git checkout -b "$BR" + # Commit ONLY the durable outputs — the raw artifact + source.json. + # data/scorecard.db is gitignored post-#74 (derived, built in CI), so + # staging it would abort the step under `set -e`. + git add "$RAW_DIR/$TARGET.json" sources/populace-reform-validation/source.json + git commit -m "Ingest reform_validation for $TARGET" + # force-with-lease so a stale branch from a prior failed attempt is + # recoverable without stranding the release (blocker 3). + git push -u --force-with-lease origin "$BR" + gh pr create --base main --head "$BR" \ + --title "Ingest reform_validation for $TARGET" \ + --body "Automated backfill by \`.github/workflows/reform-validation-backfill.yml\`. + + **Provenance (attestation):** + - scorecard commit: \`${GITHUB_SHA}\` + - artifact sha256: \`${ART_SHA}\` + - producer/driver revisions + engine pins: in the artifact's \`_attestation\` block (the ingest consumes it and rejects a missing/\`unknown\` one). + + The producer ran on Modal (\`tools/reform_validation/modal_backfill_app.py\`) against the released \`populace_us_2024.h5\` at the build's exact package versions; \`register_release.py\` landed it under \`$RAW_DIR/\`. Only the raw artifact + \`source.json\` are committed — \`data/scorecard.db\` is a derived artifact (built in CI by \`scorecard_db.build_db\`, verified green here), and the reform-validation ingest asserts non-RV tables are unchanged inside its transaction, rolling back on any cross-slice write. Opened for review rather than auto-merged: confirm the new engine pin and the exact-count ingest tests (\`tests/test_reform_validation_ingest.py\`) before merging." diff --git a/scorecard_db/README.md b/scorecard_db/README.md index 67ff6c06..fb70cfca 100644 --- a/scorecard_db/README.md +++ b/scorecard_db/README.md @@ -170,6 +170,27 @@ categories `Reform` / `Program actual` with an explicit `jct.publisher` scope is per country — run prefixes are disjoint by test, so one country's wholesale replace cannot delete another's rows. +**Per-release automation** (`tools/reform_validation/`, re-homed from +calibration-diagnostics #112 per issue #15 call 3). The scheduled workflow +`.github/workflows/reform-validation-backfill.yml` keeps this population +current without hand-running the producer: on a tick it checks whether the +latest certified release already has a `raw/.json`; if not, it spawns +the producer on Modal (`modal_backfill_app.py` drives `backfill.py` in a +64GB container at the release's exact engine pins — the sim no longer fits a +GitHub runner), and on a later tick harvests the artifact, `register_release.py` +lands it under `raw/` and adds it to `source.json`, and the ingest rebuilds +the DB slice. A new release needs no code edit: `backfill.py` stamps the +release-manifest versions into the artifact's `engine` block, which +`_base_engine` reads when the release isn't in `ENGINE_VERSIONS`, and OBBBA +scoring defaults to `jcx_stacked` (the current producer's mode). The +liveness-checked spawner (`spawn_or_wait.py`) records the Modal call id on +the Volume and re-spawns dead runs (resuming from checkpointed batch +partials). The workflow opens a PR for review rather than auto-merging, +because a new release brings a new engine pin and shifts the exact-count +ingest tests — a human confirms the DB delta and updates +`tests/test_reform_validation_ingest.py`'s corpus counts before merging. +Requires `MODAL_TOKEN_ID` / `MODAL_TOKEN_SECRET` repo secrets. + ## 2026-08-20 population: OBR published policy effects (UK) Fourth UK population and the external half of the Macro entry point diff --git a/scorecard_db/ingest_reform_validation.py b/scorecard_db/ingest_reform_validation.py index 2ffc05c2..8c9e74c3 100644 --- a/scorecard_db/ingest_reform_validation.py +++ b/scorecard_db/ingest_reform_validation.py @@ -288,9 +288,147 @@ def _resolve_existing_claim(db, claim) -> str | None: return rows[0][0] if rows else None -# US alias kept for the module's own US-specific tables below. +# US alias kept for the module's own US-specific tables below. These are +# the hand-pinned US releases (the historical backfill plus the locally +# produced spm-20260915); a genuinely new release arriving from the +# automated backfill (tools/reform_validation/) carries its own pin in the +# artifact's ``engine`` block, so it ingests without a code edit here — see +# ``_base_engine``. ENGINE_VERSIONS = COUNTRIES["US"]["engine_versions"] + +# The closed vocabulary of OBBBA scoring modes. A release the historical +# OBBBA_SCORING_MODE table below doesn't name must SAY how it scored OBBBA +# in the artifact (backfill.py stamps ``obbba_scoring_mode``); we never +# guess a default, because a silent wrong mode reads as release-over-release +# drift. ``_obbba_scoring_mode`` resolves and validates against this set. +VALID_OBBBA_SCORING_MODES = frozenset({"stacked_chained", "isolated", "jcx_stacked"}) + + +def _obbba_scoring_mode(release_id: str, artifact: dict) -> str: + """Resolve the release's OBBBA scoring mode, failing loudly. + + Hand-pinned releases are named in OBBBA_SCORING_MODE + (authoritative). Any other release must carry an explicit, valid + ``obbba_scoring_mode`` in its artifact — the current producer stamps + ``jcx_stacked`` (backfill.py), but we validate rather than assume, so a + producer that changes its stacking can't slip through as drift.""" + mode = OBBBA_SCORING_MODE.get(release_id) or artifact.get("obbba_scoring_mode") + if mode not in VALID_OBBBA_SCORING_MODES: + raise SystemExit( + f"{release_id}: OBBBA scoring mode {mode!r} is missing or not in " + f"{sorted(VALID_OBBBA_SCORING_MODES)} — the artifact must carry a " + "valid `obbba_scoring_mode` (backfill.py stamps it) or the release " + "must be added to OBBBA_SCORING_MODE. Refusing to guess a default." + ) + return mode + + +_ATTESTATION_REQUIRED = ("producer_commit", "driver_commit", "h5_sha256") +_HEX = re.compile(r"\A[0-9a-f]+\Z") + + +def _validate_attestation( + release_id: str, artifact: dict, engine_versions: dict | None = None +) -> None: + """Consume AND VERIFY the provenance attestation the automation stamps. + Hand-pinned releases (the country's ``engine_versions`` map in + COUNTRIES) predate it and are exempt; a genuinely-new + (automation-produced) release must carry a complete block whose fields are + well-FORMED and CROSS-CONSISTENT with the artifact it describes. Presence + alone (the prior check) let a forged block through — junk commits, + ``h5_sha256="x"``, an ``engine=9.9.9`` that disagreed with the artifact + (re-gate blocker 6/2). The producer independently verifies the same fields + (rehashes the H5, reads back installed engine versions); this is the + ingest-side gate on a committed raw artifact.""" + pins = ENGINE_VERSIONS if engine_versions is None else engine_versions + if release_id in pins: + return + att = artifact.get("_attestation") + if not isinstance(att, dict): + raise SystemExit( + f"{release_id}: no _attestation block — the automated backfill " + "stamps one; refusing to ingest un-attested provenance." + ) + missing = [ + k + for k in _ATTESTATION_REQUIRED + if not att.get(k) or str(att.get(k)).strip().lower() == "unknown" + ] + if missing: + raise SystemExit( + f"{release_id}: _attestation has missing/unknown fields {missing} " + "— refusing to ingest incomplete provenance." + ) + # Format: h5_sha256 is 64 lowercase hex; commit ids are 7-64 hex. + h5 = str(att["h5_sha256"]).strip().lower() + if len(h5) != 64 or not _HEX.match(h5): + raise SystemExit( + f"{release_id}: _attestation.h5_sha256 is not a 64-char sha256 ({h5!r})." + ) + for k in ("producer_commit", "driver_commit"): + v = str(att[k]).strip().lower() + if not (7 <= len(v) <= 64) or not _HEX.match(v): + raise SystemExit( + f"{release_id}: _attestation.{k} is not a git sha ({att[k]!r})." + ) + # Cross-consistency with the artifact the block claims to describe: the + # attested id and BOTH engine versions must match the artifact's own + # `engine` block (which drives _base_engine / the pe_results stamp), so a + # forged engine can't ride in behind a plausible-looking attestation. + # The attested id is REQUIRED: an absent one used to pass this check + # vacuously, which is a presence-style gate (re-gate round 3, finding 2). + if att.get("release_id") != release_id: + raise SystemExit( + f"{release_id}: _attestation.release_id {att.get('release_id')!r} " + "is missing or mismatches the artifact's release_id." + ) + engine = artifact.get("engine") or {} + for k in ("policyengine_us", "policyengine_core"): + av, ev = att.get(k), engine.get(k) + if av is None or ev is None or str(av) != str(ev): + raise SystemExit( + f"{release_id}: _attestation.{k}={av!r} inconsistent with " + f"artifact.engine.{k}={ev!r}." + ) + + +def _base_engine( + release_id: str, + artifact: dict, + engine_versions: dict | None = None, + country: str = "US", +) -> str: + """Resolve the release's engine pin. + + Hand-pinned releases are in the country's ``engine_versions`` map in + COUNTRIES (authoritative, including the l0-refit subtlety where the + release was built at 1.752.2 but named 1.729.0 rows offline — that lives + in ENGINE_OVERRIDES). A new US release from the automated backfill + self-describes: backfill.py stamps the release_manifest's versions into + the artifact's ``engine`` block, so no hand-edit is needed to ingest it. + The self-describing fallback reads ``engine.policyengine_us`` and so is + US-only; any other country still needs its pin in COUNTRIES.""" + pins = ENGINE_VERSIONS if engine_versions is None else engine_versions + pin = pins.get(release_id) + if pin is not None: + return pin + if country == "US": + pin = (artifact.get("engine") or {}).get("policyengine_us") + if pin: + return str(pin) + raise SystemExit( + f"{release_id}: no engine pin — not in the US engine_versions map " + "in COUNTRIES and the artifact carries no engine.policyengine_us " + "block. Re-run the backfill (it stamps the pin) or add the " + "release_manifest.json pin." + ) + raise SystemExit( + f"{release_id}: no engine pin — missing from the {country} " + "engine_versions map in COUNTRIES — add its release_manifest.json pin" + ) + + # Pin overrides an artifact documents about itself, keyed per category and # per row id. The l0-refit backfill note names three offline batches scored # at 1.729.0 on the released H5 (the release itself was built at 1.752.2): @@ -841,12 +979,14 @@ def _obbba_results( country: str = "US", run_prefix: str = RUN_PREFIX, registry_mark: str = REGISTRY_MARK, + mode: str | None = None, ) -> list[PEResult]: """Attach the registry's OBBBA computation to the canonical harvest claims — the FY2026 provision claim, and the FY2027 one when the registry carries that benchmark. Same PE value both times (calendar-2026 liability), so both are constructed comparisons; the construction names - the release's scoring mode and the CY-for-FY approximation.""" + the release's scoring ``mode`` (resolved+validated by the caller) and the + CY-for-FY approximation.""" # OBBBA is a JCT scoring exercise: US-only by construction. A # non-US artifact must never reach this path — its rows would take # the US run prefix/registry mark and become deletable by the US @@ -854,11 +994,14 @@ def _obbba_results( # prefixes exist to prevent. if country != "US": raise ValueError(f"OBBBA rows are US-only; got country {country!r}") + if mode is None: + # Direct callers that don't pass a mode get the release's pinned + # mode, validated — never a guessed default. + mode = _obbba_scoring_mode(release_id, {}) provision = OBBBA_PROVISIONS.get(row["id"]) if provision is None: raise ValueError(f"OBBBA row {row['id']} missing from OBBBA_PROVISIONS") measure = _pe_block(row).get("measure") or "" - mode = OBBBA_SCORING_MODE[release_id] pe_value = _pe_value(row, f"reform_delta:{measure}") results = [] for fy, score in ( @@ -1069,6 +1212,56 @@ def ingest_uk(db_path: Path, raw_dir: Path | None = None) -> dict: } +def _non_rv_fingerprint( + conn, + run_prefix: str = RUN_PREFIX, + registry_mark: str = REGISTRY_MARK, + lane: str = "populace-reform-validation", +) -> str: + """Order-independent hash over every row this ingest must NOT touch. + + The rebuild is scoped to the reform-validation slice: it deletes its own + results (``run_id LIKE 'populace-rv-%'``) and the registry-marked + external_scores, re-inserts them, and upserts the one + ``populace-reform-validation`` lane. Everything else — the harvest and + Urban claims, their results, exhibits, diagnoses, other lanes — must be + byte-identical before and after. Fingerprinting the complement and + asserting it is unchanged turns "some rows exist" into a real invariant: + a bug that dropped or rewrote a non-RV row fails the ingest loudly. + + The slice is per country (#62): the caller passes the country's run + prefix, registry mark and lane, so a UK ingest excludes its own rows + and the US slice is part of ITS complement (and vice versa).""" + import hashlib + + filters = { + "pe_results": ("run_id NOT LIKE ?", (f"{run_prefix}%",)), + "external_scores": ( + "COALESCE(json_extract(publication, '$.registry'), '') != ?", + (registry_mark,), + ), + "lanes": ("lane != ?", (lane,)), + } + tables = [ + r[0] + for r in conn.execute( + "SELECT name FROM sqlite_master WHERE type='table' " + "AND name NOT LIKE 'sqlite_%' ORDER BY name" + ) + ] + acc = 0 + for t in tables: + if t in filters: + where, params = filters[t] + cur = conn.execute(f'SELECT * FROM "{t}" WHERE {where}', params) + else: + cur = conn.execute(f'SELECT * FROM "{t}"') + for row in cur: + digest = hashlib.sha256(repr((t, tuple(row))).encode()).digest() + acc ^= int.from_bytes(digest, "big") # XOR: order-independent + return f"{acc:064x}" + + def ingest(db_path: Path, raw_dir: Path | None = None, country: str = "US") -> dict: if country not in COUNTRIES: raise ValueError(f"unknown country {country!r}; known: {sorted(COUNTRIES)}") @@ -1100,14 +1293,19 @@ def ingest(db_path: Path, raw_dir: Path | None = None, country: str = "US") -> d if country == "US" else _uk_executed_baseline(artifact, release_id) ) - base_engine = engine_versions.get(release_id) - if base_engine is None: - raise SystemExit( - f"{release_id} missing from the {country} engine_versions " - "map in COUNTRIES — add its release_manifest.json pin" - ) + base_engine = _base_engine(release_id, artifact, engine_versions, country) + _validate_attestation(release_id, artifact, engine_versions) overrides = ENGINE_OVERRIDES.get(release_id, {}) computed_at = _release_timestamp(release_id) + # Resolve+validate the OBBBA scoring mode once per release, and only + # when the release actually has OBBBA rows — fails loudly on a + # missing/unknown mode rather than defaulting. + obbba_mode = ( + _obbba_scoring_mode(release_id, artifact) + if country == "US" + and any(r["category"] == "OBBBA" for r in artifact["reforms"]) + else None + ) actuals = { r["id"]: r["jct"].get("score") for r in artifact["reforms"] @@ -1135,6 +1333,7 @@ def ingest(db_path: Path, raw_dir: Path | None = None, country: str = "US") -> d computed_at, claims, tallies, + mode=obbba_mode, validate=path == releases[-1], position=obbba_position, country=country, @@ -1237,6 +1436,11 @@ def ingest(db_path: Path, raw_dir: Path | None = None, country: str = "US") -> d score_rows = [ScorecardDB.score_row(c) for c in claims.values()] result_rows = [ScorecardDB.result_row(r) for r in results] n_claims, n_results = len(score_rows), len(result_rows) + # Snapshot the non-RV data so we can prove the rebuild touched only this + # country's reform-validation slice (issue #53 re-gate blocker 6). + pre_fingerprint = _non_rv_fingerprint( + db.conn, run_prefix, registry_mark, cfg["lane"] + ) with db.conn: db.conn.execute( "DELETE FROM pe_results WHERE run_id LIKE ?", (f"{run_prefix}%",) @@ -1260,6 +1464,22 @@ def ingest(db_path: Path, raw_dir: Path | None = None, country: str = "US") -> d "2026-08-04", ), ) + # Assert the invariant INSIDE the transaction: raising here rolls the + # whole rebuild back (sqlite's `with conn` rolls back on exception), + # so a cross-slice write is ABORTED, not merely detected after the + # commit already landed the rogue row (blocker 6). Reads inside the + # transaction see the pending writes, and the RV rebuild's own rows + # are excluded from the fingerprint, so a correct rebuild is a no-op. + post_fingerprint = _non_rv_fingerprint( + db.conn, run_prefix, registry_mark, cfg["lane"] + ) + if post_fingerprint != pre_fingerprint: + raise SystemExit( + "reform-validation ingest changed non-RV data — the rebuild " + "must only touch its own results, registry-marked claims, and " + f"the {cfg['lane']} lane (fingerprint " + f"{pre_fingerprint[:12]} -> {post_fingerprint[:12]}); rolled back." + ) cov = db.coverage() db.close() return { diff --git a/sources/populace-reform-validation/source.json b/sources/populace-reform-validation/source.json index da2ec9bc..5c27f824 100644 --- a/sources/populace-reform-validation/source.json +++ b/sources/populace-reform-validation/source.json @@ -5,7 +5,7 @@ "url": "https://github.com/PolicyEngine/populace", "data_url": "vendored from calibration-diagnostics frontend/lib/populace/reform-overrides/ (the retired External checks tab)", "fetched": "2026-08-03", - "fetch_notes": "Each raw/.json is the reform_validation.json artifact for one certified populace-us release, produced by populace's reform_validation_payload at the release's exact policyengine-us/-core pins (provenance in each file's _backfill_note). New releases arrive from the Modal-backed backfill (cal-diag #112 machinery, being re-homed here).", + "fetch_notes": "Each raw/.json is the reform_validation.json artifact for one certified populace-us release, produced by populace's reform_validation_payload at the release's exact policyengine-us/-core pins (provenance in each file's _backfill_note, and a structured engine block on artifacts produced here). New releases arrive from the Modal-backed backfill re-homed from cal-diag #112 into tools/reform_validation/ (.github/workflows/reform-validation-backfill.yml): the sim runs on Modal, register_release.py lands the artifact, and scorecard_db.ingest_reform_validation rebuilds the DB slice.", "model": { "name": "external scorekeepers + admin actuals (mixed)", "open_source": true, diff --git a/tests/test_reform_validation_automation.py b/tests/test_reform_validation_automation.py new file mode 100644 index 00000000..86c1c64c --- /dev/null +++ b/tests/test_reform_validation_automation.py @@ -0,0 +1,360 @@ +"""The re-homed backfill automation (calibration-diagnostics #112 -> here): +registering a harvested artifact into the source tree, and ingesting a +genuinely-new release whose engine pin is carried in the artifact rather than +hand-coded in ENGINE_VERSIONS.""" + +import json +import sqlite3 +import sys +from pathlib import Path + +import pytest + +from scorecard_db.ingest_reform_validation import ( + ENGINE_VERSIONS, + RAW, + RUN_PREFIX, + _non_rv_fingerprint, + ingest, +) + +TOOLS = Path(__file__).resolve().parent.parent / "tools" / "reform_validation" +sys.path.insert(0, str(TOOLS)) +import register_release # noqa: E402 + +BUILDO = "populace-us-2024-buildo-sparse-rmloss100-22bd902-20260722T232627Z" +# A plausible next release: not in ENGINE_VERSIONS / OBBBA_SCORING_MODE, with a +# parseable trailing timestamp. +NEW_RELEASE = "populace-us-2024-buildp-sparse-rmloss100-deadbeef-20260801T000000Z" + + +def _new_release_artifact() -> dict: + """The buildo artifact re-stamped as a brand-new release that carries its + own engine pin (as the updated backfill.py now emits).""" + artifact = json.loads((RAW / f"{BUILDO}.json").read_text()) + artifact["release_id"] = NEW_RELEASE + artifact["engine"] = { + "policyengine_us": "1.799.9", + "policyengine_core": "3.0.0", + } + # backfill.py stamps the scoring mode explicitly; the ingest fails loudly + # without it (no silent default). + artifact["obbba_scoring_mode"] = "jcx_stacked" + # The ingest consumes the attestation for a non-historical release and + # rejects a missing/unknown one (blocker 6), so a valid block is required. + artifact["_attestation"] = { + "release_id": NEW_RELEASE, + "producer_commit": "abc1234", + "driver_commit": "def5678", + "h5_sha256": "0" * 64, + "policyengine_us": "1.799.9", + "policyengine_core": "3.0.0", + } + return artifact + + +# --------------------------------------------------------------------------- # +# register_release +# --------------------------------------------------------------------------- # +def _source_json(tmp_path: Path, releases: list[str]) -> Path: + p = tmp_path / "source.json" + p.write_text( + json.dumps({"id": "x", "releases": releases}, indent=2) + "\n", + encoding="utf-8", + ) + return p + + +def test_register_release_lands_artifact_and_orders_releases(tmp_path): + raw_dir = tmp_path / "raw" + artifact_path = tmp_path / "in.json" + artifact_path.write_text(json.dumps(_new_release_artifact())) + # An existing older release already registered; the new one sorts after it. + source_json = _source_json(tmp_path, [BUILDO]) + + summary = register_release.register( + NEW_RELEASE, artifact_path, raw_dir=raw_dir, source_json=source_json + ) + + assert (raw_dir / f"{NEW_RELEASE}.json").exists() + assert summary["added_to_source"] is True + # Ordered oldest-first by release timestamp (buildo 2026-07-22 < new 08-01). + assert json.loads(source_json.read_text())["releases"] == [BUILDO, NEW_RELEASE] + + +def test_register_release_is_idempotent(tmp_path): + raw_dir = tmp_path / "raw" + artifact_path = tmp_path / "in.json" + artifact_path.write_text(json.dumps(_new_release_artifact())) + source_json = _source_json(tmp_path, [BUILDO, NEW_RELEASE]) + + summary = register_release.register( + NEW_RELEASE, artifact_path, raw_dir=raw_dir, source_json=source_json + ) + assert summary["added_to_source"] is False + assert json.loads(source_json.read_text())["releases"] == [BUILDO, NEW_RELEASE] + + +def test_register_release_rejects_id_mismatch(tmp_path): + artifact_path = tmp_path / "in.json" + artifact_path.write_text(json.dumps(_new_release_artifact())) + with pytest.raises(SystemExit): + register_release.register( + "some-other-release-20260101T000000Z", + artifact_path, + raw_dir=tmp_path / "raw", + source_json=_source_json(tmp_path, []), + ) + + +# --------------------------------------------------------------------------- # +# Ingesting a new release without a hand-coded engine pin +# --------------------------------------------------------------------------- # +def test_new_release_ingests_using_the_artifacts_engine_block(tmp_path): + # The whole point of the automation: a release the code has never seen + # ingests without a SystemExit, taking its engine pin from the artifact. + assert NEW_RELEASE not in ENGINE_VERSIONS + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + (raw_dir / f"{NEW_RELEASE}.json").write_text(json.dumps(_new_release_artifact())) + db_path = tmp_path / "scorecard.db" + + summary = ingest(db_path, raw_dir=raw_dir) + assert summary["releases"] == 1 + assert summary["results"] > 0 + + conn = sqlite3.connect(db_path) + # Every result for this release is stamped with the artifact's pin. + engines = { + r[0] + for r in conn.execute( + "SELECT DISTINCT engine_version FROM pe_results WHERE run_id LIKE ?", + (f"{RUN_PREFIX}%",), + ) + } + assert engines == {"1.799.9"} + # OBBBA rows default to the current producer's scoring mode. Their + # construction is reform_delta:::cy2026_for_fy; other + # reform_delta rows (state bills, repeals) carry no mode segment. + modes = { + r[0].split(":")[2] + for r in conn.execute( + "SELECT pe_construction FROM pe_results" + " WHERE pe_construction LIKE '%cy2026_for_fy%'" + ) + } + assert modes == {"jcx_stacked"} + conn.close() + + +def test_new_release_without_engine_block_fails_loudly(tmp_path): + # No pin anywhere -> a clear SystemExit, not a KeyError or a silent + # mis-stamp. + artifact = _new_release_artifact() + del artifact["engine"] + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + (raw_dir / f"{NEW_RELEASE}.json").write_text(json.dumps(artifact)) + with pytest.raises(SystemExit, match="no engine pin"): + ingest(tmp_path / "scorecard.db", raw_dir=raw_dir) + + +def test_new_release_without_scoring_mode_fails_loudly(tmp_path): + # A release the OBBBA_SCORING_MODE table doesn't name, with OBBBA rows but + # no explicit mode, must fail loudly rather than default (re-gate blocker + # 5) — a silent wrong mode reads as release-over-release drift. + artifact = _new_release_artifact() + del artifact["obbba_scoring_mode"] + assert any(r["category"] == "OBBBA" for r in artifact["reforms"]) + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + (raw_dir / f"{NEW_RELEASE}.json").write_text(json.dumps(artifact)) + with pytest.raises(SystemExit, match="OBBBA scoring mode"): + ingest(tmp_path / "scorecard.db", raw_dir=raw_dir) + + +# --------------------------------------------------------------------------- # +# Non-RV table invariant (re-gate blocker 6) +# --------------------------------------------------------------------------- # +def _fresh_db(tmp_path): + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + (raw_dir / f"{NEW_RELEASE}.json").write_text(json.dumps(_new_release_artifact())) + db_path = tmp_path / "scorecard.db" + ingest(db_path, raw_dir=raw_dir) + return db_path, raw_dir + + +def test_non_rv_fingerprint_ignores_rv_but_catches_others(tmp_path): + db_path, _ = _fresh_db(tmp_path) + conn = sqlite3.connect(db_path) + base = _non_rv_fingerprint(conn) + # Mutating the RV lane must NOT move the fingerprint (it's the ingest's + # own row). + conn.execute( + "UPDATE lanes SET detail = 'touched' WHERE lane = 'populace-reform-validation'" + ) + conn.commit() + assert _non_rv_fingerprint(conn) == base + # Inserting an unrelated (harvest-style) lane MUST move it. + conn.execute( + "INSERT INTO lanes (lane, stage, detail, updated_at)" + " VALUES ('harvest-x', 'ingested', 'seed', '2026-01-01')" + ) + conn.commit() + assert _non_rv_fingerprint(conn) != base + conn.close() + + +def test_reingest_preserves_non_rv_rows(tmp_path): + # A row this ingest must never touch (a foreign lane) survives a rebuild, + # and the rebuild completes (the invariant assertion passes). + db_path, raw_dir = _fresh_db(tmp_path) + conn = sqlite3.connect(db_path) + conn.execute( + "INSERT INTO lanes (lane, stage, detail, updated_at)" + " VALUES ('harvest-x', 'ingested', 'seed', '2026-01-01')" + ) + conn.commit() + conn.close() + ingest(db_path, raw_dir=raw_dir) # rebuild + conn = sqlite3.connect(db_path) + assert conn.execute( + "SELECT stage FROM lanes WHERE lane = 'harvest-x'" + ).fetchone() == ("ingested",) + conn.close() + + +def test_reingest_preserves_non_rv_pe_results_row(tmp_path): + # A harvest-style pe_results row (run_id NOT populace-rv-%) must survive the + # rebuild — pins the delete-scope so broadening the DELETE (or dropping the + # fingerprint abort) can't silently wipe non-RV results with all else green + # (re-gate blocker 6 / test-gap note). + db_path, raw_dir = _fresh_db(tmp_path) + conn = sqlite3.connect(db_path) + conn.execute( + "INSERT INTO pe_results (claim_id, status, engine_version, data_bundle," + " pe_construction, run_id, computed_at, annotations)" + " VALUES ('harvest-claim', 'comparable', '1.0.0', 'harvest-bundle'," + " 'direct', 'harvest-run-123', '2026-01-01', '{}')" + ) + conn.commit() + conn.close() + ingest(db_path, raw_dir=raw_dir) # rebuild must not touch the seeded row + conn = sqlite3.connect(db_path) + assert ( + conn.execute( + "SELECT COUNT(*) FROM pe_results WHERE run_id = 'harvest-run-123'" + ).fetchone()[0] + == 1 + ) + conn.close() + + +def test_new_release_without_attestation_fails_loudly(tmp_path): + # A non-historical release must carry a complete provenance attestation; + # the ingest CONSUMES it and rejects a missing/unknown one rather than + # ingesting un-attested provenance (re-gate blocker 6). + artifact = _new_release_artifact() + del artifact["_attestation"] + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + (raw_dir / f"{NEW_RELEASE}.json").write_text(json.dumps(artifact)) + with pytest.raises(SystemExit, match="attestation"): + ingest(tmp_path / "scorecard.db", raw_dir=raw_dir) + + # An `unknown`-stamped field is likewise rejected. + artifact = _new_release_artifact() + artifact["_attestation"]["producer_commit"] = "unknown" + (raw_dir / f"{NEW_RELEASE}.json").write_text(json.dumps(artifact)) + with pytest.raises(SystemExit, match="missing/unknown"): + ingest(tmp_path / "scorecard.db", raw_dir=raw_dir) + + +def test_attestation_is_verified_not_just_present(tmp_path): + # Presence isn't enough (re-gate blocker 2/6): a forged block with junk + # format or an engine that disagrees with the artifact must be rejected. + raw_dir = tmp_path / "raw" + raw_dir.mkdir() + + def _ingest(mutate, match): + art = _new_release_artifact() + mutate(art) + (raw_dir / f"{NEW_RELEASE}.json").write_text(json.dumps(art)) + with pytest.raises(SystemExit, match=match): + ingest(tmp_path / "scorecard.db", raw_dir=raw_dir) + + # h5_sha256 that isn't a 64-char hex digest. + _ingest(lambda a: a["_attestation"].update(h5_sha256="x"), "h5_sha256") + # a commit id that isn't hex. + _ingest(lambda a: a["_attestation"].update(producer_commit="not-a-sha!"), "git sha") + # an engine version that disagrees with the artifact's own engine block + # (the forged `engine=9.9.9` Max reproduced). + _ingest( + lambda a: a["_attestation"].update(policyengine_us="9.9.9"), + "inconsistent", + ) + # an attested release id that is absent or names a different release + # (re-gate round 3: absence used to pass the cross-check vacuously). + _ingest(lambda a: a["_attestation"].pop("release_id"), "release_id") + _ingest( + lambda a: a["_attestation"].update(release_id=BUILDO), + "release_id", + ) + + # The clean fixture still ingests (cross-consistent block). + art = _new_release_artifact() + (raw_dir / f"{NEW_RELEASE}.json").write_text(json.dumps(art)) + summary = ingest(tmp_path / "scorecard.db", raw_dir=raw_dir) + assert summary["results"] > 0 + + +# --------------------------------------------------------------------------- # +# backfill.merge revision refusal (re-gate blocker 4) — pure stdlib +# --------------------------------------------------------------------------- # +import backfill # noqa: E402 + + +def test_refuse_if_mixed_revisions(tmp_path): + part_dir = tmp_path / "partials" + part_dir.mkdir() + (part_dir / "a.json").write_text("{}") # something to purge + + # Mixed producer revisions -> refuse AND purge the workdir partials. + with pytest.raises(SystemExit, match="multiple or unstamped"): + backfill._refuse_if_mixed({"aaa", "bbb"}, {"drv"}, part_dir) + assert not part_dir.exists() + + # A missing/unknown stamp is refused too (not just a clean 2-way mix). + assert backfill._revision_dirty({None}) + assert backfill._revision_dirty({"unknown"}) + assert backfill._revision_dirty(set()) + assert backfill._revision_dirty({"a", "b"}) + # A single known revision on both axes is accepted (no raise). + ok_dir = tmp_path / "ok" + ok_dir.mkdir() + backfill._refuse_if_mixed({"aaa"}, {"drv"}, ok_dir) + assert ok_dir.exists() # not purged + + +def test_observed_engine_stamps_installed_versions_or_refuses(): + # The attestation's engine versions are READ BACK from the environment + # and must equal the release manifest being stamped (re-gate round 3, + # finding 2: the manifest was fetched twice and installed versions were + # never read back into the stamp). + manifest = { + "build": { + "built_with_model_package": {"version": "1.799.9"}, + "built_with_core_package": {"version": "3.0.0"}, + } + } + installed = {"policyengine-us": "1.799.9", "policyengine-core": "3.0.0"} + assert backfill._observed_engine(manifest, version=installed.__getitem__) == { + "policyengine_us": "1.799.9", + "policyengine_core": "3.0.0", + } + # The manifest moved after install (or a dependency overrode the pin): + # refuse rather than stamp versions that didn't run. + drifted = {**installed, "policyengine-us": "1.800.0"} + with pytest.raises(SystemExit, match="installed engines"): + backfill._observed_engine(manifest, version=drifted.__getitem__) diff --git a/tests/test_reform_validation_spawn.py b/tests/test_reform_validation_spawn.py new file mode 100644 index 00000000..bb52c31b --- /dev/null +++ b/tests/test_reform_validation_spawn.py @@ -0,0 +1,173 @@ +"""Spawn/liveness decision table for tools/reform_validation/spawn_or_wait.py +(re-gate blocker 2/3). Modal isn't a CI dependency and the module imports it at +top level, so we stub `modal`, `modal.exception`, and `modal_backfill_app` in +sys.modules before importing spawn_or_wait, then drive check() through the +decision table — especially the fail-closed paths that a transport blip must +never turn into a respawn.""" + +import sys +import time +import types +from pathlib import Path + +import pytest + +# --------------------------------------------------------------------------- # +# Stubs, installed before importing spawn_or_wait +# --------------------------------------------------------------------------- # +_STATE = { + "files": {}, # path -> str contents + "errors": {}, # path -> Exception to raise on read + "listing": [], # entries returned by listdir("/") + "call_get": "timeout", # "timeout" | None (completed) | Exception + "listdir_error": None, +} + +_mexc = types.ModuleType("modal.exception") +for _name in ( + "RemoteError", + "ExecutionError", + "FunctionTimeoutError", + "NotFoundError", + "InvalidError", +): + setattr(_mexc, _name, type(_name, (Exception,), {})) + + +class _Entry: + def __init__(self, path): + self.path = path + + +class _FakeVolume: + @staticmethod + def from_name(name, create_if_missing=False): + return _FakeVolume() + + def read_file(self, path): + err = _STATE["errors"].get(path) + if err is not None: + raise err + if path in _STATE["files"]: + return [_STATE["files"][path].encode()] + raise FileNotFoundError(path) + + def listdir(self, path): + if _STATE["listdir_error"] is not None: + raise _STATE["listdir_error"] + return _STATE["listing"] + + +class _FakeFunctionCall: + @staticmethod + def from_id(call_id): + return _FakeFunctionCall() + + def get(self, timeout=0): + beh = _STATE["call_get"] + if beh == "timeout": + raise TimeoutError() + if isinstance(beh, Exception): + raise beh + return None # completed successfully + + +_modal = types.ModuleType("modal") +_modal.Volume = _FakeVolume +_modal.FunctionCall = _FakeFunctionCall +_modal.Function = object +_modal.exception = _mexc +sys.modules["modal"] = _modal +sys.modules["modal.exception"] = _mexc + +_app = types.ModuleType("modal_backfill_app") +_app.APP_NAME = "app" +_app.VOLUME_NAME = "vol" +sys.modules["modal_backfill_app"] = _app + +TOOLS = Path(__file__).resolve().parent.parent / "tools" / "reform_validation" +sys.path.insert(0, str(TOOLS)) +import spawn_or_wait # noqa: E402 + +REL = "populace-us-2024-buildp-sparse-rmloss100-deadbeef-20260801T000000Z" +PREFIX = f"rv_{spawn_or_wait._sha8(REL)}" + + +def _reset(**over): + _STATE.update( + files={}, errors={}, listing=[], call_get="timeout", listdir_error=None + ) + _STATE.update(over) + + +def _started(age_hours): + return f"{REL}\nproducer abc\nstarted_at {int(time.time() - age_hours * 3600)}\n" + + +# --------------------------------------------------------------------------- # +# Decision table +# --------------------------------------------------------------------------- # +def test_call_running_waits(): + _reset(files={f"{PREFIX}/call_id": "call-1"}, call_get="timeout") + assert spawn_or_wait.check(REL)[0] == "WAIT" + + +def test_call_terminally_failed_spawns(): + _reset(files={f"{PREFIX}/call_id": "call-1"}, call_get=_mexc.RemoteError("boom")) + assert spawn_or_wait.check(REL)[0] == "SPAWN" + + +def test_completed_with_artifact_waits_not_respawn(): + # Healthy-completion TOCTOU: the call returned and the artifact IS on the + # Volume — the drain will harvest it; respawning would redo finished work. + _reset( + files={f"{PREFIX}/call_id": "call-1"}, + call_get=None, + listing=[_Entry(f"reform_validation_{REL}.json")], + ) + assert spawn_or_wait.check(REL)[0] == "WAIT" + + +def test_completed_without_artifact_spawns(): + _reset(files={f"{PREFIX}/call_id": "call-1"}, call_get=None, listing=[]) + assert spawn_or_wait.check(REL)[0] == "SPAWN" + + +def test_call_id_unreadable_fails_closed_to_wait(): + # THE blocker-2 fix: a transport blip reading the call_id marker must not + # look like "no run recorded" and double-spawn a live 64GB container. + _reset(errors={f"{PREFIX}/call_id": ConnectionError("blip")}) + assert spawn_or_wait.check(REL)[0] == "WAIT" + + +def test_indeterminate_with_recent_marker_waits(): + _reset( + files={f"{PREFIX}/call_id": "call-1", f"{PREFIX}/started": _started(1)}, + call_get=ConnectionError("blip"), # not a terminal Modal error + ) + assert spawn_or_wait.check(REL)[0] == "WAIT" + + +def test_no_run_recorded_spawns(): + _reset() + assert spawn_or_wait.check(REL)[0] == "SPAWN" + + +def test_no_call_id_stale_marker_spawns(): + _reset(files={f"{PREFIX}/started": _started(30)}) # >24h + assert spawn_or_wait.check(REL)[0] == "SPAWN" + + +def test_no_call_id_recent_marker_waits(): + _reset(files={f"{PREFIX}/started": _started(1)}) + assert spawn_or_wait.check(REL)[0] == "WAIT" + + +def test_read_fails_closed_only_on_missing_file(): + # _read returns None only for a genuinely absent file; any other error + # propagates (so callers can fail closed). + _reset(errors={"x": ConnectionError("blip")}) + vol = spawn_or_wait.modal.Volume.from_name("vol") + assert spawn_or_wait._read(vol, "absent") is None + with pytest.raises(ConnectionError): + spawn_or_wait._read(vol, "x") diff --git a/tools/reform_validation/backfill.py b/tools/reform_validation/backfill.py new file mode 100644 index 00000000..22534a4d --- /dev/null +++ b/tools/reform_validation/backfill.py @@ -0,0 +1,487 @@ +"""Generate a reform_validation.json for a populace-us release, offline. + +Reproduces a release's reform_validation.json faithfully by running the producer +(microcosm.build.us_runtime.reform_validation.reform_validation_payload) on the +released populace_us_2024.h5 — the same call the build pipeline makes — so the +Scorecard can ingest a per-release regression point for any certified release, +including builds promoted to `latest` with the reform-validation step skipped. +The artifact lands under sources/populace-reform-validation/raw/.json +and feeds scorecard_db.ingest_reform_validation. Decoupled from the build +pipeline, so the slow simulation can never break a release. + +The producer builds ~45 fresh Microsimulations; some (budget_measure= +state_income_tax) compute all 50 states' income taxes and are memory-heavy. +Running them all in one process exhausts memory/disk, so this drives the work in +small batches, each a separate short-lived subprocess whose memory and temp +files are released on exit, checkpointing a partial per batch and merging at the +end (concat + dedup by id — identical to the monolithic payload). OBBBA stays in +one batch because its provisions are scored jointly in JCX-stacked order. + +Prereqs (the workflow sets these up): the exact policyengine-us / policyengine-core +versions from the release's release_manifest are installed, and the microcosm +package tree is importable (PYTHONPATH to packages/*/src of a microcosm checkout). + +Usage: + # full run: download H5, score every batch, merge, write reform_validation.json + python backfill.py --release-id --workdir [--producer-commit ] + # (blank --release-id resolves to latest.json) + python backfill.py --workdir --only # one batch (internal) + python backfill.py --workdir --merge # merge partials + python backfill.py --workdir --plan # print batch names +""" + +from __future__ import annotations + +import argparse +import gc +import hashlib +import json +import math +import os +import re +import shutil +import subprocess +import sys +import time +import urllib.request +from pathlib import Path + +HF_REPO = "policyengine/populace-us" +PERIOD = 2024 +CHUNK = 2 # baseline + 2 reforms/batch keeps heavy state_income_tax sims bounded +LEVELS_CHUNK = 16 # baseline-level specs per subprocess (7GB runner headroom) + + +# --------------------------------------------------------------------------- # +# HTTP / release metadata +# --------------------------------------------------------------------------- # +def _resolve_url(path: str, revision: str = "main") -> str: + return f"https://huggingface.co/datasets/{HF_REPO}/resolve/{revision}/{path}" + + +def _get_json(url: str) -> dict: + with urllib.request.urlopen(url, timeout=60) as r: + return json.loads(r.read()) + + +def latest_release_id() -> str: + return _get_json(_resolve_url("latest.json"))["release_id"] + + +def release_manifest(release_id: str) -> dict: + return _get_json(_resolve_url(f"releases/{release_id}/release_manifest.json")) + + +def _download(url: str, dest: Path, expect_sha256: str | None = None) -> None: + dest.parent.mkdir(parents=True, exist_ok=True) + h = hashlib.sha256() + with urllib.request.urlopen(url, timeout=300) as r, open(dest, "wb") as f: + while True: + chunk = r.read(1 << 20) + if not chunk: + break + f.write(chunk) + h.update(chunk) + if expect_sha256 and h.hexdigest() != expect_sha256: + raise SystemExit( + f"sha256 mismatch for {dest.name}: got {h.hexdigest()}, want {expect_sha256}" + ) + + +# --------------------------------------------------------------------------- # +# Workdir layout +# --------------------------------------------------------------------------- # +def _workdir() -> Path: + return Path(os.environ["RV_WORKDIR"]).resolve() + + +def _release_id() -> str: + return os.environ["RV_RELEASE_ID"] + + +def _paths(): + wd = _workdir() + return ( + wd, + wd / "populace_us_2024.h5", + wd / "calibration_diagnostics.json", + wd / "partials", + ) + + +# --------------------------------------------------------------------------- # +# Producer glue (faithful to tools/build_us_fiscal_refresh_release.py) +# --------------------------------------------------------------------------- # +def cached_simulate_factory(dataset_path: Path): + """Load the H5 once, reuse for every reform in this (short-lived) process.""" + from policyengine_us import Microsimulation + from policyengine_us.data import USSingleYearDataset + + dataset = USSingleYearDataset(file_path=str(dataset_path)) + + def simulate(reform): + t0 = time.time() + sim = ( + Microsimulation(dataset=dataset) + if reform is None + else Microsimulation(dataset=dataset, reform=reform) + ) + print(f" [sim {time.time() - t0:.0f}s]", flush=True) + return sim + + return simulate + + +def in_sample_dicts(diag_path: Path): + """Reproduce _in_sample_estimates/_in_sample_targets(result) from the published + calibration_diagnostics.json. Serialized target names carry an @ suffix + the in-memory target.name lacks — strip it to match the JCT in-sample spec ids.""" + diag = json.loads(diag_path.read_text()) + est, tgt = {}, {} + for t in diag["targets"]: + name = re.sub(r"@\d+$", "", t["name"]) + fe, tv = t.get("final_estimate"), t.get("target") + if fe is not None and math.isfinite(float(fe)): + est[name] = float(fe) + if tv is not None and math.isfinite(float(tv)): + tgt[name] = float(tv) + return est, tgt + + +def _is_obbba(spec): + from microcosm.build.us_runtime.reform_validation import _is_obbba_spec + + return _is_obbba_spec(spec) + + +def _batches(specs): + """(name, subset_specs, levels_slice). OBBBA scored jointly; in-sample rows + ride with it (they read the estimates dict, no sim); remaining reforms chunk by + CHUNK. Baseline-levels chunk by LEVELS_CHUNK: the suite grew from 74 to ~176 + specs when the Census SPM state-rate backtests joined + default_baseline_level_specs(), and computing them all against one + accumulating baseline sim OOM-kills the 7GB ubuntu-latest runner. Each + levels chunk is a (start, stop) slice into the spec tuple, scored in its + own short-lived subprocess with a fresh baseline.""" + from microcosm.build.us_runtime.reform_validation import ( + default_baseline_level_specs, + ) + + obbba = [s for s in specs if _is_obbba(s)] + insample = [s for s in specs if s.in_sample and not _is_obbba(s)] + rest = [s for s in specs if not _is_obbba(s) and not s.in_sample] + n_levels = len(default_baseline_level_specs()) + for i in range(0, n_levels, LEVELS_CHUNK): + yield (f"levels{i // LEVELS_CHUNK}", [], (i, i + LEVELS_CHUNK)) + yield ("obbba", obbba + insample, None) + for i in range(0, len(rest), CHUNK): + yield (f"rest{i // CHUNK}", rest[i : i + CHUNK], None) + + +def _load_specs(): + from microcosm.build.us_runtime.reform_validation import load_default_reform_specs + + return load_default_reform_specs(period=PERIOD) + + +def run_batch(name, subset, levels_slice): + from microcosm.build.us_runtime.reform_validation import ( + default_baseline_level_specs, + reform_validation_payload, + ) + + _, h5, diag, part_dir = _paths() + part_dir.mkdir(exist_ok=True) + part = part_dir / f"{name}.json" + if part.exists(): + print(f"[skip] {name}", flush=True) + return + est, tgt = in_sample_dicts(diag) + levels = ( + default_baseline_level_specs()[levels_slice[0] : levels_slice[1]] + if levels_slice is not None + else () + ) + print(f"[run ] {name}: {len(subset)} specs levels={len(levels)}", flush=True) + payload = reform_validation_payload( + subset, + period=PERIOD, + simulate=cached_simulate_factory(h5), + in_sample_estimates=est, + in_sample_targets=tgt, + baseline_levels=levels, + release_id=_release_id(), + ) + # Stamp BOTH revisions this partial was scored under so merge() can refuse + # a workdir whose partials span two revisions: the producer (a resume that + # crossed a microcosm main bump) and the driver (backfill.py is re-baked + # from the current scorecard checkout on every `modal deploy`, so two + # respawns can run different driver code under one producer stamp). + payload["_producer_commit"] = os.environ.get("RV_PRODUCER_COMMIT", "unknown") + payload["_driver_commit"] = os.environ.get("RV_DRIVER_COMMIT", "unknown") + part.write_text(json.dumps(payload, indent=1, allow_nan=False)) + print(f"[done] {name}: {len(payload['reforms'])} rows", flush=True) + del payload + gc.collect() + + +def _backfill_note(manifest: dict, producer_commit: str, h5_sha: str) -> str: + b = manifest["build"] + return ( + "Full offline reproduction of reform_validation.json for a release that " + "shipped without one (the build was promoted to latest with the " + "reform-validation step skipped). Generated by " + "tools/reform_validation/backfill.py (scheduled workflow " + "reform-validation-backfill.yml) running the producer " + "(microcosm.build.us_runtime.reform_validation.reform_validation_payload) on " + f"the released populace_us_2024.h5 (sha256 {h5_sha}) under the EXACT build " + f"versions policyengine-us {b['built_with_model_package']['version']} / " + f"policyengine-core {b['built_with_core_package']['version']} (per " + "release_manifest), so no version drift. In-sample JCT rows come from the " + "release's own calibration_diagnostics.json (identical to a native build); " + "out-of-sample rows (OBBBA JCX-stacked, tax-expenditure, state-program and " + "state-reform repeals) and the baseline-level backtests (IRS SOI actual, " + "federal-EITC-by-state, state-program actual) were simulated on the H5. " + f"Producer run at microcosm {producer_commit}. Computed in small subprocess " + "batches for memory/disk headroom; scoring is identical to the monolithic " + "producer (non-OBBBA reforms independent; OBBBA scored jointly stacked)." + ) + + +def _observed_engine(manifest: dict, version=None) -> dict: + """The policyengine-us/-core versions INSTALLED in this environment, + verified against the release manifest. + + The Modal app installs the engines from its own (install-time) manifest + fetch, and this driver re-fetches the manifest at provenance time. The + attestation must stamp what actually ran (re-gate round 3, finding 2), so + read the installed versions back and refuse if they differ from the + manifest being stamped. That covers a manifest that moved between the two + fetches and a dependency that overrode the pin. ``version`` is + injectable for tests (default: importlib.metadata.version).""" + if version is None: + from importlib.metadata import version + b = manifest["build"] + expected = { + "policyengine_us": b["built_with_model_package"]["version"], + "policyengine_core": b["built_with_core_package"]["version"], + } + observed = { + "policyengine_us": version("policyengine-us"), + "policyengine_core": version("policyengine-core"), + } + if observed != expected: + raise SystemExit( + f"installed engines {observed} != release manifest {expected} — the " + "manifest moved since install or a dependency overrode the pin; " + "refusing to stamp an attestation that doesn't match what ran." + ) + return observed + + +def _revision_dirty(commits: set) -> bool: + """A partial-revision set is unusable if it is empty, mixes revisions, or + contains a missing/unknown stamp.""" + return not commits or None in commits or "unknown" in commits or len(commits) > 1 + + +def _refuse_if_mixed( + producer_commits: set, driver_commits: set, part_dir: Path +) -> None: + """Every partial in a workdir must carry a SINGLE, KNOWN producer and + driver revision. A resume that crossed a microcosm bump (producer) or + re-``modal deploy``ed a different scorecard checkout (driver) resumes in + the same release-keyed workdir and would merge two revisions while + labelling the artifact with one. Refuse on any missing/unknown OR mixed + stamp — and PURGE the partials so the next respawn recomputes clean, + instead of wedging in a silent refuse loop where every tick fails + identically (blocker 4). Pure stdlib so it is unit-testable off-Modal.""" + if _revision_dirty(producer_commits) or _revision_dirty(driver_commits): + shutil.rmtree(part_dir, ignore_errors=True) + raise SystemExit( + "partials span multiple or unstamped revisions (producer=" + f"{sorted(str(c) for c in producer_commits)}, driver=" + f"{sorted(str(c) for c in driver_commits)}) — purged the workdir " + "partials; the next run recomputes at a single revision." + ) + + +def merge(manifest: dict, producer_commit: str, h5_sha: str) -> Path: + from microcosm.build.us_runtime.reform_validation import write_reform_validation + + wd, _, _, part_dir = _paths() + names = [n for n, _, _ in _batches(_load_specs())] + seen, rows, header = set(), [], None + producer_commits, driver_commits = set(), set() + for name in names: + p = json.loads((part_dir / f"{name}.json").read_text()) + producer_commits.add(p.get("_producer_commit")) + driver_commits.add(p.get("_driver_commit")) + if header is None: + header = { + k: v + for k, v in p.items() + if k not in ("reforms", "_producer_commit", "_driver_commit") + } + for r in p["reforms"]: + rid = r.get("id") + if rid is not None and rid in seen: + continue + if rid is not None: + seen.add(rid) + rows.append(r) + _refuse_if_mixed(producer_commits, driver_commits, part_dir) + # _refuse_if_mixed guarantees a single known revision on each axis; require + # it to EQUAL the one going into the attestation. Otherwise all-old partials + # could be relabeled under a new driver/producer at merge time (re-gate + # blocker 4 — the mixed-check alone doesn't pin the singleton's value). + driver_commit = os.environ.get("RV_DRIVER_COMMIT", "unknown") + for axis, got, want in ( + ("producer", producer_commits, producer_commit), + ("driver", driver_commits, driver_commit), + ): + if got != {want}: + shutil.rmtree(part_dir, ignore_errors=True) + raise SystemExit( + f"partials {axis} revision {got} != attested {want!r} — purged " + "the workdir; the next run recomputes under the attested revision." + ) + header["reforms"] = rows + header["_backfill_note"] = _backfill_note(manifest, producer_commit, h5_sha) + # Structured engine pin (scorecard ingest reads this to resolve a new + # release's policyengine-us/-core versions without a code edit — the + # prose _backfill_note carries the same fact for humans). These are the + # versions read back from the environment the batches ran in, verified + # equal to the release_manifest (observed, not declared). + engine = _observed_engine(manifest) + header["engine"] = dict(engine) + # Explicit, validated scoring mode (ingest fails loudly if absent, never + # defaults — issue #53 re-gate blocker 5). backfill.py always drives the + # producer with OBBBA scored jointly in one JCX-stacked batch. + header["obbba_scoring_mode"] = "jcx_stacked" + # Provenance attestation binding the compute inputs (issue #53 re-gate + # blocker 6). The Modal app augments this with modal_* execution fields; + # the workflow records the scorecard commit + artifact sha256 on the PR. + header["_attestation"] = { + "release_id": _release_id(), + "producer_commit": producer_commit, + "driver_commit": os.environ.get("RV_DRIVER_COMMIT", "unknown"), + "policyengine_us": engine["policyengine_us"], + "policyengine_core": engine["policyengine_core"], + "h5_sha256": h5_sha, + "n_rows": len(rows), + "batches": names, + } + out = wd / "reform_validation.json" + write_reform_validation(header, out) + print(f"MERGED {len(rows)} rows -> {out}", flush=True) + return out + + +# --------------------------------------------------------------------------- # +# Orchestration (default mode): download, batch-in-subprocesses, merge +# --------------------------------------------------------------------------- # +def _cleanup_temp() -> None: + """pe-us leaves small tmp*.h5 in TMPDIR; sweep the ones no longer in use.""" + import glob + import tempfile + + for f in glob.glob(os.path.join(tempfile.gettempdir(), "tmp*.h5")): + try: + if time.time() - os.path.getmtime(f) > 120: + os.remove(f) + except OSError: + pass + + +def _sha256_file(path: Path) -> str: + h = hashlib.sha256() + with open(path, "rb") as f: + for chunk in iter(lambda: f.read(1 << 20), b""): + h.update(chunk) + return h.hexdigest() + + +def orchestrate(release_id: str, producer_commit: str) -> Path: + wd, h5, diag, _ = _paths() + manifest = release_manifest(release_id) + # Fail fast, before hours of compute, if the provenance-time manifest no + # longer matches the engines installed from the install-time fetch. + _observed_engine(manifest) + art = manifest["artifacts"]["populace_us_2024"] + if not h5.exists(): + print(f"downloading H5 ({art['sha256'][:12]}…)…", flush=True) + _download(_resolve_url(art["path"], art["revision"]), h5, art["sha256"]) + # Rehash the ACTUAL H5 — downloaded or CACHED — and verify against the + # manifest. A cached file must not be trusted on filename alone, and the + # attestation stamps the OBSERVED hash rather than the manifest's expected + # one (re-gate blocker 2). + h5_sha = _sha256_file(h5) + if h5_sha != art["sha256"]: + raise SystemExit( + f"H5 sha256 {h5_sha} != manifest {art['sha256']} — cached/corrupt file; " + "delete the workdir H5 and re-run." + ) + if not diag.exists(): + _download( + _resolve_url(f"releases/{release_id}/calibration_diagnostics.json"), diag + ) + plan = [n for n, _, _ in _batches(_load_specs())] + for name in plan: + part = wd / "partials" / f"{name}.json" + for attempt in range(1, 7): + if part.exists(): + break + print(f"=== batch {name} attempt {attempt} ===", flush=True) + subprocess.run( + [sys.executable, __file__, "--workdir", str(wd), "--only", name], + env={ + **os.environ, + "RV_WORKDIR": str(wd), + "RV_RELEASE_ID": release_id, + "RV_PRODUCER_COMMIT": producer_commit, + }, + ) + _cleanup_temp() + if not part.exists(): + print(f"--- {name} produced no partial; settling 20s ---", flush=True) + time.sleep(20) + if not part.exists(): + raise SystemExit(f"batch {name} failed after 6 attempts") + return merge(manifest, producer_commit, h5_sha) + + +# --------------------------------------------------------------------------- # +def main(): + ap = argparse.ArgumentParser() + ap.add_argument("--release-id", default="") + ap.add_argument("--workdir", default=os.environ.get("RV_WORKDIR", ".")) + ap.add_argument( + "--producer-commit", default=os.environ.get("RV_PRODUCER_COMMIT", "unknown") + ) + ap.add_argument("--only") + ap.add_argument("--merge", action="store_true") + ap.add_argument("--plan", action="store_true") + args = ap.parse_args() + + os.environ["RV_WORKDIR"] = str(Path(args.workdir).resolve()) + release_id = ( + args.release_id or os.environ.get("RV_RELEASE_ID") or latest_release_id() + ) + os.environ["RV_RELEASE_ID"] = release_id + + if args.plan: + print(" ".join(n for n, _, _ in _batches(_load_specs()))) + return + if args.only: + name, subset, lv = next(b for b in _batches(_load_specs()) if b[0] == args.only) + run_batch(name, subset, lv) + return + if args.merge: + merge(release_manifest(release_id), args.producer_commit, "unknown") + return + orchestrate(release_id, args.producer_commit) + + +if __name__ == "__main__": + main() diff --git a/tools/reform_validation/modal_backfill_app.py b/tools/reform_validation/modal_backfill_app.py new file mode 100644 index 00000000..2b175848 --- /dev/null +++ b/tools/reform_validation/modal_backfill_app.py @@ -0,0 +1,253 @@ +"""Modal app that runs the reform-validation backfill for a microcosm release. + +Why Modal: the producer needs one full Microsimulation per batch plus the +microcosm/torch import stack, which no longer fits GitHub's 7GB ubuntu-latest +runner — every scheduled run since 2026-07-21 died to the runner's OOM +shutdown signal, including post-chunking (#107) runs where even a single +16-spec levels chunk was killed before its first simulation finished. This +app runs the IDENTICAL producer path (tools/reform_validation/backfill.py, +mounted verbatim from the repo checkout at deploy time) in a 64GB container. + +Division of labor with the scheduled workflow +(.github/workflows/reform-validation-backfill.yml): + + GitHub Actions resolve release id -> spawn this app (fire-and- + (trigger/commit layer) forget) -> on a LATER tick, harvest the finished + artifact from the Volume -> land it under + sources/populace-reform-validation/raw/, verify the + DB builds (scorecard_db.build_db to a throwaway + path — the DB is a derived artifact post-#74, never + committed), commit the raw artifact + source.json, + open a PR. No simulation runs on the runner, so + ticks take seconds. + + This app clone microcosm at the requested ref, install the + (simulation layer) release-exact policyengine-us/-core from the + release manifest at runtime, drive backfill.py + with a Volume-backed workdir (batch partials + survive interruptions and resume), and publish + reform_validation_.json to the Volume. + +Engine versions are installed at RUNTIME (not baked into the image) because +they are release-exact and change per release; the image carries only the +heavy version-stable dependencies. The producer ref is an argument for the +same reason — no image rebuild per microcosm commit. + +Manual use: + modal deploy tools/reform_validation/modal_backfill_app.py + python -c " + import modal + fn = modal.Function.from_name('scorecard-reform-validation-backfill', 'backfill') + print(fn.spawn('', '').object_id)" + +Requires Modal credentials (MODAL_TOKEN_ID / MODAL_TOKEN_SECRET) for the +PolicyEngine workspace. +""" + +from pathlib import Path + +import modal + +APP_NAME = "scorecard-reform-validation-backfill" +app = modal.App(APP_NAME) + +VOLUME_NAME = "scorecard-reform-validation" +volume = modal.Volume.from_name(VOLUME_NAME, create_if_missing=True) + +# Heavy, version-stable dependencies only. The release-exact engine +# (policyengine-us / policyengine-core) installs at runtime per release. +image = ( + modal.Image.debian_slim(python_version="3.13") + .apt_install("git") + .pip_install( + "tables", + "microdf-python", + "h5py", + "numpy", + "pandas", + "scipy", + "scikit-learn", + "tqdm", + "requests", + ) + .pip_install("torch", index_url="https://download.pytorch.org/whl/cpu") + .add_local_file(str(Path(__file__).parent / "backfill.py"), "/root/backfill.py") +) + +PRODUCER_PACKAGES = ( + "microcosm-build", + "microcosm-frame", + "microcosm-calibrate", + "microcosm-fit", + "microcosm-data", +) + + +@app.function( + image=image, timeout=8 * 3600, memory=65536, cpu=8.0, volumes={"/vol": volume} +) +def backfill( + release_id: str, producer_ref: str = "main", driver_ref: str = "unknown" +) -> str: + """Produce reform_validation.json for ``release_id`` on the Volume. + + Writes, in order: + /vol/rv_/... durable workdir (H5, partials) + /vol/reform_validation_.json the finished artifact + The workdir's ``started`` marker (timestamped) plus the spawner-recorded + ``call_id`` let the workflow distinguish in-flight from dead runs + (spawn_or_wait.py) instead of trusting the marker alone. + """ + import hashlib + import json + import os + import subprocess + import time + import urllib.request + + subprocess.run( + ["git", "clone", "https://github.com/PolicyEngine/microcosm", "/opt/microcosm"], + check=True, + ) + subprocess.run( + ["git", "-C", "/opt/microcosm", "checkout", producer_ref], check=True + ) + producer_commit = subprocess.run( + ["git", "-C", "/opt/microcosm", "rev-parse", "--short", "HEAD"], + capture_output=True, + text=True, + check=True, + ).stdout.strip() + + manifest_url = ( + "https://huggingface.co/datasets/policyengine/populace-us/resolve/main/" + f"releases/{release_id}/release_manifest.json" + ) + with urllib.request.urlopen(manifest_url, timeout=60) as r: + build = json.loads(r.read())["build"] + peus = build["built_with_model_package"]["version"] + pecore = build["built_with_core_package"]["version"] + subprocess.run( + [ + "pip", + "install", + "--quiet", + f"policyengine-us=={peus}", + f"policyengine-core=={pecore}", + ], + check=True, + ) + print( + f"engines installed: policyengine-us {peus} / policyengine-core {pecore}", + flush=True, + ) + + # Install the checked-out microcosm shards so THEIR declared dependencies + # resolve — microcosm.fit eagerly imports quantile-forest, which putting the + # src dirs on PYTHONPATH alone never installed, so the producer couldn't + # start. A constraints file keeps the release-exact engines pinned while the + # shards' own deps install. + with open("/tmp/constraints.txt", "w") as c: + c.write(f"policyengine-us=={peus}\npolicyengine-core=={pecore}\n") + subprocess.run( + ["pip", "install", "--quiet", "-c", "/tmp/constraints.txt"] + + [f"/opt/microcosm/packages/{p}" for p in PRODUCER_PACKAGES], + check=True, + ) + + # Read back the ACTUAL installed engine versions (not the manifest's + # expected) so the attestation stamps what really ran (blocker: attestation + # must be verified, not declarative). A shard dep must not have pulled a + # different engine in past the constraint. + def _installed(pkg): + out = subprocess.run( + ["pip", "show", pkg], capture_output=True, text=True, check=True + ).stdout + for line in out.splitlines(): + if line.startswith("Version:"): + return line.split(":", 1)[1].strip() + return "" + + installed_peus, installed_pecore = ( + _installed("policyengine-us"), + _installed("policyengine-core"), + ) + if (installed_peus, installed_pecore) != (peus, pecore): + raise RuntimeError( + f"installed engines {installed_peus}/{installed_pecore} != manifest " + f"{peus}/{pecore} — a shard dependency overrode the release pin" + ) + + # Smoke the plan BEFORE the multi-hour run: this imports the producer module + # and every shard dependency, so a missing dep (blocker 1) fails fast and + # loud here instead of deep inside a batch. + subprocess.run( + ["python", "/root/backfill.py", "--plan", "--release-id", release_id], + check=True, + ) + + workdir = f"/vol/rv_{hashlib.sha256(release_id.encode()).hexdigest()[:8]}" + os.makedirs(workdir, exist_ok=True) + with open(f"{workdir}/started", "w") as f: + f.write( + f"{release_id}\nproducer {producer_commit}\nstarted_at {int(time.time())}\n" + ) + volume.commit() + + # No PYTHONPATH shard-src hack: the shards are pip-installed above, so their + # deps resolve. Only the driver revision is threaded through — the scorecard + # checkout backfill.py was baked from at `modal deploy` time; merge() refuses + # partials that mix driver revisions (blocker 4). + env = {**os.environ, "RV_DRIVER_COMMIT": driver_ref} + proc = subprocess.run( + [ + "python", + "/root/backfill.py", + "--release-id", + release_id, + "--workdir", + workdir, + "--producer-commit", + producer_commit, + ], + env=env, + ) + volume.commit() # keep partials even on failure so a re-spawn resumes + proc.check_returncode() + + artifact = json.loads(Path(workdir, "reform_validation.json").read_text()) + n_rows = len(artifact["reforms"]) + # Bind the compute environment into the attestation backfill.py started + # (blocker 6): which Modal app/call/image and producer ref actually + # produced this artifact. Defensive getattr — the call-id accessor has + # moved across Modal client versions. + call_id = "" + for getter in ("current_function_call_id", "current_call_id"): + fn = getattr(modal, getter, None) + if callable(fn): + try: + call_id = fn() or "" + except Exception: + call_id = "" + break + att = artifact.setdefault("_attestation", {}) + att.update( + { + "modal_app": APP_NAME, + # Real, container-populated identifiers only. The old `modal_image_id` + # read MODAL_IMAGE_ID, which the pinned client never sets (always + # ""), so it was a forged-provenance no-op — dropped (blocker 6). + "modal_call_id": call_id or os.environ.get("MODAL_TASK_ID", ""), + "modal_task_id": os.environ.get("MODAL_TASK_ID", ""), + "producer_ref": producer_ref, + "driver_ref": driver_ref, + } + ) + with open(f"/vol/reform_validation_{release_id}.json", "w") as f: + json.dump(artifact, f, indent=1) + volume.commit() + print( + f"published {n_rows} rows for {release_id} (producer {producer_commit})", + flush=True, + ) + return f"{n_rows} rows" diff --git a/tools/reform_validation/register_release.py b/tools/reform_validation/register_release.py new file mode 100644 index 00000000..29065ad6 --- /dev/null +++ b/tools/reform_validation/register_release.py @@ -0,0 +1,86 @@ +"""Land a harvested reform_validation.json into the Scorecard's source tree. + +The Modal backfill (modal_backfill_app.py) produces one release's +reform_validation.json; this drops it under +``sources/populace-reform-validation/raw/.json`` and adds the +release to that source's ``source.json`` ``releases`` list (deduped, ordered +oldest-first by the release timestamp — the same order the ingest globs in). +It does NOT run the ingest; the workflow does that next so the rebuilt DB is +part of the same commit. + +Usage: + python tools/reform_validation/register_release.py +""" + +from __future__ import annotations + +import json +import sys +from pathlib import Path + +REPO = Path(__file__).resolve().parent.parent.parent +SOURCE_DIR = REPO / "sources" / "populace-reform-validation" +RAW_DIR = SOURCE_DIR / "raw" +SOURCE_JSON = SOURCE_DIR / "source.json" + + +def _timestamp_key(release_id: str) -> str: + # Reuse the ingest's parser so raw/ ordering and the releases list agree. + from scorecard_db.ingest_reform_validation import _release_timestamp + + return _release_timestamp(release_id) + + +def register( + release_id: str, + artifact_path: Path, + *, + raw_dir: Path = RAW_DIR, + source_json: Path = SOURCE_JSON, +) -> dict: + """Copy the artifact into raw/ and register the release in source.json. + + Returns a summary dict. Idempotent: re-registering an existing release + overwrites its raw file and leaves the releases list unchanged. + """ + artifact = json.loads(Path(artifact_path).read_text(encoding="utf-8")) + if artifact.get("release_id") != release_id: + raise SystemExit( + f"artifact release_id {artifact.get('release_id')!r} != {release_id!r}" + ) + _timestamp_key(release_id) # fail fast if the id has no parseable timestamp + + raw_dir.mkdir(parents=True, exist_ok=True) + dest = raw_dir / f"{release_id}.json" + dest.write_text(json.dumps(artifact, indent=1), encoding="utf-8") + + doc = json.loads(source_json.read_text(encoding="utf-8")) + releases = doc.get("releases", []) + added = release_id not in releases + if added: + releases.append(release_id) + doc["releases"] = sorted(set(releases), key=_timestamp_key) + source_json.write_text( + json.dumps(doc, indent=2, ensure_ascii=False) + "\n", encoding="utf-8" + ) + try: + raw_file = str(dest.relative_to(REPO)) + except ValueError: # a caller (tests) may write outside the repo tree + raw_file = str(dest) + return { + "release_id": release_id, + "raw_file": raw_file, + "added_to_source": added, + "releases": doc["releases"], + } + + +def main() -> None: + if len(sys.argv) != 3: + raise SystemExit("usage: register_release.py ") + summary = register(sys.argv[1], Path(sys.argv[2])) + print(json.dumps(summary, indent=1)) + + +if __name__ == "__main__": + main() diff --git a/tools/reform_validation/spawn_or_wait.py b/tools/reform_validation/spawn_or_wait.py new file mode 100644 index 00000000..3d527f64 --- /dev/null +++ b/tools/reform_validation/spawn_or_wait.py @@ -0,0 +1,282 @@ +"""Decide whether a Modal backfill run is in flight, dead, or missing — and +spawn (or re-spawn) it when needed. + +The app writes ``rv_/started`` when a run begins, and the spawner +records the Modal call id at ``rv_/call_id``. A marker alone is not +proof of life: if the producer dies after the marker is written, every later +tick would see "in flight" and nothing would ever retry (PR #112 review). So +``check`` asks Modal for the recorded call's status first, and only falls +back to the marker's age (24h, comfortably past the 8h function timeout) +when no call id is readable. Re-spawning is always safe: the Volume-backed +workdir checkpoints batch partials, so a re-spawn resumes rather than +restarts. + +Usage (from the repo root, Modal credentials in the environment): + python3 tools/reform_validation/spawn_or_wait.py check + prints WAIT or SPAWN on stdout (reasons go to stderr) + python3 tools/reform_validation/spawn_or_wait.py spawn + spawns the backfill and records its call id on the Volume +""" + +import hashlib +import re +import sys +import tempfile +import time +from pathlib import Path + +import modal +import modal.exception as _mexc + +from modal_backfill_app import APP_NAME, VOLUME_NAME + +STALE_AFTER_SECONDS = 24 * 3600 + +# Modal errors that mean the recorded call is genuinely dead/finished-failed +# (as opposed to a transport blip we can't interpret). On these we re-spawn; +# on anything else unexpected we treat the status as indeterminate and lean +# on the marker-age backstop rather than respawning live work (blocker 2). +# Built defensively so a Modal client that renames/omits one of these doesn't +# break the import. +_TERMINAL_CALL_ERRORS = tuple( + e + for e in ( + getattr(_mexc, "RemoteError", None), + getattr(_mexc, "ExecutionError", None), + getattr(_mexc, "FunctionTimeoutError", None), + getattr(_mexc, "NotFoundError", None), + getattr(_mexc, "InvalidError", None), + ) + if isinstance(e, type) and issubclass(e, BaseException) +) + + +def _sha8(release_id: str) -> str: + return hashlib.sha256(release_id.encode()).hexdigest()[:8] + + +def _read(vol: modal.Volume, path: str) -> str | None: + """Read a small marker file. Returns None ONLY when the file is genuinely + absent (FileNotFoundError); any other error (a transport blip) propagates, + so callers fail closed instead of mistaking an unreadable marker for a + missing one. Collapsing every error to None was the shared root of the + blocker-2 double-spawn and the blocker-4 pin overwrite: a live run's + call_id/producer_ref that momentarily fails to read looked like 'no run + recorded' and triggered a respawn / fresh-main re-pin.""" + try: + return b"".join(vol.read_file(path)).decode() + except FileNotFoundError: + return None + + +def _artifact_present(vol: modal.Volume, release_id: str) -> bool: + """Whether the finished ``reform_validation_.json`` is on the Volume. + Used to guard the healthy-completion TOCTOU (blocker 2): a call that + returns just after the drain listed the Volume has published its artifact, + so a completed call is not proof the work still needs redoing.""" + target = f"reform_validation_{release_id}.json" + for entry in vol.listdir("/"): + if getattr(entry, "path", str(entry)).rsplit("/", 1)[-1] == target: + return True + return False + + +def check(release_id: str) -> tuple[str, str]: + vol = modal.Volume.from_name(VOLUME_NAME, create_if_missing=True) + prefix = f"rv_{_sha8(release_id)}" + + # Consult the recorded call id FIRST. The spawner writes call_id right + # after fn.spawn(), but the app only writes the `started` marker after + # git clone + engine install (minutes later). Checking the marker first + # (old behaviour) returned SPAWN in that window and double-spawned a live + # run (blocker 2). An unreadable call_id marker fails closed to WAIT — it + # is NOT proof the run is absent. + try: + call_id = _read(vol, f"{prefix}/call_id") + except Exception as e: + return ( + "WAIT", + f"call_id marker unreadable ({type(e).__name__}) — failing closed", + ) + indeterminate = None + if call_id: + call_id = call_id.strip() + try: + modal.FunctionCall.from_id(call_id).get(timeout=0) + except TimeoutError: + return "WAIT", f"call {call_id} is still running" + except _TERMINAL_CALL_ERRORS as e: + return ( + "SPAWN", + f"call {call_id} terminally failed ({type(e).__name__}) — " + "re-spawning; the workdir checkpoints make this a resume", + ) + except Exception as e: + # Could not determine status — a transport/transient error, NOT + # proof of death. Don't respawn live work on a blip; fall through + # to the marker-age backstop. + indeterminate = f"{type(e).__name__}: {e}" + else: + # Completed. Guard the TOCTOU: the drain may have listed the Volume + # just before the app published, so a completed call is not proof + # the artifact is missing — re-check before respawning finished + # work. An unreadable re-check fails closed to WAIT. + try: + published = _artifact_present(vol, release_id) + except Exception as e: + return ( + "WAIT", + f"call {call_id} completed; artifact re-check failed " + f"({type(e).__name__}) — failing closed", + ) + if published: + return ( + "WAIT", + f"call {call_id} finished and the artifact is present — " + "the drain will harvest it", + ) + return ( + "SPAWN", + f"call {call_id} finished but no artifact was published — re-spawning", + ) + + try: + marker = _read(vol, f"{prefix}/started") + except Exception as e: + return ( + "WAIT", + f"started marker unreadable ({type(e).__name__}) — failing closed", + ) + if marker is None and call_id is None: + return "SPAWN", "no run recorded for this release" + + started_at = None + for line in (marker or "").splitlines(): + if line.startswith("started_at "): + started_at = float(line.split()[1]) + age = None if started_at is None else time.time() - started_at + if age is not None and age < STALE_AFTER_SECONDS: + tail = f"; call status indeterminate ({indeterminate})" if indeterminate else "" + return "WAIT", f"marker is {age / 3600:.1f}h old{tail}" + if age is None and call_id and indeterminate: + # A call id exists and we merely couldn't reach Modal (no timestamp + # yet either) — presume alive rather than respawn a just-spawned run. + return ( + "WAIT", + f"call {call_id} recorded; status indeterminate ({indeterminate})", + ) + return ( + "SPAWN", + "marker stale (>24h or unstamped) — re-spawning; the workdir " + "checkpoints make this a resume", + ) + + +def spawn(release_id: str, producer_ref: str, driver_ref: str = "unknown") -> None: + vol = modal.Volume.from_name(VOLUME_NAME, create_if_missing=True) + prefix = f"rv_{_sha8(release_id)}" + # Pin the producer ref for the life of this release: the FIRST spawn + # records it, every re-spawn reuses it. Otherwise a re-spawn would + # resolve a fresh microcosm main SHA and, resuming in the same + # release-keyed workdir, merge partials from two producer revisions + # (blocker 4). The workflow passes the current main SHA as the candidate; + # a recorded ref overrides it. The pin read fails CLOSED: an unreadable + # producer_ref aborts rather than silently re-resolving fresh main and + # overwriting the pin (the blocker-2 mechanism reappearing inside b4). + try: + recorded = _read(vol, f"{prefix}/producer_ref") + except Exception as e: + raise SystemExit( + f"producer_ref unreadable ({type(e).__name__}) — refusing to spawn " + "rather than risk overwriting the pin with a fresh ref" + ) + ref = recorded.strip() if recorded else producer_ref + + fn = modal.Function.from_name(APP_NAME, "backfill") + call = fn.spawn(release_id, ref, driver_ref) + with tempfile.TemporaryDirectory() as tmp: + call_p = Path(tmp, "call_id") + call_p.write_text(call.object_id) + # release_id in the workdir lets the schedule re-find a run that was + # spawned, became non-latest, and died before publishing (blocker 3) — + # sha8 is one-way, so the id has to be stored to be enumerable. + rid_p = Path(tmp, "release_id") + rid_p.write_text(release_id) + with vol.batch_upload(force=True) as batch: + batch.put_file(call_p, f"{prefix}/call_id") + batch.put_file(rid_p, f"{prefix}/release_id") + if recorded is None: + ref_p = Path(tmp, "producer_ref") + ref_p.write_text(ref) + batch.put_file(ref_p, f"{prefix}/producer_ref") + pinned = " (pinned from first spawn)" if recorded else "" + print( + f"spawned Modal backfill {call.object_id} for {release_id} " + f"at microcosm {ref[:9]}{pinned}", + file=sys.stderr, + ) + + +def _id_timestamp(release_id: str) -> str: + # Trailing YYYYMMDD(THHMMSSZ) — enough to order releases oldest-first + # without importing the ingest (kept dependency-free for the workflow). + m = re.search(r"(\d{8})(T\d{6}Z)?$", release_id) + return m.group(0) if m else release_id + + +def list_artifacts() -> list[str]: + """Release ids that have a published reform_validation_.json on the + Volume, oldest-first. The workflow drains these into raw/ regardless of + which release is currently ``latest`` — so a run that finished after + ``latest`` advanced is never stranded (blocker 3).""" + vol = modal.Volume.from_name(VOLUME_NAME, create_if_missing=True) + ids = [] + for entry in vol.listdir("/"): + name = getattr(entry, "path", str(entry)).rsplit("/", 1)[-1] + if name.startswith("reform_validation_") and name.endswith(".json"): + ids.append(name[len("reform_validation_") : -len(".json")]) + return sorted(ids, key=_id_timestamp) + + +def list_pending() -> list[str]: + """Release ids that have a spawned workdir (``rv_/release_id``) but + no published artifact yet — runs in flight OR dead-before-publish. The + schedule uses this to revisit a release that became non-latest and then + died before publishing, which the latest-only spawn path would never + retry (blocker 3). Oldest-first; artifacts already draining are excluded.""" + vol = modal.Volume.from_name(VOLUME_NAME, create_if_missing=True) + published = set(list_artifacts()) + ids = [] + for entry in vol.listdir("/"): + name = getattr(entry, "path", str(entry)).rstrip("/").rsplit("/", 1)[-1] + if not name.startswith("rv_"): + continue + rid = _read(vol, f"{name}/release_id") # None if not written yet + if rid and rid.strip() and rid.strip() not in published: + ids.append(rid.strip()) + return sorted(set(ids), key=_id_timestamp) + + +def main() -> None: + mode = sys.argv[1] + if mode == "check": + decision, reason = check(sys.argv[2]) + print(reason, file=sys.stderr) + print(decision) + elif mode == "spawn": + # spawn [driver_ref] + spawn(sys.argv[2], sys.argv[3], sys.argv[4] if len(sys.argv) > 4 else "unknown") + elif mode == "artifacts": + for rid in list_artifacts(): + print(rid) + elif mode == "pending": + for rid in list_pending(): + print(rid) + else: + raise SystemExit( + f"unknown mode {mode!r}; use check, spawn, artifacts or pending" + ) + + +if __name__ == "__main__": + main()