Generate complete ACS WIC participation and reject missing release inputs

changelog.d/1155.fixed.md

@@ -0,0 +1 @@
+Generate complete WIC participation decisions for ACS local-area records after demographic completion and before donor assembly, preserving existing donor decisions and rejecting missing participation at release boundaries.

docs/agent-guide.md

@@ -251,6 +251,14 @@ register entry with its reason, bound to its producer in
 name. A new US lane that writes a release H5 must run the check where it
 records `build.built_with_model_package`.
 
+The supported ACS local-area path additionally requires complete boolean
+`takes_up_wic_if_eligible`: ACS decisions are generated after demographic
+completion and before donor assembly; donor decisions are not regenerated.
+Calculation/export cannot default-fill missing WIC decisions, and packaging
+checks actual person values as well as stored names. The code repair does not
+replace the published dataset; see [the WIC replacement instructions](us-acs-local-wic-replacement.md)
+and #1154 for the separate rebuild, publication, certification, and `.py` removal.
+
 A US release or release-gate preflight that receives a multispine pool through
 `--base-h5` must authenticate its sibling terminal manifest. A current stacked
 pool whose terminal battery is red remains fail-closed unless the operator

docs/evidence/spec-engine/us-f0-coverage.json

@@ -1656,13 +1656,13 @@
           "compiler_ir.node_slices"
         ],
         "expected": {
-          "map_sha256": "b4d6ec9c11ce323d83319a3d0289e0cf7216a7936d6c6f08cef0a463ad369641",
-          "protocol_sha256": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560"
+          "map_sha256": "02ec8173113ac6d7256181439831e1c51ad8fca5bbcda657f1630bfc6b5a1c28",
+          "protocol_sha256": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6"
         },
         "failures": [],
         "observed": {
-          "map_sha256": "b4d6ec9c11ce323d83319a3d0289e0cf7216a7936d6c6f08cef0a463ad369641",
-          "protocol_sha256": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560"
+          "map_sha256": "02ec8173113ac6d7256181439831e1c51ad8fca5bbcda657f1630bfc6b5a1c28",
+          "protocol_sha256": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6"
         },
         "status": "covered"
       },
@@ -1677,7 +1677,7 @@
           "compiler_ir.seed_stream_map"
         ],
         "expected": {
-          "implementation_sha256": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560",
+          "implementation_sha256": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6",
           "protocol": "legacy-v1",
           "streams": [
             "build_model",
@@ -1698,7 +1698,7 @@
         },
         "failures": [],
         "observed": {
-          "implementation_sha256": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560",
+          "implementation_sha256": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6",
           "protocol": "legacy-v1",
           "streams": [
             "build_model",
@@ -2599,7 +2599,7 @@
       "country": "us",
       "schema_id": "country_spec",
       "schema_version": 1,
-      "spec_sha256": "f76f921d4f74392b106b8c0729ce250494e0a55455eaba7b910b19cdf7825178"
+      "spec_sha256": "5048cabc59e5d7d9789e751cc9a5c5b4d97ab6fb3f15429cabce3b568615073b"
     }
   },
   "report_schema_version": 3,
@@ -2609,7 +2609,7 @@
     "country": "us",
     "schema_id": "country_spec",
     "schema_version": 1,
-    "spec_sha256": "f76f921d4f74392b106b8c0729ce250494e0a55455eaba7b910b19cdf7825178"
+    "spec_sha256": "5048cabc59e5d7d9789e751cc9a5c5b4d97ab6fb3f15429cabce3b568615073b"
   },
   "status": "pass"
 }

docs/us-acs-local-wic-replacement.md

@@ -0,0 +1,108 @@
+# Replace incomplete ACS local-area WIC participation
+
+## Scope and ownership
+
+[Issue #1154](https://github.com/PolicyEngine/microcosm/issues/1154) remains open
+until a dataset owner rebuilds and publishes the replacement, `.py` certifies
+it, and the temporary assumption is removed. [Issue #1155](https://github.com/PolicyEngine/microcosm/issues/1155)
+covers this code repair only. Merging the code does not change existing H5
+files or certify a population.
+
+The temporary `.py` change is [PR #562](https://github.com/PolicyEngine/policyengine.py/pull/562).
+Its separate removal draft, [PR #563](https://github.com/PolicyEngine/policyengine.py/pull/563),
+must remain blocked until the actual replacement is published and certified.
+It contains a deliberately failing check against the known incomplete pin.
+
+This work does not publish a dataset, modify the national default/latest
+pointer, deploy services, or add environment variables or database changes.
+
+## Generation order
+
+The supported ACS local-area build calls `build_optional_acs_multispine`:
+
+1. Require complete boolean `takes_up_wic_if_eligible` on the donor population.
+   This is the participation decision conditional on eligibility, not the
+   separate `receives_wic` reported-receipt input. Do not recreate donor draws.
+2. Load ACS source records and map their native age, sex, income, and family
+   membership. Keep `source_year`, `source_household_id`, and `source_person_id`.
+3. Complete the demographic inputs through the existing transfer plan,
+   including pregnancy and own children. Missing required inputs remain errors.
+4. On ACS records only, run `with_acs_wic_claim_input`, which requires complete
+   source identity and delegates to `with_us_wic_claim_input`. Use the existing
+   build seed and ACS source year. The existing generator owns category order,
+   sourced FNS CY2022 rates, and deterministic source-person hash draws. It
+   assigns pregnant, postpartum, infant, child, or no-category probabilities;
+   it does not invent a breastfeeding observation or change eligibility rules.
+5. Combine ACS records with donors, preserving donor decisions exactly. Require
+   complete boolean participation over the combined population. Record the
+   generator seed, source year, output column, row count, and rate-source URL
+   in the existing orchestration provenance.
+
+The country engine still determines income and demographic eligibility and
+benefit amounts. The generator does not assume every eligible ACS person
+claims WIC. Reordering records cannot change a person's seeded decision.
+
+## Release validation
+
+The local release tool rejects missing or malformed participation before
+sampling/materialization, before consumer export, and before packaging.
+`fill_reviewed_nulls` checks WIC before loading the reviewed-null register:
+even an explicit register entry cannot fill participation from the engine's
+`True` default. Other reviewed-null handling remains unchanged.
+
+Packaging retains the existing stored-input-name checks, including rejection
+of obsolete `would_claim_wic`. It additionally checks the actual participation
+column in every person row and records the checked row count in the existing
+stored-input report, bound to the packaged H5 hash. H5 reads request batches
+of 65,536 rows. Table-format files project only the participation column;
+fixed-format files require all columns and pandas may unpickle an entire
+object block despite the requested row bounds. This is not a claim of a strict
+memory bound for historical fixed-format files.
+
+## Dataset-owner steps (not performed by this PR)
+
+1. Choose a qualified donor release with complete current-name participation
+   and every other required donor input. Preserve its immutable identity.
+   Existing donor qualification, receipt, hours, geography, and model checks
+   still apply; an incompatible donor must fail, not receive an implicit fill.
+2. Use a new output directory and rebuild ACS staging with the supported
+   `tools/build_us_acs_multispine_base.py` entry point. It is deprecated for
+   new general multispine builds but still supports this local-area release
+   path. Use pinned ACS archives, the PUMA mapping, donor inputs, and explicit
+   build configuration. A small smoke build may validate the setup but cannot
+   be published as the full replacement.
+3. Check `orchestration.provenance.wic_claim` in the new staging summary and
+   complete boolean participation in the H5. Compare donor decisions to the
+   selected parent; inspect ACS rates by demographic category. Never edit the
+   existing published artifact in place.
+4. Rebuild materialization, calibration, consumer export, QA, finalization, and
+   packaging using `tools/build_us_acs_local_release.py --stage all` and the
+   recorded full-scale configuration. Do not reuse old target matrices,
+   calibrated weights, consumer evidence, prepared years, or QA reports:
+   changing participation can change WIC and dependent outputs. Satisfy every
+   existing release requirement, not just the new participation check.
+5. Review and publish the new immutable **non-default local-area** release with
+   the existing `microcosm-publish-release <release-dir> --no-latest` route.
+   Do not change the national latest pointer or default dataset.
+6. Supply the actual revision, artifact hash, and regional manifest to `.py`
+   PR #563. Use its existing certification process and model-compatibility
+   requirements; do not invent replacement metadata or bypass validation.
+7. After the compatibility parent merges, rebase and retarget the removal PR
+   to `main`, certify the replacement, run its checks, and only then merge it.
+   Keep ordinary legacy-name mapping for other datasets. Close #1154 only
+   after publication, certification, and removal are all complete.
+
+## Focused tests
+
+Existing CI discovers the new tests through the test-directory registry:
+
+- `engine_free/us/test_us_acs_wic.py`: deterministic generator delegation,
+  reordered identities, required source inputs, strict participation validation.
+- `engine_free/us/test_us_acs_multispine.py`: stage order, donor preservation,
+  early donor validation, and unchanged disabled behavior.
+- `engine_free/us/test_us_acs_local_release_tool.py`: no WIC default fills,
+  export refusal, packaging refusal, and hash-bound row-count evidence.
+- `engine_workflow/us/test_us_acs_wic.py`: H5 round-trip, live country-engine
+  consumption, fixed/table validation, and reads over multiple batches.
+
+These small synthetic tests do not rebuild, publish, or certify the real dataset.

packages/microcosm-build/src/microcosm/build/spec_engine/inventory_coverage.py

@@ -359,8 +359,8 @@
     "late_schedule": "e59c019d3d454eac99ac0ac209b6c5b6faaf9bdfcaeee18c36a25be19bf7da2f",
     "ownership": "5f64f0aac49e2313177564f71876bffc8c81b3ded4df701e70930e60e9c98356",
     "primary_tuples": "fdf23da429f8c501e198a2f5b719763d523a41565c2153fdf17b0344f21bcd7c",
-    "seed_map": "b4d6ec9c11ce323d83319a3d0289e0cf7216a7936d6c6f08cef0a463ad369641",
-    "seed_protocol": "114217a2918329b0601d94deb8ed83f73919584b85ca8aabe45a40665afb6560",
+    "seed_map": "02ec8173113ac6d7256181439831e1c51ad8fca5bbcda657f1630bfc6b5a1c28",
+    "seed_protocol": "e3035d45cd8e5ce29fc3291894a28596897dfd58955fcf0139eaafb7409a0bc6",
     "source_manifest": "f4f8bfeb79268a4043a843b47cec1d1c0e58e9f5dd6bc23417ada95446dd7b8e",
     "take_up": "9522ce40f7dea569312dd7a7beb474e5a5afadd2f1e10cb5876534c3ef623d35",
     "tail": "ac92829c88a1a4fb6460d61190918d5d99c6c377fc8dd8f62f02b332d09bf59c",

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

@@ -52,6 +52,12 @@
     us_puma_ladder_assignment_summary,
 )
 from microcosm.build.us_runtime.support_provenance import BASE_ASEC_SUPPORT_CHANNEL
+from microcosm.build.us_runtime.wic_claim import (
+    US_WIC_CLAIM_OUTPUT_COLUMNS,
+    WIC_CLAIM_FNS_SOURCE_URL,
+    require_complete_us_wic_claim_input,
+    with_acs_wic_claim_input,
+)
 from microcosm.frame import Frame
 
 __all__ = ["AcsMultispineResult", "build_optional_acs_multispine"]
@@ -115,6 +121,9 @@ def build_optional_acs_multispine(
             provenance={"enabled": False},
         )
 
+    # Donors must already carry their qualified decisions. Never regenerate
+    # them with ACS categories or silently use the country model's True default.
+    _require_complete_pre_acs_donor_wic_input(base)
     raw_acs, loader_metadata = build_acs_pums_unit_frame(
         source,
         chunksize=chunksize,
@@ -206,6 +215,19 @@ def build_optional_acs_multispine(
     transferred_frame = transferred.frame
     del transferred
 
+    # Pregnancy/parent inputs are now complete. Generate decisions on ACS only,
+    # while its stable Census identities are intact, before donor assembly.
+    transferred_frame = with_acs_wic_claim_input(
+        transferred_frame, seed=seed, time_period=source.vintage
+    )
+    wic_provenance: dict[str, object] = {
+        "seed": seed,
+        "source_year": source.vintage,
+        "person_rows": len(transferred_frame.table("person")),
+        "output_columns": list(US_WIC_CLAIM_OUTPUT_COLUMNS),
+        "category_rates_source": WIC_CLAIM_FNS_SOURCE_URL,
+    }
+
     pool_options: dict[str, Any] = {"acs_share": acs_share}
     if puma_ladder is not None:
         pool_options.update(
@@ -219,12 +241,14 @@ def build_optional_acs_multispine(
         )
     pooled = with_optional_acs_spine(base, transferred_frame, **pool_options)
     del transferred_frame
+    require_complete_us_wic_claim_input(pooled.table("person"))
 
     provenance = {
         "enabled": True,
         "acs_share": _json_ready(acs_share),
         "loader": loader_provenance,
         "native_inputs": native_provenance,
+        "wic_claim": wic_provenance,
         "imputed_inputs": imputed_provenance,
         "deferred_inputs": deferred_provenance,
         "adult_care_recipient_gate": _json_ready(adult_care_gate)
@@ -283,6 +307,11 @@ def build_optional_acs_multispine(
     )
 
 
+def _require_complete_pre_acs_donor_wic_input(base: Frame) -> None:
+    """Validate donor participation before loading or appending ACS records."""
+    require_complete_us_wic_claim_input(base.table("person"))
+
+
 def _require_recipient_adult_care_structure(
     frame: Frame,
 ) -> dict[str, Any] | None:

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

@@ -29,6 +29,7 @@
 import hashlib
 from collections.abc import Mapping
 from importlib.resources import files
+from pathlib import Path
 
 import numpy as np
 import pandas as pd
@@ -63,6 +64,9 @@
     "WIC_CLAIM_ARCHIVED_RANDOMNESS_URL",
     "WIC_CLAIM_FNS_SOURCE_URL",
     "derive_us_wic_claim_from_manifest",
+    "require_complete_us_wic_claim_input",
+    "require_complete_us_wic_claim_h5",
+    "with_acs_wic_claim_input",
     "us_wic_claim_signal_gate",
     "us_wic_claim_stage_spec",
     "us_wic_claim_summary",
@@ -88,9 +92,7 @@
 )
 
 US_WIC_CLAIM_STAGE_NAME = "wic_claim_input"
-US_WIC_CLAIM_OUTPUT_COLUMNS: tuple[str, ...] = (
-    "takes_up_wic_if_eligible",
-)
+US_WIC_CLAIM_OUTPUT_COLUMNS: tuple[str, ...] = ("takes_up_wic_if_eligible",)
 US_WIC_CLAIM_NONCONSTANT_PERSON_COLUMNS = US_WIC_CLAIM_OUTPUT_COLUMNS
 
 # These are persisted PolicyEngine-facing columns already carried or derived
@@ -527,6 +529,63 @@ def with_us_wic_claim_input(
     )
 
 
+def require_complete_us_wic_claim_input(person: pd.DataFrame) -> None:
+    """Reject absent, nullable, or non-boolean participation, without filling."""
+    if _OUTPUT not in person:
+        raise ValueError(f"WIC participation requires person column {_OUTPUT!r}.")
+    if person[_OUTPUT].isna().any():
+        raise ValueError(f"WIC participation {_OUTPUT!r} contains missing values.")
+    if not pd.api.types.is_bool_dtype(person[_OUTPUT].dtype):
+        raise ValueError(f"WIC participation {_OUTPUT!r} must have boolean dtype.")
+
+
+def with_acs_wic_claim_input(frame: Frame, *, seed: int, time_period: int) -> Frame:
+    """Generate ACS decisions after demographic completion, before assembly.
+
+    Require the actual source identity instead of accepting the generic
+    generator's person-id fallback. No donor records may be passed here.
+    Categories, rates, and hash draws remain owned by the existing generator.
+    """
+    person = frame.table("person")
+    missing = [column for column in _SOURCE_IDENTITY_COLUMNS if column not in person]
+    if missing:
+        raise ValueError(
+            f"ACS WIC generation requires source identity columns: {missing}."
+        )
+    identity = person.loc[:, list(_SOURCE_IDENTITY_COLUMNS)]
+    if identity.isna().any(axis=None):
+        raise ValueError("ACS WIC generation requires complete source identity.")
+    result = with_us_wic_claim_input(frame, seed=seed, time_period=time_period)
+    require_complete_us_wic_claim_input(result.table("person"))
+    return result
+
+
+def require_complete_us_wic_claim_h5(path: Path) -> int:
+    """Check persisted participation in bounded row batches; return row count.
+
+    Table-format H5 reads only the participation column. Fixed-format H5
+    requires all columns for each row batch; pandas cannot project columns
+    from that format. Numeric/boolean blocks are sliced, but pandas may
+    unpickle an entire fixed-format object block despite the row bounds.
+    """
+    with pd.HDFStore(path, mode="r") as store:
+        if "/person" not in store.keys():
+            raise ValueError("WIC participation requires a person H5 table.")
+        storer = store.get_storer("person")
+        rows = int(storer.nrows if storer.is_table else storer.shape[0])
+        if rows == 0:
+            raise ValueError("WIC participation cannot certify an empty person table.")
+        for start in range(0, rows, 65_536):
+            if storer.is_table:
+                person = store.select(
+                    "person", start=start, stop=start + 65_536, columns=[_OUTPUT]
+                )
+            else:
+                person = store.select("person", start=start, stop=start + 65_536)
+            require_complete_us_wic_claim_input(person)
+    return rows
+
+
 def _person_weights(frame: Frame) -> np.ndarray:
     weights = np.asarray(frame.resolve_weights("person").values, dtype=np.float64)
     valid = np.isfinite(weights) & (weights >= 0.0)

packages/microcosm-build/tests/engine_free/shared/test_spec_engine_loader.py

@@ -12,7 +12,7 @@ def test_semantic_hash_has_golden_vector_and_surface_separation(tmp_path) -> Non
     # Pin the domain separator, normalization rules, schema-set receipt, and
     # exact normative projection as one reviewable golden vector.
     assert first.spec_sha256 == (
-        "69de1ffc232c3a4da09925658c9d105af55bdaae90852824ac6283306932aab9"
+        "e66fe425edb9b814a9133801fadf2ebf5dade584bd250e2124f6a51a57704bf8"
     )
 
     second_root = _rich_minimal(tmp_path / "xy", note="second", store="local:b")

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

@@ -43,8 +43,16 @@ def _frame(
     person["hours_worked_last_week"] = person["A_HRS1"]
     person["age"] = np.resize([40, 30, 35, 32], n)
     person["is_female"] = np.resize([False, True], n)
+    person["is_pregnant"] = False
+    person["own_children_in_household"] = 0
+    person["source_year"] = 2024
+    person["source_household_id"] = ids
+    person["source_person_id"] = ids
     tables["household"]["state_fips"] = 6
     if role is not None:
+        # Already-built donors have participation decisions. Raw ACS fixtures
+        # deliberately omit them so the real post-transfer generator is used.
+        person["takes_up_wic_if_eligible"] = np.resize([False, True], n)
         for entity, table in tables.items():
             table[support_channel_column(entity)] = role
             table[support_source_id_column(entity)] = ids
@@ -417,6 +425,15 @@ def fit(self, frame, predictors, targets, *, weights):
     acs = result.frame.person.loc[
         result.frame.person["person_spine"].eq("acs_2024_1yr")
     ]
+    assert acs["takes_up_wic_if_eligible"].dtype == bool
+    assert not acs["takes_up_wic_if_eligible"].isna().any()
+    donors = result.frame.person.loc[
+        ~result.frame.person["person_spine"].eq("acs_2024_1yr")
+    ].set_index("person_id")
+    pd.testing.assert_series_equal(
+        donors["takes_up_wic_if_eligible"].sort_index(),
+        original.person.set_index("person_id")["takes_up_wic_if_eligible"].sort_index(),
+    )
     assert acs["weekly_hours_worked_before_lsr"].tolist() == (
         [40, 0, 19.5 if unresolved else 38, 0] * 2
     )