Gate persisted input mass and carry the #278 bases through the sparse pipeline

packages/populace-build/src/populace/build/__init__.py

@@ -58,6 +58,7 @@ def _assert_frame_compatible(version: str, required: tuple[int, int]) -> None:
     export_surface_gate,
     exported_nonzero_gate,
     formula_owned_export_gate,
+    input_mass_parity_gate,
     macro_realism_gate,
     nonconstant_columns_gate,
     nonnegative_columns_gate,
@@ -127,6 +128,7 @@ def _assert_frame_compatible(version: str, required: tuple[int, int]) -> None:
     "export_surface_gate",
     "exported_nonzero_gate",
     "formula_owned_export_gate",
+    "input_mass_parity_gate",
     "macro_realism_gate",
     "nonconstant_columns_gate",
     "nonnegative_columns_gate",

packages/populace-build/src/populace/build/gates.py

@@ -25,6 +25,11 @@
 - :func:`export_surface_gate` and :func:`target_surface_gate` — replacement
   builds can prove they cover a reference artifact's export variables and
   calibration targets, e.g. UK Populace against eFRS.
+- :func:`input_mass_parity_gate` — persisted input columns must carry their
+  weighted mass through a derived artifact. A sparse selection or a rebuilt
+  base pipeline that silently drops an input base (IRA contributions,
+  childcare expenses) scores every reform on that base at ~$0 while hitting
+  its own target surface perfectly (populace issue #278).
 
 Scoring uses :func:`relative_error_loss` — the calibrator's own objective —
 so there is no calibrator-vs-scorer objective mismatch: what the solver
@@ -48,6 +53,7 @@
     "export_surface_gate",
     "formula_owned_export_gate",
     "exported_nonzero_gate",
+    "input_mass_parity_gate",
     "macro_realism_gate",
     "nonconstant_columns_gate",
     "nonnegative_columns_gate",
@@ -798,6 +804,137 @@ def export_surface_gate(
     )
 
 
+def input_mass_parity_gate(
+    candidate_totals: Mapping[str, float],
+    reference_totals: Mapping[str, float],
+    *,
+    candidate_name: str = "candidate",
+    reference_name: str = "reference",
+    relative_tolerance: float = 0.5,
+    minimum_reference_total: float = 0.0,
+    reviewed_exclusions: Mapping[str, str] | None = None,
+) -> GateResult:
+    """Require persisted input mass to survive into a derived artifact.
+
+    A sparse selection, a saved-weight reconstruction, or a rebuilt base
+    pipeline can zero an untargeted input column while every calibrated
+    target still fits: the on-surface residuals look perfect and every
+    reform touching that base silently scores ~$0 (populace issue #278,
+    where the sparse release dropped IRA/HSA/pension-contribution and
+    childcare inputs the dense parent carried). This gate compares
+    weighted per-column totals between the artifact and its parent or a
+    certified reference release; columns that lose more than
+    ``relative_tolerance`` of their reference mass — or disappear
+    entirely — fail unless a reviewed exclusion documents why.
+
+    Columns whose reference mass is at most ``minimum_reference_total``
+    in absolute value are skipped: a near-zero reference total makes
+    relative drift meaningless. Columns only the candidate carries are
+    reported in details but never fail — added signal is the export
+    surface gate's concern, not lost mass.
+
+    Args:
+        candidate_totals: Column -> weighted total in the derived artifact.
+        reference_totals: Column -> weighted total in the parent build or
+            certified reference release.
+        candidate_name: Label for the derived artifact in messages.
+        reference_name: Label for the reference artifact in messages.
+        relative_tolerance: Maximum allowed ``|candidate - reference| /
+            |reference|`` before a column fails.
+        minimum_reference_total: Reference-mass floor below which a column
+            is not checked.
+        reviewed_exclusions: Column -> reason for columns allowed to drift
+            or disappear (each needs a non-empty reason; unused entries are
+            reported so the register cannot rot).
+
+    Returns:
+        Pass iff every material reference column keeps its mass within
+        tolerance or carries a reviewed exclusion.
+    """
+    if relative_tolerance < 0:
+        raise ValueError(
+            f"relative_tolerance must be non-negative, got {relative_tolerance!r}."
+        )
+    if minimum_reference_total < 0:
+        raise ValueError(
+            "minimum_reference_total must be non-negative, got "
+            f"{minimum_reference_total!r}."
+        )
+    exclusions = _reviewed_exclusion_reasons(reviewed_exclusions)
+    reference = {str(name): float(total) for name, total in reference_totals.items()}
+    candidate = {str(name): float(total) for name, total in candidate_totals.items()}
+
+    failures: list[str] = []
+    drifts: dict[str, float] = {}
+    checked: list[str] = []
+    skipped_below_floor: list[str] = []
+    for name in sorted(reference):
+        reference_total = reference[name]
+        if not np.isfinite(reference_total):
+            raise ValueError(f"Reference total for {name!r} must be finite.")
+        if abs(reference_total) <= minimum_reference_total:
+            skipped_below_floor.append(name)
+            continue
+        if name in exclusions:
+            continue
+        checked.append(name)
+        if name not in candidate:
+            drifts[name] = -1.0
+            failures.append(
+                f"{name}: {reference_name} carries {reference_total:.6g} but the "
+                f"column is absent from {candidate_name}; carry the input "
+                "through the build or add a reviewed exclusion."
+            )
+            continue
+        candidate_total = candidate[name]
+        if not np.isfinite(candidate_total):
+            raise ValueError(f"Candidate total for {name!r} must be finite.")
+        drift = (candidate_total - reference_total) / abs(reference_total)
+        drifts[name] = float(drift)
+        if candidate_total == 0.0:
+            # Total loss is the issue #278 signature: a zeroed input base is
+            # the same failure as an absent one, at any drift tolerance.
+            failures.append(
+                f"{name}: {reference_name} carries {reference_total:.6g} but "
+                f"{candidate_name} mass is zero ({drift:+.1%}); carry the "
+                "input through the build or add a reviewed exclusion."
+            )
+        elif abs(drift) > relative_tolerance:
+            failures.append(
+                f"{name}: {candidate_name} mass {candidate_total:.6g} vs "
+                f"{reference_name} {reference_total:.6g} ({drift:+.1%}, beyond "
+                f"±{relative_tolerance:.0%}); carry the input through the build "
+                "or add a reviewed exclusion."
+            )
+
+    worst = sorted(
+        ((name, drift) for name, drift in drifts.items()),
+        key=lambda item: abs(item[1]),
+        reverse=True,
+    )
+    return GateResult(
+        name="input_mass_parity",
+        passed=not failures,
+        failures=tuple(failures),
+        details={
+            "candidate_name": candidate_name,
+            "reference_name": reference_name,
+            "relative_tolerance": float(relative_tolerance),
+            "minimum_reference_total": float(minimum_reference_total),
+            "columns_checked": len(checked),
+            "columns_below_reference_floor": len(skipped_below_floor),
+            "candidate_only_columns": sorted(set(candidate) - set(reference)),
+            "worst_drifts": {name: drift for name, drift in worst[:20]},
+            "reviewed_exclusions": {
+                name: reason
+                for name, reason in sorted(exclusions.items())
+                if name in reference
+            },
+            "unused_reviewed_exclusions": sorted(set(exclusions) - set(reference)),
+        },
+    )
+
+
 def target_surface_gate(
     candidate_targets: Iterable[str],
     reference_targets: Iterable[str],

packages/populace-build/src/populace/build/us/source_stages.json

@@ -81,6 +81,8 @@
         "real_estate_taxes",
         "home_mortgage_interest",
         "student_loan_interest",
+        "traditional_ira_contributions_desired",
+        "self_employed_pension_contributions_desired",
         "rental_income",
         "estate_income",
         "farm_income",
@@ -93,7 +95,8 @@
         "first_home_mortgage_interest",
         "second_home_mortgage_interest",
         "first_home_mortgage_origination_year",
-        "second_home_mortgage_origination_year"
+        "second_home_mortgage_origination_year",
+        "health_savings_account_ald"
       ],
       "nonnegative_outputs": [
         "employment_income_before_lsr",
@@ -114,6 +117,9 @@
         "real_estate_taxes",
         "home_mortgage_interest",
         "student_loan_interest",
+        "traditional_ira_contributions_desired",
+        "self_employed_pension_contributions_desired",
+        "health_savings_account_ald",
         "first_home_mortgage_balance",
         "second_home_mortgage_balance",
         "first_home_mortgage_interest",

packages/populace-build/src/populace/build/us_runtime/__init__.py

@@ -61,6 +61,7 @@
 from populace.build.us_runtime.cps_carried import (
     CPS_CARRIED_FORMULA_OWNED_COLUMNS,
     CPS_CARRIED_PERSON_INPUTS,
+    CPS_CARRIED_SPM_UNIT_INPUTS,
     derive_us_cps_carried_inputs,
 )
 from populace.build.us_runtime.demographics import (
@@ -106,6 +107,9 @@
     us_immigration_stage_spec,
     with_us_immigration_inputs,
 )
+from populace.build.us_runtime.input_mass import (
+    us_input_mass_totals,
+)
 from populace.build.us_runtime.puf_support import (
     BASE_ASEC_SUPPORT_CHANNEL,
     PUF_TAX_DETAIL_DEFAULT_PERSON_OUTPUTS,
@@ -150,6 +154,7 @@
     "BASE_ASEC_SUPPORT_CHANNEL",
     "CPS_CARRIED_FORMULA_OWNED_COLUMNS",
     "CPS_CARRIED_PERSON_INPUTS",
+    "CPS_CARRIED_SPM_UNIT_INPUTS",
     "SimpleTaxExpenditureReform",
     "ReformValidationSpec",
     "REFORM_VALIDATION_SCHEMA_VERSION",
@@ -227,6 +232,7 @@
     "pool_asec_sources",
     "reform_validation_payload",
     "source_gap_family_ids",
+    "us_input_mass_totals",
     "us_plan",
     "us_source_operation_handlers",
     "write_reform_validation",

packages/populace-build/src/populace/build/us_runtime/cps_carried.py

@@ -18,6 +18,7 @@
 __all__ = [
     "CPS_CARRIED_FORMULA_OWNED_COLUMNS",
     "CPS_CARRIED_PERSON_INPUTS",
+    "CPS_CARRIED_SPM_UNIT_INPUTS",
     "derive_us_cps_carried_inputs",
 ]
 
@@ -79,6 +80,12 @@
     }
 )
 
+CPS_CARRIED_SPM_UNIT_INPUTS = frozenset(
+    {
+        "spm_unit_pre_subsidy_childcare_expenses",
+    }
+)
+
 
 def derive_us_cps_carried_inputs(frame: Frame) -> Frame:
     """Carry raw CPS ASEC values onto PE input leaves.
@@ -161,6 +168,7 @@ def derive_us_cps_carried_inputs(frame: Frame) -> Frame:
         _fill_missing(person, output, _source(person, source))
 
     _fill_health_coverage_inputs(person)
+    _fill_spm_unit_childcare_inputs(person, tables["spm_unit"])
 
     formula_owned = sorted(CPS_CARRIED_FORMULA_OWNED_COLUMNS.intersection(person))
     if formula_owned:
@@ -269,6 +277,35 @@ def _fill_health_coverage_inputs(person: pd.DataFrame) -> None:
         _fill_bool_missing(person, output, _yes_code(person, source))
 
 
+def _fill_spm_unit_childcare_inputs(
+    person: pd.DataFrame,
+    spm_unit: pd.DataFrame,
+) -> None:
+    """Carry ASEC SPM childcare expenses onto the SPM-unit input leaf.
+
+    ``SPM_CHILDCAREXPNS`` is an SPM-unit value replicated on every member's
+    person record, so the unit's value is its members' maximum. The engine
+    derives person, tax-unit, and CDCC childcare expenses from this leaf;
+    a base without it zeroes every CDCC baseline (populace issue #278).
+    """
+
+    column = "spm_unit_pre_subsidy_childcare_expenses"
+    if column in spm_unit.columns:
+        return
+    member_values = pd.DataFrame(
+        {
+            "person_spm_unit_id": person["person_spm_unit_id"],
+            column: _source(person, "SPM_CHILDCAREXPNS"),
+        }
+    )
+    unit_values = member_values.groupby("person_spm_unit_id", sort=False)[column].max()
+    spm_unit[column] = (
+        unit_values.reindex(spm_unit["spm_unit_id"])
+        .fillna(0.0)
+        .to_numpy(dtype=np.float64)
+    )
+
+
 def _ira_distributions(person: pd.DataFrame) -> np.ndarray:
     values = np.zeros(len(person), dtype=np.float64)
     for suffix in ("1", "2", "1_YNG", "2_YNG"):

packages/populace-build/src/populace/build/us_runtime/input_mass.py

@@ -0,0 +1,74 @@
+"""Weighted per-column input mass for US frames.
+
+These totals feed :func:`populace.build.gates.input_mass_parity_gate`:
+comparing a derived artifact's persisted-input totals against its dense
+parent (or a certified reference release) catches input bases that a sparse
+selection or a rebuilt base pipeline silently zeroes — the failure mode of
+populace issue #278, where the sparse default release carried ~$0 in
+IRA-contribution, HSA, pension-contribution, and childcare inputs while
+hitting its own calibration target surface.
+"""
+
+from __future__ import annotations
+
+from collections.abc import Iterable
+
+import numpy as np
+import pandas as pd
+
+from populace.frame import Frame
+
+__all__ = ["us_input_mass_totals"]
+
+
+def us_input_mass_totals(
+    frame: Frame,
+    *,
+    columns: Iterable[str] | None = None,
+) -> dict[str, float]:
+    """Weighted totals of the frame's numeric and boolean value columns.
+
+    Every non-structural numeric column is summed under the owning entity's
+    effective weights (household weights broadcast through membership for
+    entities without their own vector); boolean columns total their weighted
+    ``True`` mass. String/enum columns and structural columns (entity ids and
+    person membership ids) are skipped.
+
+    Args:
+        frame: A US-schema frame.
+        columns: Optional restriction — when given, only these columns are
+            totalled. Pass the engine's input-variable list on raw build
+            frames so source-survey scratch columns that never persist do not
+            enter the comparison.
+
+    Returns:
+        Column name -> weighted total. The mapping is flat because the frame
+        already enforces globally unique column names across entity tables.
+    """
+
+    schema = frame.schema
+    structural = {schema.person_id_column}
+    for group in schema.group_entities:
+        structural.add(schema.id_column(group))
+        structural.add(schema.membership_column(group))
+    requested = None if columns is None else {str(name) for name in columns}
+
+    totals: dict[str, float] = {}
+    for entity in frame.entities:
+        table = frame.table(entity)
+        weights = np.asarray(frame.resolve_weights(entity).values, dtype=np.float64)
+        for column in table.columns:
+            if column in structural:
+                continue
+            if requested is not None and column not in requested:
+                continue
+            values = table[column]
+            if pd.api.types.is_bool_dtype(values):
+                numeric = values.fillna(False).to_numpy(dtype=np.float64)
+            elif pd.api.types.is_numeric_dtype(values):
+                numeric = pd.to_numeric(values, errors="coerce")
+                numeric = numeric.fillna(0.0).to_numpy(dtype=np.float64)
+            else:
+                continue
+            totals[column] = float(numeric @ weights)
+    return totals