Fix SNAP state take-up crash: Frame.broadcast only targets person

packages/populace-build/tests/test_us_fiscal_refresh_builder.py

@@ -6873,3 +6873,30 @@ def test_main_runs_cross_register_and_take_up_contract_preflights() -> None:
         "assert_take_up_treatments_consistent",
     ):
         assert preflight in called, f"main() no longer calls {preflight}"
+
+
+def test_spm_unit_state_fips_collapses_person_broadcast_per_unit() -> None:
+    """_spm_unit_state_fips must not ask Frame.broadcast for a group target.
+
+    Frame.broadcast only supports the person entity; the SNAP state take-up
+    stage previously crashed on every release run by broadcasting to
+    spm_unit. The helper must collapse the person-level broadcast to one
+    zero-padded FIPS text code per SPM unit, in spm_unit table order.
+    """
+    builder = _load_builder_module()
+    sys.path.insert(0, str(Path(__file__).parent))
+    try:
+        from test_us_snap_state_take_up import _us_frame
+    finally:
+        sys.path.pop(0)
+
+    frame = _us_frame(
+        [
+            {"person_id": 1, "person_spm_unit_id": 10, "state_fips": 6},
+            {"person_id": 2, "person_spm_unit_id": 10, "state_fips": 6},
+            {"person_id": 3, "person_spm_unit_id": 20, "state_fips": 36},
+            {"person_id": 4, "person_spm_unit_id": 30, "state_fips": 2},
+        ]
+    )
+    out = builder._spm_unit_state_fips(frame)
+    assert list(out) == ["06", "36", "02"]

tools/build_us_fiscal_refresh_release.py

@@ -2732,10 +2732,28 @@ def _snap_state_target_table(target_specs: tuple) -> pd.DataFrame:
 
 
 def _spm_unit_state_fips(frame: Frame) -> np.ndarray:
-    """SPM-unit-aligned state FIPS text codes via the frame's linkage."""
-    return np.asarray(
-        _state_fips_text(frame.broadcast("state_fips", to="spm_unit").to_numpy())
+    """SPM-unit-aligned state FIPS text codes via the person linkage.
+
+    ``Frame.broadcast`` only targets the person entity, so state_fips is
+    broadcast to persons and collapsed to one value per SPM unit (members of
+    an SPM unit share a household, hence a state).
+    """
+    person = frame.table("person")
+    person_state = frame.broadcast("state_fips", to="person").to_numpy()
+    first_by_unit = (
+        pd.Series(person_state, index=person["person_spm_unit_id"].to_numpy())
+        .groupby(level=0)
+        .first()
     )
+    spm_unit_ids = frame.table("spm_unit")["spm_unit_id"].to_numpy()
+    aligned = first_by_unit.reindex(spm_unit_ids)
+    if aligned.isna().any():
+        missing = int(aligned.isna().sum())
+        raise RuntimeError(
+            f"SNAP state take-up: {missing} spm_unit(s) have no member "
+            "persons to carry state_fips."
+        )
+    return np.asarray(_state_fips_text(aligned.to_numpy()))
 
 
 def _snap_spm_unit_eligibility(