Bind the ACS local artifact to the code that wrote it

changelog.d/1263-us-acs-local-artifact-writer-binding.changed.md

@@ -0,0 +1 @@
+The ACS local-area release's package stage now refuses to name a release after code that did not write its calibrated artifact: calibrate records its code identity in `consumer_export.json`, and package requires the same commit. The interim WIC limitation now states exactly whom the assumption covers, and its receipt records how many people were allowed to lack a decision.

docs/us-modal-stage-runbook.md

@@ -263,6 +263,13 @@ one stage of a run at a time, because two concurrent stages would race on the
 same state directory. The attempt ledger enforces this, best effort (see
 "Preemption and restarts").
 
+For `us-acs-local-release`, run package at the commit calibrate ran at.
+Calibrate records the code that wrote the calibrated H5 in
+`consumer_export.json`, and package refuses to name a release after any
+other commit (microcosm#1263). A commit landing between the two means
+re-running calibrate, qa and finalize at the new commit, or packaging at the
+old one.
+
 For a gated or private Hub input, set `MICROCOSM_MODAL_HF_SECRET` to the name
 of a Modal secret that holds `HF_TOKEN`, for example `huggingface-token` in
 the `policyengine` workspace. Otherwise no secret is attached. Only the

packages/microcosm-build/src/microcosm/build/us_runtime/retired_stored_inputs.py

@@ -37,7 +37,6 @@
 
 from collections.abc import Collection
 from dataclasses import dataclass
-from numbers import Real
 from typing import Any
 
 import numpy as np
@@ -179,23 +178,23 @@ def _live_wic(
         "stored_decisions_kept": int(present.sum()),
         "stored_decisions_true": int(decisions[present].sum()),
         "people_assumed_true": int(assumed.sum()),
+        "people_allowed_without_decision": int(allowed.sum()),
         "rule": ACS_WIC_ASSUMPTION,
         "status": "interim",
         "flip_condition": ACS_WIC_FLIP_CONDITION,
     }
 
 
 def _booleans(values: pd.Series) -> np.ndarray:
-    """Stored decisions as Booleans: bools, or real numbers equal to 0 or 1."""
+    """Stored decisions as Booleans: bools, or int or float numbers equal to 0 or 1."""
 
     out = np.empty(len(values), dtype=bool)
     for position, value in enumerate(values.to_numpy(dtype=object)):
         if isinstance(value, (bool, np.bool_)):
             out[position] = bool(value)
-        elif (
-            isinstance(value, Real)
-            and not isinstance(value, (bool, np.bool_))
-            and value in (0, 1)
+        elif isinstance(value, (int, float, np.integer, np.floating)) and value in (
+            0,
+            1,
         ):
             out[position] = bool(value)
         else:

packages/microcosm-build/tests/engine_free/us/test_us_acs_local_release_tool.py

@@ -118,6 +118,12 @@ def _spm_measurement_gate_passes(monkeypatch):
     )
 
 
+def _writer_code(module) -> dict:
+    """This checkout's code identity, as the calibrate stage records it."""
+
+    return module._repo_code_identity(allow_dirty=True)
+
+
 def _load_tool_module():
     root = _TEST_PATHS.repository
     path = root / "tools" / "build_us_acs_local_release.py"
@@ -332,6 +338,7 @@ def limitation(consumer_export):
                 "from": "would_claim_wic",
                 "to": "takes_up_wic_if_eligible",
                 "people_assumed_true": 12,
+                "people_allowed_without_decision": 20,
                 "flip_condition": "Drop it once microcosm#1156 lands.",
             }
         }
@@ -340,6 +347,10 @@ def limitation(consumer_export):
     assert entry["status"] == "reviewed_interim_assumption"
     assert entry["column"] == "takes_up_wic_if_eligible"
     assert entry["people_assumed_true"] == 12
+    # The text claims only what the rule did: the people without a decision.
+    assert (
+        "every person without one (12, all on the ACS spine, of 20" in entry["reason"]
+    )
     assert entry["treatment"] == "Drop it once microcosm#1156 lands."
     assert entry["calibration_blocker"] is False
     assert limitation(None) is None
@@ -400,6 +411,82 @@ def test_only_acs_spine_people_may_lack_a_wic_decision() -> None:
         )
 
 
+@pytest.mark.parametrize(
+    "spine",
+    [np.nan, None, b"acs_2024_1yr", "ACS_2024_1YR"],
+    ids=["nan", "none", "bytes", "other-label"],
+)
+def test_an_unreadable_spine_tag_refuses_rather_than_fills(spine) -> None:
+    """A person whose tag is not exactly the ACS spine may not lack one."""
+
+    module = _load_tool_module()
+    frame = _spine_frame(["acs_2024_1yr", spine], [None, None])
+
+    with pytest.raises(SystemExit, match="did not allow"):
+        module._with_live_stored_inputs(frame)
+
+
+def test_a_person_table_without_the_spine_tag_is_refused() -> None:
+    from microcosm.frame import Frame
+
+    module = _load_tool_module()
+    frame = _spine_frame(["acs_2024_1yr"], [None])
+    tables = {entity: frame.table(entity) for entity in frame.entities}
+    tables["person"] = tables["person"].drop(columns="person_spine")
+    untagged = Frame(
+        tables,
+        frame.schema,
+        {entity: frame.weights_for(entity) for entity in frame.weighted_entities},
+    )
+
+    with pytest.raises(SystemExit, match="person_spine"):
+        module._with_live_stored_inputs(untagged)
+
+
+@pytest.mark.parametrize(
+    ("written_by", "allow_dirty", "match"),
+    [
+        (None, True, "records no code identity"),
+        ({"sha": "abc1234"}, True, "records no code identity"),
+        ({"commit": ""}, True, "records no code identity"),
+        ({"commit": "0" * 40, "dirty": False}, True, "written by code 0000000"),
+        ({"commit": "a" * 40, "dirty": True}, False, "dirty tree"),
+    ],
+    ids=["absent", "short-sha-only", "empty", "other-commit", "dirty-writer"],
+)
+def test_package_names_the_release_only_after_the_code_that_wrote_it(
+    written_by, allow_dirty, match
+) -> None:
+    """The release id names the packaging code, so that code must have
+    written the calibrated bytes (review of microcosm#1309)."""
+
+    module = _load_tool_module()
+    code = {"sha": "aaaaaaaaa", "commit": "a" * 40, "dirty": False, "branch": "main"}
+    export = {} if written_by is None else {"code": written_by}
+
+    with pytest.raises(SystemExit, match=match):
+        module._require_artifact_written_by(export, code, allow_dirty=allow_dirty)
+
+    # The abbreviated sha may differ in length between checkouts; only the
+    # full commit is compared.
+    module._require_artifact_written_by(
+        {"code": {"sha": "aaaaaaa", "commit": "a" * 40, "dirty": False}},
+        code,
+        allow_dirty=False,
+    )
+    module._require_artifact_written_by(
+        {"code": {"commit": "a" * 40, "dirty": True}}, code, allow_dirty=True
+    )
+
+
+def test_the_recorded_code_identity_carries_the_full_commit() -> None:
+    module = _load_tool_module()
+    code = module._repo_code_identity(allow_dirty=True)
+
+    assert len(code["commit"]) == 40
+    assert code["commit"].startswith(code["sha"])
+
+
 def test_parse_args_enforces_stage_requirements(tmp_path: Path) -> None:
     module = _load_tool_module()
     base = [
@@ -588,7 +675,10 @@ def _package_evidence_args(module, tmp_path: Path, monkeypatch, *, hours_report)
             "artifact_sha256": artifact_sha,
             "per_spine": {},
         },
-        "consumer_export.json": {"staging_sha256": module._sha256(staging)},
+        "consumer_export.json": {
+            "staging_sha256": module._sha256(staging),
+            "code": _writer_code(module),
+        },
         "held_back_columns.json": {"total": 0},
         "reviewed_null_fills.json": {"columns_filled": []},
         "materialize_rss.json": {
@@ -1401,7 +1491,10 @@ def test_finalize_report_round_trips_into_package(tmp_path, monkeypatch) -> None
             "artifact_sha256": artifact_sha,
             "per_spine": {},
         },
-        "consumer_export.json": {"staging_sha256": artifact_sha},
+        "consumer_export.json": {
+            "staging_sha256": artifact_sha,
+            "code": _writer_code(module),
+        },
         "consumer_reviewed_null_fills.json": {"columns_filled": []},
     }
     for name, value in evidence.items():
@@ -1472,6 +1565,7 @@ def test_a_sparse_era_calibration_is_bound_through_finalize_and_package(
         "run_identity_sha256": stamp,
         "out_h5_sha256": artifact_sha,
         "weights_sha256": "w",
+        "code": _writer_code(module),
     }
     qa = {
         "plain_consumption": True,
@@ -1559,7 +1653,10 @@ def _package_args_with_hours(module, tmp_path, monkeypatch, *, gate_state):
             "artifact_sha256": artifact_sha,
             "per_spine": {},
         },
-        "consumer_export.json": {"staging_sha256": artifact_sha},
+        "consumer_export.json": {
+            "staging_sha256": artifact_sha,
+            "code": _writer_code(module),
+        },
         "consumer_reviewed_null_fills.json": {"columns_filled": []},
     }
     for name, value in evidence.items():
@@ -1599,6 +1696,44 @@ def test_package_accepts_passing_hours_gate_bound_to_the_packaged_bytes(
     assert gate["artifact_sha256"] == result["root_artifact"]["sha256"] == copied_sha
 
 
+def _set_writer(args, **changes) -> None:
+    path = args.checkpoint_dir / "consumer_export.json"
+    export = json.loads(path.read_text())
+    export["code"] = {**export["code"], **changes}
+    path.write_text(json.dumps(export))
+
+
+@requires_pytables
+def test_package_refuses_another_writer_before_any_release_dir(tmp_path, monkeypatch):
+    """The writer binding runs before the release directory exists."""
+
+    module = _load_tool_module()
+    args = _package_args_with_hours(module, tmp_path, monkeypatch, gate_state="passed")
+    _set_writer(args, commit="0" * 40)
+
+    with pytest.raises(SystemExit, match="written by code 0000000"):
+        module.do_package(args)
+    assert not (args.out / "releases").exists()
+
+
+@requires_pytables
+def test_a_dirty_writer_marks_the_release_dirty(tmp_path, monkeypatch):
+    """A clean packaging tree cannot hide a dirty writer (review of #1311)."""
+
+    module = _load_tool_module()
+    # A clean packaging tree, whatever this checkout's state, so only the
+    # writer's flag can make the release dirty.
+    clean = {**module._repo_code_identity(allow_dirty=True), "dirty": False}
+    monkeypatch.setattr(module, "_repo_code_identity", lambda allow_dirty: dict(clean))
+    args = _package_args_with_hours(module, tmp_path, monkeypatch, gate_state="passed")
+    _set_writer(args, dirty=True)
+
+    release_dir = Path(module.do_package(args)["release_dir"])
+
+    build = json.loads((release_dir / "build_manifest.json").read_text())
+    assert build["build_dirty"] is True
+
+
 @requires_pytables
 def test_package_records_the_stored_input_gate_bound_to_the_packaged_bytes(
     tmp_path, monkeypatch

packages/microcosm-build/tests/engine_free/us/test_us_retired_stored_inputs.py

@@ -1,8 +1,10 @@
 """Moving the donor's retired stored inputs to the live ones (microcosm#1263).
 
 Every frame here is invented. The engine is a variable-name set, so no
-policyengine-us is imported; ``test_us_retired_stored_inputs_engine`` checks
-the names against the installed engine.
+policyengine-us is imported. The installed engine's side of the premise (the
+live input is a Person Boolean defaulting to True, and neither retired name is
+defined) is pinned by
+``packages/microcosm-data/tests/engine_contract/us/test_stored_inputs.py``.
 """
 
 from __future__ import annotations
@@ -32,13 +34,11 @@
 
 
 def _remedy(frame, engine=ENGINE, allowed=None):
-    """The rule with the release tool's mask: the ACS people may lack one."""
+    """The rule with the release tool's mask: the ACS-spine people may lack one."""
 
     if allowed is None:
-        channels = frame.table("person")["person_support_channel"].map(
-            lambda value: value.decode() if isinstance(value, bytes) else value
-        )
-        allowed = channels.eq(ACS_CHANNEL).to_numpy(dtype=bool)
+        spine = frame.table("person")["person_spine"]
+        allowed = spine.eq(ACS_CHANNEL).fillna(False).to_numpy(dtype=bool)
     return with_live_stored_inputs(
         frame, engine_variables=engine, may_lack_decision=allowed
     )
@@ -50,6 +50,9 @@ def _frame(channels, decisions=None, part_b=None) -> Frame:
     frame = acs_frame([[(20, 40.0)] for _ in channels])
     person = frame.table("person").copy()
     person["person_support_channel"] = channels
+    person["person_spine"] = [
+        ACS_CHANNEL if channel == ACS_CHANNEL else "asec_puf" for channel in channels
+    ]
     if decisions is not None:
         person[RETIRED_WIC_INPUT] = pd.Series(decisions, dtype=object)
     if part_b is not None:
@@ -80,6 +83,7 @@ def test_donor_decisions_are_kept_and_acs_gaps_take_wic_up():
     assert wic["stored_decisions_kept"] == 2
     assert wic["stored_decisions_true"] == 1
     assert wic["people_assumed_true"] == 2
+    assert wic["people_allowed_without_decision"] == 2
     assert wic["status"] == "interim"
     assert "microcosm#1156" in wic["flip_condition"]
 
@@ -166,6 +170,58 @@ def test_it_refuses_rather_than_guessing(channels, decisions, match):
         _remedy(_frame(channels, decisions))
 
 
+@pytest.mark.parametrize(
+    ("stored", "expected"),
+    [
+        (pd.array([True, False, None], dtype="boolean"), [True, False, True]),
+        (np.array([True, False, True]), [True, False, True]),
+        (np.array([1.0, 0.0, np.nan]), [True, False, True]),
+        (pd.Series([True, False, None], dtype=object), [True, False, True]),
+        (pd.array([1, 0, None], dtype="Int64"), [True, False, True]),
+        (pd.array([1.0, 0.0, None], dtype="Float64"), [True, False, True]),
+        (np.array([1, 0, 1], dtype=np.int8), [True, False, True]),
+        (np.array([1, 0, 1], dtype=np.uint8), [True, False, True]),
+    ],
+    ids=[
+        "pandas-boolean-NA",
+        "numpy-bool",
+        "float-0-1-nan",
+        "object",
+        "nullable-Int64-NA",
+        "nullable-Float64-NA",
+        "int8",
+        "uint8",
+    ],
+)
+def test_every_stored_boolean_dtype_reads_alike(stored, expected):
+    frame = _frame(["asec", "puf_tax_detail", ACS_CHANNEL])
+    tables = {entity: frame.table(entity) for entity in frame.entities}
+    tables["person"] = tables["person"].assign(**{RETIRED_WIC_INPUT: stored})
+    typed = Frame(
+        tables,
+        frame.schema,
+        {entity: frame.weights_for(entity) for entity in frame.weighted_entities},
+    )
+
+    person = _remedy(typed).frame.table("person")
+
+    assert person[LIVE_WIC_INPUT].dtype == bool
+    assert person[LIVE_WIC_INPUT].tolist() == expected
+
+
+@pytest.mark.parametrize("value", ["fraction", "decimal", "complex"])
+def test_only_bools_and_real_zero_or_one_are_decisions(value):
+    """Bools and int/float 0/1 only. policyengine.py's reader refuses a
+    Fraction and a Decimal too; this rule is stricter about complex."""
+
+    from decimal import Decimal
+    from fractions import Fraction
+
+    stored = {"fraction": Fraction(1), "decimal": Decimal(1), "complex": 1 + 0j}
+    with pytest.raises(ValueError, match="not a Boolean"):
+        _remedy(_frame(["asec"], [stored[value]]))
+
+
 def test_zero_one_numbers_are_read_as_decisions():
     frame = _frame(["asec", "asec", "asec"], [1, 0.0, np.float64(1.0)])
 

packages/microcosm-build/tests/engine_free/us/test_us_spine_blindness.py

@@ -307,6 +307,11 @@
         "release_gate_preflight.py",
         "release_input_coverage.py",
         "release_target_parity.py",
+        # Consumer-artifact remedy for two retired stored inputs (microcosm#1263):
+        # renames the WIC draw (keeping stored decisions and defaulting missing
+        # ones to True) and drops a dead column. It reads no provenance column;
+        # the release tool passes the people who may lack a decision.
+        "retired_stored_inputs.py",
         "scf_auto_loans.py",
         "scf_wealth.py",
         "sipp_financial_assets.py",
@@ -342,10 +347,6 @@
         "spm_universe_source.py",
         # Reviewed held-out SSI/Social Security diagnostic; no population treatment.
         "ssi_social_security_holdout.py",
-        # Consumer-artifact remedy for two retired stored inputs (microcosm#1263):
-        # renames the WIC draw and drops a dead column. It reads no provenance
-        # column; the release tool passes the people who may lack a decision.
-        "retired_stored_inputs.py",
         "stacked_spine.py",  # Provenance owner (#578 revision); see owners list.
         "support_provenance.py",
         "take_up.py",