Add per-vintage reported-coverage signal gate (#720)

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.

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

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

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

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