diff --git a/changelog.d/720-reported-coverage-vintage-gate.added.md b/changelog.d/720-reported-coverage-vintage-gate.added.md new file mode 100644 index 000000000..35a0fdb4e --- /dev/null +++ b/changelog.d/720-reported-coverage-vintage-gate.added.md @@ -0,0 +1 @@ +Add `us_reported_coverage_vintage_signal_gate`, a US release gate requiring every pooled source vintage (`person_support_channel` × `source_year`) to carry a fully populated, boolean, non-zero-signal column for each of the nine reported-coverage person inputs; it closes the partial-vintage hole behind the certified Build P artifact's `has_medicaid_health_coverage_at_interview` thinning (24.6M vs ~58M survey under 65, #720), which presence-style checks cannot see. Wired into `tools/build_us_fiscal_refresh_release.py` as a base-frame hard gate with a persisted failure receipt, re-run on the export frame, and a `reported_coverage_vintage_signal` manifest entry. diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py index e55690cb5..e9e756473 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/__init__.py @@ -162,6 +162,8 @@ CPS_REPORTED_TANF_AMOUNT_RAW_COLUMN, CPS_REPORTED_TANF_TYPE_RAW_COLUMN, CPS_REPORTED_WIC_RAW_COLUMN, + US_REPORTED_COVERAGE_PERSON_INPUTS, + US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, WIC_CARRIER_ADJUDICATION_URL, derive_us_cps_carried_inputs, reported_tanf_enrollment_by_spm_unit, @@ -964,6 +966,9 @@ with_us_ssi_take_up, write_us_ssi_take_up_diagnostics, ) +from microcosm.build.us_runtime.support_provenance import ( + us_reported_coverage_vintage_signal_gate, +) from microcosm.build.us_runtime.take_up import ( US_TAKE_UP_SHARE_BAND, SeededTakeUpResult, @@ -1095,6 +1100,8 @@ "CPS_REPORTED_TANF_AMOUNT_RAW_COLUMN", "CPS_REPORTED_TANF_TYPE_RAW_COLUMN", "CPS_REPORTED_WIC_RAW_COLUMN", + "US_REPORTED_COVERAGE_PERSON_INPUTS", + "US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS", "WIC_CARRIER_ADJUDICATION_URL", "SimpleTaxExpenditureReform", "ReformValidationSpec", @@ -1852,6 +1859,7 @@ "derive_us_cps_carried_inputs", "reported_tanf_enrollment_by_spm_unit", "reported_wic_receipt_carrier", + "us_reported_coverage_vintage_signal_gate", "disaggregate_us_puf_aggregate_records_from_manifest", "finalize_us_puf_tax_detail_predictions", "hard_target_package_aliases", diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py b/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py index 08d761260..0c93ad543 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/cps_carried.py @@ -31,6 +31,8 @@ "CPS_REPORTED_TANF_AMOUNT_RAW_COLUMN", "CPS_REPORTED_TANF_TYPE_RAW_COLUMN", "CPS_REPORTED_WIC_RAW_COLUMN", + "US_REPORTED_COVERAGE_PERSON_INPUTS", + "US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS", "WIC_CARRIER_ADJUDICATION_URL", "derive_us_cps_carried_inputs", "reported_snap_receipt_by_spm_unit", @@ -50,6 +52,26 @@ "https://github.com/PolicyEngine/microcosm/issues/591#issuecomment-5160668979" ) +# The nine reported-coverage person inputs _fill_health_coverage_inputs derives +# from the ASEC NOW_* at-interview recodes (microcosm #720). +US_REPORTED_COVERAGE_PERSON_INPUTS: tuple[str, ...] = ( + "has_champva_health_coverage_at_interview", + "has_esi", + "has_indian_health_service_coverage_at_interview", + "has_marketplace_health_coverage_at_interview", + "has_medicaid_health_coverage_at_interview", + "has_non_marketplace_direct_purchase_health_coverage_at_interview", + "has_other_means_tested_health_coverage_at_interview", + "has_tricare_health_coverage_at_interview", + "has_va_health_coverage_at_interview", +) + +# Below this row count a vintage is a smoke pool and zero reporters for a rare +# flag is sampling noise; at or above it, a zero is a missing-source symptom +# (the rarest flag, CHAMPVA at ~0.3% of persons, is expected ~15 times in +# 5,000 rows). +US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS = 5_000 + CPS_CARRIED_FORMULA_OWNED_COLUMNS = frozenset( { "capital_gains", diff --git a/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py b/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py index e0f8d40fc..da5bf9e08 100644 --- a/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py +++ b/packages/microcosm-build/src/microcosm/build/us_runtime/support_provenance.py @@ -8,6 +8,8 @@ import numpy as np import pandas as pd +from microcosm.build.gates import GateResult + __all__ = [ "BASE_ASEC_SUPPORT_CHANNEL", "PERSON_SUPPORT_CHANNEL_COLUMN", @@ -24,6 +26,7 @@ "support_clone_index_column", "support_role_series", "support_source_id_column", + "us_reported_coverage_vintage_signal_gate", "validate_assembly_provenance", "without_support_role_metadata", ] @@ -511,3 +514,144 @@ def thaw(item: Any) -> Any: return item return {str(key): thaw(item) for key, item in value.items()} + + +def us_reported_coverage_vintage_signal_gate( + frame: _ProvenanceFrame, + *, + min_vintage_rows: int | None = None, +) -> GateResult: + """Require every pooled source vintage to carry reported-coverage signal. + + Microcosm #720: the pooled income-year 2022/2023 ASEC inputs carried only + ``NOW_GRP``/``NOW_MRK`` of the 18 ``NOW_*`` at-interview recodes, so + :func:`derive_us_cps_carried_inputs` silently mapped every 2022/2023 + -vintage person to ``False`` for seven of the nine reported-coverage + flags and the certified Build P artifact reported 24.6M under-65 + Medicaid at interview against ~58M survey. A flag populated for one + vintage passes the presence-style checks (``release_input_coverage``, + ``degenerate_input_signal``); this gate enforces the per-vintage + invariant those checks cannot see. + + Groups are ``person_support_channel`` x ``source_year`` when the channel + column is present (ACS-spine rows also carry ``source_year``, so a + year-only key would let ACS signal mask a missing ASEC recode), else + ``source_year`` alone. Every group with at least ``min_vintage_rows`` + person rows must have, for every reported-coverage input, a boolean-like + column with no nulls and at least one reporter. Provenance must be + present: a missing ``source_year`` column, null source years, or an + empty person table fail the gate rather than collapsing into one group. + + Groups below ``min_vintage_rows`` (smoke pools) are recorded in the + details but not enforced. This is a zero-signal sentinel, not a survey + mass check: it observes that a vintage has no reporters; the #720 + cause (a source input lacking the recode) is the documented reading. + """ + + # Lazy import: cps_carried (the derivation owner) imports alimony, which + # imports this module; the gate lives here because it is origin-aware by + # charter (provenance owners only may read the support channel). + from microcosm.build.us_runtime.cps_carried import ( + US_REPORTED_COVERAGE_PERSON_INPUTS, + US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, + ) + + if min_vintage_rows is None: + min_vintage_rows = US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS + person = frame.table("person") + missing = [ + column + for column in US_REPORTED_COVERAGE_PERSON_INPUTS + if column not in person.columns + ] + if "source_year" not in person.columns: + missing.append("source_year") + if missing: + return GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=tuple(f"person column missing: {column}." for column in missing), + details={"missing": missing}, + ) + if len(person) == 0: + # Unreachable through a valid Frame (weights cannot be empty); kept + # so a direct caller cannot pass an empty table as "no failures". + return GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=("person table is empty: no vintage carries any signal.",), + details={"rows": 0}, + ) + failures: list[str] = [] + null_years = int(person["source_year"].isna().sum()) + if null_years: + failures.append( + f"source_year: {null_years} person rows have no source year; the " + "per-vintage invariant cannot be proven for unprovenanced rows." + ) + keys: list[pd.Series] = [] + if PERSON_SUPPORT_CHANNEL_COLUMN in person.columns: + keys.append(person[PERSON_SUPPORT_CHANNEL_COLUMN].astype(str)) + keys.append(person["source_year"]) + vintages: dict[str, dict[str, object]] = {} + for key, group in person.groupby(keys, sort=True, dropna=True): + parts = key if isinstance(key, tuple) else (key,) + label = "/".join(str(part) for part in parts) + rows = int(len(group)) + reporter_counts: dict[str, int] = {} + null_counts: dict[str, int] = {} + dtype_failures: list[str] = [] + for column in US_REPORTED_COVERAGE_PERSON_INPUTS: + values = group[column] + if not ( + pd.api.types.is_bool_dtype(values) + or pd.api.types.is_numeric_dtype(values) + ): + dtype_failures.append(column) + reporter_counts[column] = 0 + null_counts[column] = int(values.isna().sum()) + continue + null_counts[column] = int(values.isna().sum()) + reporter_counts[column] = int(values.fillna(False).astype(bool).sum()) + enforced = rows >= min_vintage_rows + vintages[label] = { + "rows": rows, + "enforced": enforced, + "reporter_counts": reporter_counts, + "null_counts": null_counts, + } + if not enforced: + continue + for column in dtype_failures: + failures.append( + f"{column}: vintage {label} stores a non-boolean dtype " + f"({group[column].dtype}); reported-coverage inputs must be " + "boolean." + ) + for column, count in null_counts.items(): + if count and column not in dtype_failures: + failures.append( + f"{column}: vintage {label} has {count} null values over " + f"{rows} person rows; the flag must be fully populated." + ) + for column, count in reporter_counts.items(): + if count == 0 and column not in dtype_failures: + failures.append( + f"{column}: vintage {label} has 0 reporters over {rows} " + "person rows (consistent with a source input lacking the " + "at-interview recode, microcosm #720)." + ) + return GateResult( + name="reported_coverage_vintage_signal", + passed=not failures, + failures=tuple(failures), + details={ + "min_vintage_rows": int(min_vintage_rows), + "grouping": ( + [PERSON_SUPPORT_CHANNEL_COLUMN, "source_year"] + if PERSON_SUPPORT_CHANNEL_COLUMN in person.columns + else ["source_year"] + ), + "vintages": vintages, + }, + ) diff --git a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py index 1c4c4a623..ebb7d3999 100644 --- a/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py +++ b/packages/microcosm-build/tests/test_us_fiscal_refresh_builder.py @@ -2237,6 +2237,63 @@ def test_release_gate_failures_include_health_input_signal() -> None: ] +def test_release_gate_failures_include_reported_coverage_vintage_signal() -> None: + builder = _load_builder_module() + result = SimpleNamespace( + skipped=(), + diagnostics=_passing_critical_diagnostics(builder), + initial_loss=10.0, + final_loss=5.0, + ) + reported_coverage_vintage_gate = builder.GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=( + "has_medicaid_health_coverage_at_interview: source_year 2022 has 0 " + "reporters over 54464 person rows — the vintage source lacks the " + "at-interview recode (microcosm #720).", + ), + ) + + assert builder._release_gate_failures( + result, + {"dropped_target_names": []}, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, + ) == [ + "Reported-coverage vintage signal failed: " + "has_medicaid_health_coverage_at_interview: source_year 2022 has 0 " + "reporters over 54464 person rows — the vintage source lacks the " + "at-interview recode (microcosm #720).", + ] + + +def test_reported_coverage_vintage_gate_receipt_is_written(tmp_path) -> None: + builder = _load_builder_module() + gate = builder.GateResult( + name="reported_coverage_vintage_signal", + passed=False, + failures=( + "has_medicaid_health_coverage_at_interview: vintage asec/2022 has 0 " + "reporters over 54464 person rows (consistent with a source input " + "lacking the at-interview recode, microcosm #720).", + ), + details={"min_vintage_rows": 5000, "vintages": {"asec/2022": {"rows": 54464}}}, + ) + release_dir = tmp_path / "releases" / "rid" + + path = builder._write_reported_coverage_vintage_gate_receipt(release_dir, gate) + + assert path == release_dir / builder.US_REPORTED_COVERAGE_VINTAGE_GATE_RECEIPT + payload = json.loads(path.read_text()) + assert payload["gate"] == "reported_coverage_vintage_signal" + assert payload["passed"] is False + assert payload["failures"] == list(gate.failures) + assert payload["details"]["vintages"]["asec/2022"]["rows"] == 54464 + # A receipt-only directory is not a certified release (#568): reruns under + # the same id stay allowed. + builder._refuse_certified_release_dir_reuse(release_dir) + + def test_base_population_scale_gate_rejects_underweighted_base(small_frame) -> None: builder = _load_builder_module() @@ -4584,6 +4641,15 @@ def fake_retirement_distributions_signal_gate(frame): details={"checked": True}, ), ) + monkeypatch.setattr( + builder, + "us_reported_coverage_vintage_signal_gate", + lambda frame: builder.GateResult( + name="reported_coverage_vintage_signal", + passed=True, + details={"checked": True}, + ), + ) monkeypatch.setattr( builder, "us_wic_claim_signal_gate", diff --git a/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py b/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py new file mode 100644 index 000000000..670f4c008 --- /dev/null +++ b/packages/microcosm-build/tests/test_us_reported_coverage_vintage_gate.py @@ -0,0 +1,303 @@ +"""Per-vintage reported-coverage signal gate tests (microcosm #720).""" + +from __future__ import annotations + +import numpy as np +import pandas as pd + +from microcosm.build.us_runtime import ( + CPS_CARRIED_PERSON_INPUTS, + US_REPORTED_COVERAGE_PERSON_INPUTS, + US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS, + us_reported_coverage_vintage_signal_gate, +) +from microcosm.build.us_runtime.cps_carried import _fill_health_coverage_inputs +from microcosm.frame import US_SCHEMA, Frame, WeightKind, Weights + +_ALL_FALSE = {column: False for column in US_REPORTED_COVERAGE_PERSON_INPUTS} + + +def _person_table(rows: list[dict]) -> pd.DataFrame: + records = [] + for index, row in enumerate(rows): + record = {"A_AGE": 40} + record.update(row) + record.setdefault("person_id", index + 1) + record.setdefault("person_household_id", index + 1) + records.append(record) + return pd.DataFrame(records) + + +def _us_frame(person_rows: list[dict]) -> Frame: + return _us_frame_from_person(_person_table(person_rows)) + + +def _us_frame_from_person(person: pd.DataFrame) -> Frame: + person = person.copy() + n = len(person) + household_ids = person["person_household_id"].to_numpy() + unique_households = np.unique(household_ids) + person["person_tax_unit_id"] = person["person_household_id"] + 1_000 + person["person_spm_unit_id"] = person["person_household_id"] + 2_000 + person["person_family_id"] = person["person_household_id"] + 3_000 + person["person_marital_unit_id"] = np.arange(n, dtype="int64") + 4_000 + tables = { + "person": person, + "household": pd.DataFrame({"household_id": unique_households}), + "tax_unit": pd.DataFrame({"tax_unit_id": unique_households + 1_000}), + "spm_unit": pd.DataFrame({"spm_unit_id": unique_households + 2_000}), + "family": pd.DataFrame({"family_id": unique_households + 3_000}), + "marital_unit": pd.DataFrame( + {"marital_unit_id": np.arange(n, dtype="int64") + 4_000} + ), + } + return Frame( + tables, + US_SCHEMA, + { + "household": Weights( + values=np.ones(len(unique_households), dtype=np.float64), + kind=WeightKind.DESIGN, + ) + }, + ) + + +def _vintage_rows( + source_year: int, + *, + reporters: tuple[str, ...] = US_REPORTED_COVERAGE_PERSON_INPUTS, + rows: int = 3, +) -> list[dict]: + """``rows`` persons of one vintage; the first row reports every column in + ``reporters``, the rest report nothing.""" + + out: list[dict] = [] + for index in range(rows): + record = {"source_year": source_year, **_ALL_FALSE} + if index == 0: + record.update({column: True for column in reporters}) + out.append(record) + return out + + +class TestContract: + def test_the_nine_inputs_are_cps_carried_flags(self) -> None: + assert len(US_REPORTED_COVERAGE_PERSON_INPUTS) == 9 + assert set(US_REPORTED_COVERAGE_PERSON_INPUTS) <= CPS_CARRIED_PERSON_INPUTS + assert all( + column == "has_esi" or column.endswith("_coverage_at_interview") + for column in US_REPORTED_COVERAGE_PERSON_INPUTS + ) + + def test_the_nine_inputs_are_exactly_the_derived_health_coverage_flags( + self, + ) -> None: + person = pd.DataFrame({"NOW_GRP": [1, 2]}) + _fill_health_coverage_inputs(person) + derived = {column for column in person.columns if column.startswith("has_")} + assert derived == set(US_REPORTED_COVERAGE_PERSON_INPUTS) + + def test_default_threshold_is_a_release_pool_scale(self) -> None: + assert US_REPORTED_COVERAGE_VINTAGE_GATE_MIN_ROWS == 5_000 + + +class TestGate: + def test_every_vintage_with_signal_passes(self) -> None: + frame = _us_frame(_vintage_rows(2023) + _vintage_rows(2024)) + gate = us_reported_coverage_vintage_signal_gate(frame, min_vintage_rows=3) + assert gate.passed, gate.failures + assert gate.name == "reported_coverage_vintage_signal" + vintages = gate.details["vintages"] + assert set(vintages) == {"2023", "2024"} + for summary in vintages.values(): + assert summary["rows"] == 3 + assert summary["enforced"] is True + assert ( + summary["reporter_counts"]["has_medicaid_health_coverage_at_interview"] + == 1 + ) + + def test_vintage_without_medicaid_reporters_fails(self) -> None: + without_medicaid = tuple( + column + for column in US_REPORTED_COVERAGE_PERSON_INPUTS + if column != "has_medicaid_health_coverage_at_interview" + ) + frame = _us_frame( + _vintage_rows(2022, reporters=without_medicaid) + _vintage_rows(2024) + ) + gate = us_reported_coverage_vintage_signal_gate(frame, min_vintage_rows=3) + assert not gate.passed + assert len(gate.failures) == 1 + assert "has_medicaid_health_coverage_at_interview" in gate.failures[0] + assert "vintage 2022 has 0 reporters" in gate.failures[0] + assert "#720" in gate.failures[0] + + def test_issue_720_shape_fails_seven_flags_for_two_vintages(self) -> None: + # Build P: ASEC 2022/2023 inputs carried only NOW_GRP and NOW_MRK, so + # every other at-interview flag was False for those vintages. + carried = ("has_esi", "has_marketplace_health_coverage_at_interview") + frame = _us_frame( + _vintage_rows(2022, reporters=carried) + + _vintage_rows(2023, reporters=carried) + + _vintage_rows(2024) + ) + gate = us_reported_coverage_vintage_signal_gate(frame, min_vintage_rows=3) + assert not gate.passed + assert len(gate.failures) == 14 + failed_vintages = { + failure.split("vintage ")[1].split(" ")[0] for failure in gate.failures + } + assert failed_vintages == {"2022", "2023"} + assert not any("has_esi" in failure for failure in gate.failures) + assert ( + gate.details["vintages"]["2024"]["reporter_counts"][ + "has_medicaid_health_coverage_at_interview" + ] + == 1 + ) + assert ( + gate.details["vintages"]["2022"]["reporter_counts"][ + "has_medicaid_health_coverage_at_interview" + ] + == 0 + ) + + def test_small_vintage_is_recorded_but_not_enforced(self) -> None: + frame = _us_frame( + _vintage_rows(2022, reporters=(), rows=2) + _vintage_rows(2024) + ) + gate = us_reported_coverage_vintage_signal_gate(frame, min_vintage_rows=3) + assert gate.passed, gate.failures + assert gate.details["vintages"]["2022"]["enforced"] is False + assert gate.details["vintages"]["2024"]["enforced"] is True + assert gate.details["min_vintage_rows"] == 3 + + def test_default_threshold_skips_unit_sized_frames(self) -> None: + frame = _us_frame(_vintage_rows(2022, reporters=())) + gate = us_reported_coverage_vintage_signal_gate(frame) + assert gate.passed, gate.failures + assert gate.details["vintages"]["2022"]["enforced"] is False + + def test_missing_source_year_fails_closed(self) -> None: + rows = [{**row} for row in _vintage_rows(2024)] + for row in rows: + del row["source_year"] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame(rows), min_vintage_rows=3 + ) + assert not gate.passed + assert gate.failures == ("person column missing: source_year.",) + + def test_null_source_year_rows_fail(self) -> None: + rows = _vintage_rows(2024) + [{**_vintage_rows(2024)[0], "source_year": None}] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame(rows), min_vintage_rows=3 + ) + assert not gate.passed + assert any("1 person rows have no source year" in f for f in gate.failures) + + def test_support_channel_separates_acs_from_asec(self) -> None: + # An ACS-spine vintage with signal must not mask an ASEC vintage + # whose source lacked the recode: both carry source_year 2024. + asec = [ + {**row, "person_support_channel": "asec"} + for row in _vintage_rows( + 2024, + reporters=("has_esi", "has_marketplace_health_coverage_at_interview"), + ) + ] + acs = [{**row, "person_support_channel": "acs"} for row in _vintage_rows(2024)] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame(asec + acs), min_vintage_rows=3 + ) + assert not gate.passed + assert gate.details["grouping"] == ["person_support_channel", "source_year"] + assert set(gate.details["vintages"]) == {"acs/2024", "asec/2024"} + assert len(gate.failures) == 7 + assert all("vintage asec/2024" in failure for failure in gate.failures) + + def test_null_flag_values_fail_completeness(self) -> None: + person = _person_table(_vintage_rows(2024)) + person["has_medicaid_health_coverage_at_interview"] = pd.array( + [True, None, False], dtype="boolean" + ) + gate = us_reported_coverage_vintage_signal_gate( + _us_frame_from_person(person), min_vintage_rows=3 + ) + assert not gate.passed + assert len(gate.failures) == 1 + assert "1 null values" in gate.failures[0] + assert ( + gate.details["vintages"]["2024"]["null_counts"][ + "has_medicaid_health_coverage_at_interview" + ] + == 1 + ) + + def test_non_boolean_dtype_fails(self) -> None: + person = _person_table(_vintage_rows(2024)) + person["has_esi"] = ["True", "False", "False"] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame_from_person(person), min_vintage_rows=3 + ) + assert not gate.passed + assert len(gate.failures) == 1 + assert "non-boolean dtype" in gate.failures[0] + + def test_missing_column_fails(self) -> None: + rows = [{**row} for row in _vintage_rows(2024)] + for row in rows: + del row["has_esi"] + gate = us_reported_coverage_vintage_signal_gate( + _us_frame(rows), min_vintage_rows=3 + ) + assert not gate.passed + assert gate.failures == ("person column missing: has_esi.",) + assert gate.details == {"missing": ["has_esi"]} + + +class TestMechanism: + def test_nan_source_recode_becomes_false_and_fails_the_vintage(self) -> None: + """The #720 chain: a vintage whose source lacks ``NOW_MCAID`` derives + ``False`` for every person, and only this gate sees it.""" + + person = _person_table( + [ + {"source_year": 2023, "NOW_GRP": 1, "NOW_MRK": 2, "NOW_MCAID": np.nan}, + {"source_year": 2023, "NOW_GRP": 2, "NOW_MRK": 1, "NOW_MCAID": np.nan}, + {"source_year": 2023, "NOW_GRP": 1, "NOW_MRK": 2, "NOW_MCAID": np.nan}, + {"source_year": 2024, "NOW_GRP": 1, "NOW_MRK": 2, "NOW_MCAID": 1}, + {"source_year": 2024, "NOW_GRP": 2, "NOW_MRK": 1, "NOW_MCAID": 2}, + {"source_year": 2024, "NOW_GRP": 1, "NOW_MRK": 2, "NOW_MCAID": 1}, + ] + ) + for column in ( + "NOW_NONM", + "NOW_CHAMPVA", + "NOW_MIL", + "NOW_VACARE", + "NOW_OTHMT", + "NOW_IHSFLG", + ): + person[column] = [np.nan] * 3 + [1, 2, 1] + _fill_health_coverage_inputs(person) + + medicaid = person["has_medicaid_health_coverage_at_interview"] + assert not medicaid[person["source_year"] == 2023].any() + assert medicaid[person["source_year"] == 2024].sum() == 2 + assert person["has_esi"].sum() == 4 # NOW_GRP carried for both vintages + + gate = us_reported_coverage_vintage_signal_gate( + _us_frame_from_person(person), min_vintage_rows=3 + ) + assert not gate.passed + failed = {failure.split(":")[0] for failure in gate.failures} + assert failed == set(US_REPORTED_COVERAGE_PERSON_INPUTS) - { + "has_esi", + "has_marketplace_health_coverage_at_interview", + } + assert all( + "vintage 2023 has 0 reporters" in failure for failure in gate.failures + ) diff --git a/tools/build_us_fiscal_refresh_release.py b/tools/build_us_fiscal_refresh_release.py index 1a1f73cba..048f6a535 100644 --- a/tools/build_us_fiscal_refresh_release.py +++ b/tools/build_us_fiscal_refresh_release.py @@ -147,6 +147,7 @@ us_relationship_inputs_signal_gate, us_release_input_coverage_gate, us_release_target_parity_gate, + us_reported_coverage_vintage_signal_gate, us_retirement_contributions_signal_gate, us_retirement_distributions_signal_gate, us_salt_refund_income_signal_gate, @@ -5020,6 +5021,41 @@ def _health_input_signal_gate(frame: Frame) -> GateResult: ) +US_REPORTED_COVERAGE_VINTAGE_GATE_RECEIPT = ( + "reported_coverage_vintage_gate_failure.json" +) + + +def _write_reported_coverage_vintage_gate_receipt( + release_dir: Path, gate: GateResult +) -> Path: + """Persist a RED reported-coverage vintage gate before the early raise. + + The base-frame gate raises before the collector, diagnostics, and + manifests run, so without this receipt a failure would leave only + optional staging telemetry and an exception (microcosm #720 review). + The receipt never marks a directory certified: certification is keyed + on ``release_manifest.json`` (#568), so a rerun under the same id is + not blocked by it. + """ + + release_dir.mkdir(parents=True, exist_ok=True) + path = release_dir / US_REPORTED_COVERAGE_VINTAGE_GATE_RECEIPT + path.write_text( + json.dumps( + { + "gate": gate.name, + "passed": gate.passed, + "failures": list(gate.failures), + "details": dict(gate.details), + }, + indent=1, + allow_nan=False, + ) + ) + return path + + def _engine_input_variables() -> tuple[str, ...]: """Persistable PolicyEngine input variables (formula-owned excluded).""" return tuple(PolicyEngineUSEngine().variables()) @@ -6395,6 +6431,7 @@ def _release_gate_failures( snap_take_up_gate: GateResult | None = None, eligibility_inputs_gate: GateResult | None = None, pregnancy_gate: GateResult | None = None, + reported_coverage_vintage_gate: GateResult | None = None, snap_discretionary_exemption_gate: GateResult | None = None, target_registry: TargetRegistry | None = None, ) -> list[str]: @@ -6438,6 +6475,14 @@ def _release_gate_failures( failures.extend( f"Pregnancy signal failed: {failure}" for failure in pregnancy_gate.failures ) + if ( + reported_coverage_vintage_gate is not None + and not reported_coverage_vintage_gate.passed + ): + failures.extend( + f"Reported-coverage vintage signal failed: {failure}" + for failure in reported_coverage_vintage_gate.failures + ) if ( snap_discretionary_exemption_gate is not None and not snap_discretionary_exemption_gate.passed @@ -6942,6 +6987,7 @@ def _write_release_calibration_diagnostics( snap_take_up_gate: GateResult | None = None, eligibility_inputs_gate: GateResult | None = None, pregnancy_gate: GateResult | None = None, + reported_coverage_vintage_gate: GateResult | None = None, snap_discretionary_exemption_gate: GateResult | None = None, gate_failures: Iterable[str], timing: Mapping[str, object] | None = None, @@ -7070,6 +7116,15 @@ def _write_release_calibration_diagnostics( if pregnancy_gate is not None else None ), + "reported_coverage_vintage_signal": ( + { + "passed": reported_coverage_vintage_gate.passed, + "failures": list(reported_coverage_vintage_gate.failures), + "details": dict(reported_coverage_vintage_gate.details), + } + if reported_coverage_vintage_gate is not None + else None + ), "snap_discretionary_exemption_signal": ( { "passed": snap_discretionary_exemption_gate.passed, @@ -7336,6 +7391,7 @@ def _build_manifests( snap_take_up_gate: GateResult | None = None, eligibility_inputs_gate: GateResult | None = None, pregnancy_gate: GateResult | None = None, + reported_coverage_vintage_gate: GateResult | None = None, snap_discretionary_exemption_gate: GateResult | None = None, timing: Mapping[str, object] | None = None, warm_start_calibration: Mapping[str, object] | None = None, @@ -7374,6 +7430,7 @@ def _build_manifests( snap_take_up_gate=snap_take_up_gate, eligibility_inputs_gate=eligibility_inputs_gate, pregnancy_gate=pregnancy_gate, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, target_registry=registry, ) @@ -7582,6 +7639,17 @@ def _build_manifests( if pregnancy_gate is not None else {} ), + **( + { + "reported_coverage_vintage_signal": { + "passed": reported_coverage_vintage_gate.passed, + "failures": list(reported_coverage_vintage_gate.failures), + "details": dict(reported_coverage_vintage_gate.details), + } + } + if reported_coverage_vintage_gate is not None + else {} + ), **( { "snap_discretionary_exemption_signal": { @@ -7713,6 +7781,16 @@ def _build_manifests( if pregnancy_gate is not None else {} ), + **( + { + "reported_coverage_vintage_signal": { + "passed": reported_coverage_vintage_gate.passed, + "details": dict(reported_coverage_vintage_gate.details), + } + } + if reported_coverage_vintage_gate is not None + else {} + ), **( { "snap_discretionary_exemption_signal": { @@ -9302,6 +9380,29 @@ def main(argv: Sequence[str] | None = None) -> None: for failure in pregnancy_gate.failures ) ) + reported_coverage_vintage_gate = us_reported_coverage_vintage_signal_gate( + base_frame + ) + if not reported_coverage_vintage_gate.passed: + receipt_path = _write_reported_coverage_vintage_gate_receipt( + release_dir, reported_coverage_vintage_gate + ) + if telemetry is not None: + telemetry.stage( + "reported_coverage_vintage_gate", + status="failed", + message="Reported-coverage vintage signal gate failed.", + failures=list(reported_coverage_vintage_gate.failures), + force_upload=True, + ) + raise RuntimeError( + "Release gates failed: " + + "; ".join( + f"Reported-coverage vintage signal failed: {failure}" + for failure in reported_coverage_vintage_gate.failures + ) + + f" (receipt: {receipt_path})" + ) if telemetry is not None: telemetry.stage( "wic_claim_input", @@ -10494,6 +10595,18 @@ def main(argv: Sequence[str] | None = None) -> None: "Health-input signal evaluation crashed in degraded mode; " f"recorded instead of masking earlier failures: {error}" ) + try: + reported_coverage_vintage_gate = us_reported_coverage_vintage_signal_gate( + export_frame + ) + except Exception as error: + if not early_terminal_gate_failures: + raise + reported_coverage_vintage_gate = None + early_terminal_gate_failures.append( + "Reported-coverage vintage signal evaluation crashed in degraded " + f"mode; recorded instead of masking earlier failures: {error}" + ) try: other_health_insurance_gate = us_other_health_insurance_signal_gate( export_frame @@ -10668,6 +10781,7 @@ def main(argv: Sequence[str] | None = None) -> None: snap_take_up_gate=snap_take_up_gate, eligibility_inputs_gate=eligibility_inputs_gate, pregnancy_gate=pregnancy_gate, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, target_registry=registry, ) @@ -10714,6 +10828,7 @@ def main(argv: Sequence[str] | None = None) -> None: snap_take_up_gate=snap_take_up_gate, eligibility_inputs_gate=eligibility_inputs_gate, pregnancy_gate=pregnancy_gate, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, support_value_repairs={ "social_security_components": social_security_component_repair, @@ -11301,6 +11416,7 @@ def main(argv: Sequence[str] | None = None) -> None: snap_take_up_gate=snap_take_up_gate, eligibility_inputs_gate=eligibility_inputs_gate, pregnancy_gate=pregnancy_gate, + reported_coverage_vintage_gate=reported_coverage_vintage_gate, snap_discretionary_exemption_gate=snap_discretionary_exemption_gate, timing=timing, warm_start_calibration=warm_start_calibration,